Ingestion pipelines

Load JSON rows and staged files through a saved, retry-safe pipeline.

Availability

Ingestion pipelines are the Phase 1 API contract for customer-managed data loading and are available in every workspace. Current Engine grants determine who can create, inspect, or execute a pipeline; there is no separate workspace rollout allowlist.

How the model fits together

A pipeline saves a target table plus write, schema, and error policy. An upload is a temporary reservation for exact staged bytes. A run is one durable validation or commit attempt. A pipeline is configuration, not a credential: every request is authorized from the caller's current Engine grants.

Use an admin-granted human session to create and configure a pipeline bound to a stable Table node ID.
On the workspace's regional data plane, create a principal-bound API-key authenticator with a table-scoped Editor ceiling for routine append or upsert execution, then exchange the API key there for a short-lived machine JWT.
Submit inline JSON rows, or stage a JSON, CSV, TSV, XLSX, or Parquet file.
Poll the returned run until it reaches a terminal state.

Permissions

Human users, groups, Agents, and the public Everyoneprincipal all use the same Engine grant resolver. API keys and MCP OAuth authorizations authenticate principals; neither is itself a principal. Workspace ownership, an authenticator secret, a URL, a pipeline ID, and an upload ID do not grant data access.

Create and manage API-key authenticators on the owning regional dataPlaneUrl through /v1/workspaces/{workspacePublicId}/credentials. These management calls use the human's Supabase access JWT; the workspace-scoped URL selects the request context. They are not served from an app-domain organization route. Each authenticator names its effective principalId. Node rules use one stable regionalresourceId, with an allow role or a roleless explicit deny. The Engine derives the node kind and current Namespace ancestry. An applicable deny is absolute. A key's effective access is the bound principal's live authority intersected with its node rules and capability ceiling on every request. The creating human is audit provenance and contributes no authority to an Agent-bound key.

Operation	Required effective grant
List a pipeline or inspect its runs	`viewer` on the target or an ancestor
Validate rows against an existing target	`viewer` on the target or an ancestor
Append or upsert into an existing target	`editor` on the target or an ancestor
Replace every row in an existing target	`admin` on the target or an ancestor
Create or change pipeline configuration	`admin` on the target, parent Namespace, or an ancestor
Add nullable columns in additive schema mode	`admin` on the target or an ancestor

The long-lived API key belongs only in a server-side secret store. Exchange it through regional POST /v1/workspaces/{workspacePublicId}/token/exchange; never put the API key in a URL, browser bundle, source file, or staged object metadata. This is a machine-credential exchange, not an exchange of the human's Supabase JWT. The key and exchanged JWT identify the bound principal and authenticator but contain no grant state; the Engine's current grants intersected with the authenticator's ceilings are the authority. See the authentication guide for the exchange and refresh flow.

Create a pipeline

Resolve the destination table first, then bind the pipeline to its stable Table node tableId, such as rw-us-east1-node-0195a3c47d3e7ab18c4f123456789abe. Renaming that table does not retarget the pipeline. Deleting and recreating a same-named table does not silently rebind it. A Table name or Namespace path is not a substitute for this ID.

ADMIN_JWT may be a human's Supabase access JWT or the exchanged JWT from a workspace-scoped Admin API key. Both use this same endpoint. Use a table-scoped Editor API key when an integration only needs to execute append or upsert runs.

Create the target with standard SQL

An authenticated human Admin can create a target through the ordinary POST /v1/workspaces/{workspacePublicId}/query API. Relation paths are absolute from the workspace Root and may have any Namespace depth. IF NOT EXISTS makes an identical bootstrap request retry-safe; it does not reconcile or replace the schema of an existing table. Use the catalog metadata API to obtain the created Table's stable node ID before creating the pipeline.

bash

WORKSPACE_ID="rw-us-east1-ws-0195a3c47d3e7ab18c4f123456789abc"
DATA_PLANE_URL="https://rw-us-east1.api.loamdata.com"

curl -X POST "$DATA_PLANE_URL/v1/workspaces/$WORKSPACE_ID/query" \
  -H "Authorization: Bearer $ADMIN_JWT" \
  -H "Content-Type: application/json" \
  -d '{
    "sql": "CREATE TABLE IF NOT EXISTS training.private.workouts (workout_id UUID NOT NULL, started_at DateTime NOT NULL, duration_seconds UInt64 NOT NULL, distance_metres Float64) ENGINE = Loam ORDER BY (workout_id)",
    "format": "json"
  }'

Bind the existing target

bash

curl -X POST "$DATA_PLANE_URL/v1/workspaces/$WORKSPACE_ID/ingestion-pipelines" \
  -H "Authorization: Bearer $ADMIN_JWT" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "application_events",
    "target": {
      "mode": "existingTable",
      "tableId": "rw-us-east1-node-0195a3c47d3e7ab18c4f123456789abe"
    },
    "input": {
      "formats": ["json", "csv", "tsv", "parquet", "xlsx"],
      "columnMappings": {}
    },
    "write": {
      "mode": "upsert",
      "keyColumns": ["source_event_id"]
    },
    "schema": {
      "mode": "strict",
      "coercions": []
    },
    "errors": {
      "mode": "failRun",
      "quarantine": false
    }
  }'

Optional create-if-missing target

Where the workspace rollout enables it, an Admin may instead create a pipeline with target.mode: "createIfMissing". Phase 1 requires an explicit initialSchema; schema inference is disabled. The namespaceId must be the stable node ID of an ordinary existing Namespace. The server derives its current path and ancestry from that node; callers do not submit either. This capability gate does not bypass Admin authorization, and an Editor integration cannot use it to create a table.

json

{
  "name": "application_events",
  "target": {
    "mode": "createIfMissing",
    "namespaceId": "rw-us-east1-node-0195a3c47d3e7ab18c4f123456789abd",
    "tableName": "application_events",
    "initialSchema": [
      { "name": "source_event_id", "type": "String", "nullable": false },
      { "name": "occurred_at", "type": "DateTime", "nullable": false },
      { "name": "value", "type": "Int64", "nullable": true }
    ]
  },
  "input": { "formats": ["json"], "columnMappings": {} },
  "write": { "mode": "upsert", "keyColumns": ["source_event_id"] },
  "schema": { "mode": "strict", "coercions": [] },
  "errors": { "mode": "failRun", "quarantine": false }
}

The response envelope's data object contains the opaque pipeline ID and immutable configuration version; metacontains request and trace IDs. Keep the ID as application configuration. It conveys no authority without a valid JWT and a current grant.

Create and exchange a regional API-key authenticator

Configure the workspace's structured public ID and exact regional API origin together. The region prefix selects the plane; the human Supabase JWT and current local Admin grant authorize authenticator creation. Bind the key to exactly one human or Agent principal. The secret is returned once.

bash

WORKSPACE_ID="rw-us-east1-ws-0195a3c47d3e7ab18c4f123456789abc"
DATA_PLANE_URL="https://rw-us-east1.api.loamdata.com"

curl -X POST \
  "$DATA_PLANE_URL/v1/workspaces/$WORKSPACE_ID/credentials" \
  -H "Authorization: Bearer $SUPABASE_ACCESS_JWT" \
  -H "Content-Type: application/json" \
  -d '{
    "kind": "api_key",
    "principalId": "0195a3c4-7d3e-7ab1-8c4f-123456789abc",
    "name": "Ingestion writer",
    "nodeRules": [
      {
        "effect": "allow",
        "resourceId": "rw-us-east1-node-0195a3c47d3e7ab18c4f123456789abe",
        "role": "editor"
      }
    ]
  }' > authenticator.json

# Copy .secret to a server-side secret store, then discard authenticator.json.
API_KEY="$(jq -r .secret authenticator.json)"

curl -X POST "$DATA_PLANE_URL/v1/workspaces/$WORKSPACE_ID/token/exchange" \
  -H "Content-Type: application/json" \
  -d "$(jq -n --arg token "$API_KEY" '{token: $token}')" > access.json

JWT="$(jq -r .jwt access.json)"
test "$(jq -r .workspaceId access.json)" = "$WORKSPACE_ID"
test "$(jq -r .dataPlaneUrl access.json)" = "$DATA_PLANE_URL"

Keep using the configured regional DATA_PLANE_URL; the exchange response repeats that exact dataPlaneUrl and the workspace public ID so the integration can fail closed on a mismatch. Listing, inspecting, updating, and revoking use these workspace authenticator paths on that host:

text

GET    /v1/workspaces/{workspacePublicId}/credentials
POST   /v1/workspaces/{workspacePublicId}/credentials
GET    /v1/workspaces/{workspacePublicId}/credentials/{authenticatorId}
PATCH  /v1/workspaces/{workspacePublicId}/credentials/{authenticatorId}
DELETE /v1/workspaces/{workspacePublicId}/credentials/{authenticatorId}

Submit inline JSON rows

Inline submission accepts a JSON object containing rows. Include a stable sourceBatchId when the producer has a source load or event ID. The Phase 1 inline request limit is 10 MiB; use staged file upload above that limit.

bash

SOURCE_BATCH_ID="billing-export-2026-07-25T02:00:00Z"
IDEMPOTENCY_KEY="billing:${SOURCE_BATCH_ID}"

curl -X POST "$DATA_PLANE_URL/v1/workspaces/$WORKSPACE_ID/ingestion-pipelines/$PIPELINE_ID/rows" \
  -H "Authorization: Bearer $JWT" \
  -H "Content-Type: application/json" \
  -H "Idempotency-Key: $IDEMPOTENCY_KEY" \
  -d '{
    "sourceBatchId": "billing-export-2026-07-25T02:00:00Z",
    "rows": [
      {
        "source_event_id": "evt_123",
        "occurred_at": "2026-07-25T02:00:00Z",
        "value": 42
      }
    ]
  }'

A successful submission returns 202 Accepted with a runId, initial status, and statusUrl. Acceptance does not mean the table has committed yet.

TypeScript with same-key retry

typescript

type RunRef = {
  runId: string;
  status: string;
  statusUrl: string;
};

type Envelope<T> = {
  data: T;
  meta: { requestId: string; traceId: string };
};

async function submitRows(args: {
  dataPlaneUrl: string;
  workspacePublicId: string;
  jwt: string;
  pipelineId: string;
  sourceBatchId: string;
  rows: Record<string, unknown>[];
}): Promise<RunRef> {
  const idempotencyKey = `producer:${args.sourceBatchId}`;
  const body = JSON.stringify({
    sourceBatchId: args.sourceBatchId,
    rows: args.rows,
  });

  // Reuse this exact body and key after timeouts or retryable 5xx responses.
  for (let attempt = 0; attempt < 4; attempt += 1) {
    const response = await fetch(
      `${args.dataPlaneUrl}/v1/workspaces/${args.workspacePublicId}/ingestion-pipelines/${args.pipelineId}/rows`,
      {
        method: "POST",
        headers: {
          Authorization: `Bearer ${args.jwt}`,
          "Content-Type": "application/json",
          "Idempotency-Key": idempotencyKey,
        },
        body,
      },
    );

    if (response.ok) {
      const envelope = (await response.json()) as Envelope<RunRef>;
      return envelope.data;
    }
    if (response.status === 429) {
      const seconds = Number(response.headers.get("retry-after") ?? "1");
      await new Promise((resolve) => setTimeout(resolve, seconds * 1000));
      continue;
    }
    if (response.status >= 500 && attempt < 3) continue;
    throw new Error(`row submission failed: ${response.status} ${await response.text()}`);
  }

  throw new Error("row submission retry budget exhausted");
}

Upload CSV, JSON, or another supported file

File bytes go directly to a short-lived signed object capability. The data does not pass through the control plane. The client receives an opaque uploadId and must not construct or submit an object key. The same create, PUT, commit, and poll flow applies to JSON, CSV, TSV, XLSX, and Parquet. NDJSON/JSONL is not a separate Phase 1 file format; convert it to a JSON array or use the inline rows endpoint.

For a one-off file import without a saved pipeline, target one ordinary Namespace by stable node ID at /v1/workspaces/{workspacePublicId}/nodes/{nodePublicId}/file-imports. File imports create ordinary direct-child Tables in that Namespace.

Example CSV

csv

source_event_id,occurred_at,value
evt_123,2026-07-25T02:00:00Z,42
evt_124,2026-07-25T02:01:00Z,43

Equivalent JSON file

json

[
  {
    "source_event_id": "evt_123",
    "occurred_at": "2026-07-25T02:00:00Z",
    "value": 42
  },
  {
    "source_event_id": "evt_124",
    "occurred_at": "2026-07-25T02:01:00Z",
    "value": 43
  }
]

1. Create the upload

bash

FILE="events.json"
SIZE_BYTES="$(wc -c < "$FILE" | tr -d ' ')"
SHA256="$(sha256sum "$FILE" | cut -d ' ' -f 1)"

curl -X POST "$DATA_PLANE_URL/v1/workspaces/$WORKSPACE_ID/ingestion-pipelines/$PIPELINE_ID/uploads" \
  -H "Authorization: Bearer $JWT" \
  -H "Content-Type: application/json" \
  -d "{
    \"filename\": \"events.json\",
    \"format\": \"json\",
    \"sizeBytes\": $SIZE_BYTES,
    \"mediaType\": \"application/json\",
    \"sha256\": \"$SHA256\",
    \"sourceBatchId\": \"events-2026-07-25\"
  }" > upload.json

UPLOAD_ID="$(jq -r .data.uploadId upload.json)"
UPLOAD_MODE="$(jq -r .data.uploadMode upload.json)"

UPLOAD_URL="$(jq -r .data.uploadUrl upload.json)"
AUTH_HEADER_ARGS=()
case "$UPLOAD_MODE" in
  gateway)
    EXPECTED_PATH="/v1/workspaces/$WORKSPACE_ID/ingestion-pipelines/$PIPELINE_ID/uploads/$UPLOAD_ID/content"
    if [[ "$UPLOAD_URL" != "$EXPECTED_PATH" ]] ||
       [[ "$(jq -r .data.uploadOrigin upload.json)" != "" ]]; then
      echo "Unexpected authenticated upload destination" >&2
      exit 1
    fi
    UPLOAD_URL="${DATA_PLANE_URL%/}$EXPECTED_PATH"
    jq -e --arg size "$SIZE_BYTES" '.data.requiredHeaders | with_entries(.key |= ascii_downcase) | keys == ["content-length", "content-type"] and .["content-length"] == $size and .["content-type"] == "application/json"' upload.json >/dev/null || exit 1
    AUTH_HEADER_ARGS=(-H "Authorization: Bearer $JWT")
    ;;
  singlePut)
    # Legacy provider-direct mode: never send the Loam bearer to storage.
    UPLOAD_ORIGIN="$(jq -r .data.uploadOrigin upload.json)"
    python3 - "$UPLOAD_ORIGIN" "$UPLOAD_URL" <<'PYTHON' || exit 1
import sys
from urllib.parse import urlsplit
origin, target = map(urlsplit, sys.argv[1:])
assert origin.scheme == target.scheme == "https"
assert origin.netloc == target.netloc and origin.netloc
assert origin.username is None and origin.password is None
assert target.username is None and target.password is None
assert not origin.fragment and not target.fragment
assert not origin.query and origin.path == ""
PYTHON
    ;;
  *)
    echo "Use the authenticated multipart flow documented below" >&2
    exit 1
    ;;
esac

# Returned transport headers must never override caller authentication.
jq -e '.data.requiredHeaders // {} | to_entries | all((.key | ascii_downcase) as $key | $key != "authorization" and $key != "cookie" and $key != "host" and (.value | type == "string") and (.value | explode | all(. != 0 and . != 10 and . != 13)))' upload.json >/dev/null || exit 1
mapfile -t REQUIRED_HEADERS < <(
  jq -r '.data.requiredHeaders // {} | to_entries[] | "\(.key): \(.value)"' upload.json
)

2. PUT the exact bytes

A protected saved-pipeline upload returns uploadMode: "gateway" and a canonical relative uploadUrl on the selected regional API. Verify the exact workspace, pipeline, and upload path before joining it to the configured regional API URL. Send the current Supabase human bearer and the exact returned media type and byte length. This single-request gateway accepts at most 64 MiB. Protected saved uploads currently require a human session; unsupported machine credentials fail closed.

A legacy, unprotected target may still return uploadMode: "singlePut" with a signed provider URL. Validate its exact uploadOrigin, send the returned transport headers, and send no Loam bearer or cookies to that provider. The SDK handles both responses. Do not follow redirects for either upload mode.

bash

CURL_HEADER_ARGS=()
for header in "${REQUIRED_HEADERS[@]}"; do
  CURL_HEADER_ARGS+=(-H "$header")
done

curl --fail-with-body -X PUT "$UPLOAD_URL" \
  "${AUTH_HEADER_ARGS[@]}" \
  "${CURL_HEADER_ARGS[@]}" \
  --data-binary "@$FILE"

Larger objects can return uploadMode: "multipart" with an opaque completionId and partSizeBytes. Protected uploads use bounded 16 MiB parts by default. Request the part records from the authenticated upload /parts endpoint. Each part URL must exactly match the same workspace, pipeline, upload, and opaque part ID at /parts/{partId}/content. Join only that verified relative path to the canonical regional API and send the current human bearer with its exact declared media type and part length. PUT only the requested partNumber values, retain each actual server response ETag unchanged, then complete with the original opaque part IDs and completion ID. Provider-direct legacy multipart URLs retain their exact provider-origin validation and receive no Loam bearer:

bash

cat > multipart-complete.json <<'JSON'
{
  "completionId": "01980c54-f7a2-7d31-8eb1-566cf1b44aa2",
  "parts": [
    {
      "partId": "01980c54-f7a2-7d31-8eb1-566cf1b44aa3",
      "partNumber": 1,
      "etag": "\"9b2cf535f27731c974343645a3985328\""
    }
  ]
}
JSON

curl -X POST \
  "$DATA_PLANE_URL/v1/workspaces/$WORKSPACE_ID/ingestion-pipelines/$PIPELINE_ID/uploads/$UPLOAD_ID/complete" \
  -H "Authorization: Bearer $JWT" \
  -H "Content-Type: application/json" \
  --data-binary "@multipart-complete.json"

Then include the completionId in the upload commit request. Multipart bytes count toward the same 256 MiB file and 2 GiB workspace staging limits.

3. Commit the completed upload

bash

curl -X POST \
  "$DATA_PLANE_URL/v1/workspaces/$WORKSPACE_ID/ingestion-pipelines/$PIPELINE_ID/uploads/$UPLOAD_ID/commit" \
  -H "Authorization: Bearer $JWT" \
  -H "Content-Type: application/json" \
  -H "Idempotency-Key: files:events-2026-07-25" \
  -d '{"validateOnly": false}' > run.json

RUN_ID="$(jq -r .data.runId run.json)"

For a completed multipart upload, use {"completionId": "...", "validateOnly": false} as the commit body. A single gateway or single-PUT commit does not include a completion ID.

To check parsing and schema compatibility without changing the table, set validateOnly to true. Validation-only requests still need an idempotency key and create a durable run.

Poll a run

typescript

type IngestionRun = {
  runId: string;
  pipelineVersionId: string;
  status:
    | "accepted"
    | "validating"
    | "staged"
    | "committing"
    | "succeeded"
    | "partial"
    | "failed"
    | "cancelled";
  counters: {
    receivedRows: number;
    insertedRows: number;
    updatedRows: number;
    rejectedRows: number;
    receivedBytes: number;
    processedBytes: number;
  };
  commitId: string | null;
  cleanupState:
    | "notScheduled"
    | "pending"
    | "leased"
    | "succeeded"
    | "failed";
  terminalError: { code: string; message: string } | null;
};

const terminal = new Set(["succeeded", "partial", "failed", "cancelled"]);

async function waitForRun(
  dataPlaneUrl: string,
  workspacePublicId: string,
  jwt: string,
  runId: string,
): Promise<IngestionRun> {
  for (;;) {
    const response = await fetch(
      `${dataPlaneUrl}/v1/workspaces/${workspacePublicId}/ingestion-runs/${runId}`,
      {
        headers: {
          Authorization: `Bearer ${jwt}`,
        },
      },
    );
    if (!response.ok) {
      throw new Error(`run lookup failed: ${response.status}`);
    }
    const envelope = (await response.json()) as Envelope<IngestionRun>;
    const run = envelope.data;
    if (terminal.has(run.status)) return run;
    await new Promise((resolve) => setTimeout(resolve, 1_000));
  }
}

On failure, inspect GET /v1/workspaces/{workspacePublicId}/ingestion-runs/{runId}/errors. Error reports contain bounded row indexes, column names, codes, and safe samples. Ordinary logs do not contain source rows, secrets, or signed URLs.

Retry rules

Retry a timed-out submission with the same operation, exact content, and Idempotency-Key. Loam returns the original run.
Reusing a key with different content returns IDEMPOTENCY_CONFLICT.
A stable sourceBatchId prevents the same source load from being applied with a different request key. Changed content returns SOURCE_BATCH_CONFLICT.
Upsert keyColumns provide row-level business deduplication. They are not a replacement for either retry key.

Ownership and shared MCP access

The regional host is selected from the workspace public ID. A human Supabase JWT supplies identity, while the workspace-scoped URL supplies request context; a regional machine JWT is already bound to its exact workspace and audience. The pipeline target must belong to that same workspace and region. Ingestion never changes ownership or visibility. Private application data remains private unless an authorized human Admin creates an Everyone Viewer grant on the Table or an ancestor Namespace.

MCP clients connect to the single configured resource https://www.loamdata.com/mcp. During OAuth, the human must explicitly choose a workspace-qualified My Account identity or an eligible Agent. The selected principal's live Engine grants are the sole authority; the related human is retained for audit and never contributes grants. MCP remains read-only, so submit ingestion writes through a principal-bound API-key authenticator.

Last updated 2026-08-23