> ## Documentation Index
> Fetch the complete documentation index at: https://docs.e6data.com/llms.txt
> Use this file to discover all available pages before exploring further.

# HTTP API

> Ingest Engine exposes an HTTP API for managing pipelines, connections, and jobs programmatically.

## Base URL

```
http://<controller-host>:8000/api/v1
```

All endpoints below are relative to this base URL.

## Authentication

Include the API token in the `Authorization` header:

```bash theme={"theme":{"light":"github-light","dark":"github-dark"}}
curl -H "Authorization: Bearer <token>" http://localhost:8000/api/v1/pipelines
```

## Response Format

All responses are JSON. Errors return:

```json theme={"theme":{"light":"github-light","dark":"github-dark"}}
{
  "error": "descriptive error message"
}
```

List endpoints use cursor-based pagination:

```json theme={"theme":{"light":"github-light","dark":"github-dark"}}
{
  "data": [...],
  "has_more": true
}
```

Pass `?starting_after=<last_id>&limit=10` to paginate.

***

## Pipelines

### Create Pipeline

```
POST /pipelines
```

Creates a pipeline and starts it immediately.

**Request:**

```json theme={"theme":{"light":"github-light","dark":"github-dark"}}
{
  "name": "my-pipeline",
  "query": "INSERT INTO sink SELECT * FROM source",
  "parallelism": 4,
  "checkpoint_interval_micros": 30000000
}
```

| Field | Type | Required | Description |
| - | - | - | - |
| `name` | string | yes | Pipeline name |
| `query` | string | yes | SQL query defining the pipeline |
| `parallelism` | integer | yes | Number of parallel subtasks |
| `checkpoint_interval_micros` | integer | no | Checkpoint interval in microseconds (default: 30s) |

**Response:** `Pipeline` object

***

### Create Preview Pipeline

```
POST /pipelines/preview
```

Creates a short-lived pipeline (60s TTL) for testing queries interactively.

**Request:**

```json theme={"theme":{"light":"github-light","dark":"github-dark"}}
{
  "query": "SELECT * FROM source WHERE amount > 100",
  "enable_sinks": false
}
```

| Field | Type | Required | Description |
| - | - | - | - |
| `query` | string | yes | SQL query to preview |
| `enable_sinks` | boolean | no | Whether to write to real sinks (default: false) |

**Response:** `Pipeline` object

***

### List Pipelines

```
GET /pipelines
```

**Query Parameters:**

| Parameter | Type | Description |
| - | - | - |
| `starting_after` | string | Cursor for pagination |
| `limit` | integer | Items per page (default: 10) |

**Response:**

```json theme={"theme":{"light":"github-light","dark":"github-dark"}}
{
  "data": [Pipeline, ...],
  "has_more": false
}
```

***

### Get Pipeline

```
GET /pipelines/:id
```

**Response:** `Pipeline` object

```json theme={"theme":{"light":"github-light","dark":"github-dark"}}
{
  "id": "abc123",
  "name": "my-pipeline",
  "query": "INSERT INTO sink SELECT * FROM source",
  "parallelism": 4,
  "checkpoint_interval_micros": 30000000,
  "stop": "none",
  "created_at": 1700000000000000,
  "graph": {
    "nodes": [...],
    "edges": [...]
  },
  "action": "stop",
  "action_text": "Stop",
  "action_in_progress": false,
  "preview": false
}
```

***

### Update Pipeline

```
PATCH /pipelines/:id
```

Updates pipeline configuration. Only provided fields are changed.

**Request:**

```json theme={"theme":{"light":"github-light","dark":"github-dark"}}
{
  "parallelism": 8,
  "checkpoint_interval_micros": 60000000,
  "stop": "checkpoint"
}
```

| Field | Type | Description |
| - | - | - |
| `parallelism` | integer | New parallelism (triggers rescaling) |
| `checkpoint_interval_micros` | integer | New checkpoint interval |
| `stop` | string | Stop mode: `"none"`, `"graceful"`, `"immediate"`, `"checkpoint"`, `"force"` |

**Response:** `Pipeline` object

***

### Restart Pipeline

```
POST /pipelines/:id/restart
```

Restarts a stopped or failed pipeline.

**Request:**

```json theme={"theme":{"light":"github-light","dark":"github-dark"}}
{
  "force": false
}
```

| Field | Type | Description |
| - | - | - |
| `force` | boolean | Skip taking a final checkpoint before restart (default: false) |

**Response:** `Pipeline` object

***

### Delete Pipeline

```
DELETE /pipelines/:id
```

Deletes a pipeline. The pipeline must be in a terminal state (Stopped, Finished, or Failed).

**Response:** `200 OK`

***

### Validate Query

```
POST /pipelines/validate_query
```

Validates a SQL query without creating a pipeline. Returns the execution graph or validation errors.

**Request:**

```json theme={"theme":{"light":"github-light","dark":"github-dark"}}
{
  "query": "INSERT INTO sink SELECT * FROM source"
}
```

**Response:**

```json theme={"theme":{"light":"github-light","dark":"github-dark"}}
{
  "graph": {
    "nodes": [...],
    "edges": [...]
  },
  "errors": []
}
```

***

## Jobs

A job represents a single execution of a pipeline. Each time a pipeline starts or restarts, a new job is created.

### List Jobs

```
GET /jobs
```

Returns all jobs across all pipelines.

```
GET /pipelines/:pipeline_id/jobs
```

Returns jobs for a specific pipeline.

**Response:**

```json theme={"theme":{"light":"github-light","dark":"github-dark"}}
{
  "data": [
    {
      "id": 1,
      "running_desired": true,
      "state": "Running",
      "run_id": 3,
      "start_time": 1700000000000000,
      "finish_time": null,
      "tasks": 8,
      "failure_message": null,
      "created_at": 1700000000000000
    }
  ]
}
```

***

### Get Job Errors

```
GET /pipelines/:pipeline_id/jobs/:job_id/errors
```

Returns error and warning messages for a job.

**Query Parameters:**

| Parameter | Type | Description |
| - | - | - |
| `starting_after` | string | Cursor for pagination |
| `limit` | integer | Items per page |

**Response:**

```json theme={"theme":{"light":"github-light","dark":"github-dark"}}
{
  "data": [
    {
      "id": "msg-123",
      "created_at": 1700000000000000,
      "operator_id": "operator_2",
      "task_index": 0,
      "level": "Error",
      "message": "Connection refused",
      "details": "..."
    }
  ],
  "has_more": false
}
```

***

### Get Job Checkpoints

```
GET /pipelines/:pipeline_id/jobs/:job_id/checkpoints
```

Returns all checkpoints for a job.

**Response:**

```json theme={"theme":{"light":"github-light","dark":"github-dark"}}
{
  "data": [
    {
      "epoch": 5,
      "backend": "s3",
      "start_time": 1700000150000000,
      "finish_time": 1700000151000000
    }
  ]
}
```

***

### Get Checkpoint Details

```
GET /pipelines/:pipeline_id/jobs/:job_id/checkpoints/:epoch/operator_checkpoint_groups
```

Returns per-operator checkpoint details for a specific epoch.

**Response:**

```json theme={"theme":{"light":"github-light","dark":"github-dark"}}
{
  "data": [
    {
      "operator_id": "operator_0",
      "bytes": 4096,
      "finish_time": 1700000151000000,
      "subtasks": [...]
    }
  ]
}
```

***

### Get Job Metrics

```
GET /pipelines/:pipeline_id/jobs/:job_id/operator_metric_groups
```

Returns per-operator metrics for a running job.

***

### Get Job Output (SSE)

```
GET /pipelines/:pipeline_id/jobs/:job_id/output
```

Streams pipeline output in real-time via Server-Sent Events. Only available for pipelines with a preview sink.

**Response:** `text/event-stream`

Each event contains:

```json theme={"theme":{"light":"github-light","dark":"github-dark"}}
{
  "column_names": ["id", "name", "amount"],
  "data": [["1", "Alice", "100"], ["2", "Bob", "200"]]
}
```

***

## Connection Profiles

A connection profile stores reusable connection credentials (e.g., Kafka bootstrap servers, authentication).

### Create Connection Profile

```
POST /connection_profiles
```

**Request:**

```json theme={"theme":{"light":"github-light","dark":"github-dark"}}
{
  "name": "production-kafka",
  "connector": "kafka",
  "config": {
    "bootstrap_servers": "kafka-1:9092,kafka-2:9092",
    "authentication": {
      "type": "sasl_plain",
      "username": "user",
      "password": "pass"
    }
  }
}
```

**Response:** `ConnectionProfile` object

Typed endpoints are also available for specific connectors:

```
POST /connectors/kafka/profiles
POST /connectors/confluent/profiles
```

***

### List Connection Profiles

```
GET /connection_profiles
```

**Response:**

```json theme={"theme":{"light":"github-light","dark":"github-dark"}}
{
  "data": [
    {
      "id": "prof-123",
      "name": "production-kafka",
      "connector": "kafka",
      "config": {...},
      "description": "Kafka cluster"
    }
  ]
}
```

***

### Get Connection Profile

```
GET /connection_profiles/:id
```

***

### Delete Connection Profile

```
DELETE /connection_profiles/:id
```

Fails if the profile is referenced by any connection tables.

***

### Test Connection Profile

```
POST /connection_profiles/test
```

Tests connectivity with the provided configuration.

**Request:** Same as create.

**Response:**

```json theme={"theme":{"light":"github-light","dark":"github-dark"}}
{
  "message": "Successfully connected",
  "done": true
}
```

***

### Autocomplete

```
GET /connection_profiles/:id/autocomplete
```

Returns available values for the connection (e.g., Kafka topic names).

**Response:**

```json theme={"theme":{"light":"github-light","dark":"github-dark"}}
{
  "values": {
    "topics": ["orders", "users", "events"]
  }
}
```

***

## Connection Tables

A connection table defines a specific source or sink - binding a connector, connection profile, schema, and table-specific configuration together.

### Create Connection Table

```
POST /connection_tables
```

**Request:**

```json theme={"theme":{"light":"github-light","dark":"github-dark"}}
{
  "name": "orders",
  "connector": "kafka",
  "connection_profile_id": "prof-123",
  "config": {
    "topic": "orders",
    "read_mode": "read_uncommitted"
  },
  "schema": {
    "format": { "json": {} },
    "fields": [
      { "field_name": "id", "field_type": { "primitive": "Int64" } },
      { "field_name": "amount", "field_type": { "primitive": "Float64" } }
    ]
  }
}
```

Typed endpoints are also available:

```
POST /connectors/kafka/tables
POST /connectors/confluent/tables
POST /connectors/cdc/tables
POST /connectors/iceberg/tables
POST /connectors/delta/tables
POST /connectors/filesystem/tables
POST /connectors/kinesis/tables
POST /connectors/http_source/tables
POST /connectors/stdout/tables
```

**Response:** `ConnectionTable` object

```json theme={"theme":{"light":"github-light","dark":"github-dark"}}
{
  "id": 1,
  "pub_id": "tbl-456",
  "name": "orders",
  "created_at": 1700000000000000,
  "connector": "kafka",
  "table_type": "source",
  "config": {...},
  "schema": {...},
  "consumers": 1
}
```

***

### List Connection Tables

```
GET /connection_tables
```

**Query Parameters:**

| Parameter | Type | Description |
| - | - | - |
| `starting_after` | string | Cursor for pagination |
| `limit` | integer | Items per page |

***

### Get Connection Table

```
GET /connection_tables/:id
```

***

### Delete Connection Table

```
DELETE /connection_tables/:id
```

***

### Test Connection Table

```
POST /connection_tables/test
```

Tests reading from a source with the provided configuration. Returns results as a Server-Sent Events stream.

***

### Validate Schema

```
POST /connection_tables/schemas/test
```

Validates a schema definition.

***

## Connectors

### List Available Connectors

```
GET /connectors
```

Returns metadata about all available connectors.

**Response:**

```json theme={"theme":{"light":"github-light","dark":"github-dark"}}
{
  "data": [
    {
      "name": "kafka",
      "description": "Apache Kafka source and sink",
      "connection_config": {...},
      "table_config": {...}
    }
  ]
}
```

***

## Health

### Ping

```
GET /ping
```

Returns `"Pong"` - used for health checks. No authentication required.


This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.