Stream Ingest

Use the stream ingest API to land JSON rows into any lakehouse table with real, typed columns. It is the general-purpose write path: a single endpoint that can append, upsert, or fully replace a table, autocreate the table if it does not exist, and grow its schema as the incoming data grows. It is the on-prem equivalent of the Definite cloud stream API, so a client repoints by changing only the URL and the token.

POST /api/v1/stream
Authorization: Bearer <token>

Use a normal Definite session token or a def_ API token (see API reference). The endpoint is open to any authenticated user, not admin-only. The token writes with the full permissions of the user behind it.

When to use this vs Event Ingest

Stream and Event Ingest both append JSON to a DuckLake table, but they are shaped for different jobs:

• Stream writes arbitrary real columns. Each top-level key in a row becomes its own typed column, and the endpoint supports merge (upsert on a primary key) and replace (full overwrite) in addition to append. Reach for it when you are syncing records from another system and want them queryable as first-class columns, or when you need upsert/replace semantics.
• Event Ingest is append-only into a fixed six-column product-event schema (event_id, event_time, event_name, distinct_id, properties, ingested_at); everything that is not one of those lands in the properties JSON blob. Reach for it for high-frequency event tracking (page views, clicks, signups) and for browser/webhook writers that use a public evpub_... write key.

In short: Event Ingest for product events into one stable schema; Stream for arbitrary tables with merge/replace.

Request shape

The body is a JSON object with data and config:

{
  "data": { },
  "config": { "table": "schema.table", "mode": "append" }
}

data is either a single JSON object (one row) or an array of objects (a batch). Each top-level key in a row becomes a column. Nested objects and arrays are encoded as JSON strings and stored in a VARCHAR column, so a heterogeneous nested shape will not break type inference.

Config fields

An invalid mode, or merge without a non-empty primary_key, is rejected as a 422 validation error before any write.

Modes

merge and replace run inside a single transaction (BEGIN … COMMIT), so a reader never sees a half-applied batch.

What merge preserves vs. overwrites

merge only touches the columns your payload actually sends, so you can update a subset of a wide row without wiping the rest:

• A column absent from the payload keeps its current value on an existing row. This is the key difference from a full-row replace: derived or system-managed columns (for example a first_name_lower you never send) are no longer nulled out by a merge that omits them.
• A column present with an explicit null ("col": null) does set that column to NULL. "Absent" and "explicitly null" are different: only absent columns are preserved.
• For a new key (an insert), columns the payload omits land as NULL — the row did not exist, so there is nothing to preserve.
• The column set is batch-level: the union of keys across every row in the payload. If one row in a batch carries a key that another row omits, the omitting row gets NULL for that cell (standard for a JSON batch).
• A payload that carries only the primary-key column(s) has nothing to update, so existing rows are left untouched and only brand-new keys are inserted.

Merge validation

merge validates the whole batch before writing and rejects it with a 400 if:

• any row is missing a value for one of the primary_key columns (or the value is null), or
• two rows in the same batch collide on the same key (the tuple of key columns for a composite key).

Deduplicate within a batch before sending, and make sure every row carries a non-null key.

Table autocreate and schema evolution

With create: true (the default):

• The schema and table are created if they do not exist (CREATE SCHEMA IF NOT EXISTS + CREATE TABLE IF NOT EXISTS). Column types are inferred from the data: booleans become BOOLEAN, integers BIGINT, floats DOUBLE, decimals DECIMAL, timestamps/dates/times their matching types, and everything else (strings, and the JSON-stringified nested values) VARCHAR.
• Schema evolution is additive. If a later batch carries a column the table does not have, the column is added with ALTER TABLE … ADD COLUMN; rows that predate it stay NULL. Columns are never dropped or retyped: a column absent from a later batch simply stays NULL for the new rows, and an incoming type that differs from the existing column is logged and left to the insert's implicit cast (so a genuinely incompatible change still fails loudly).

With create: false, no schema or table is created. If the target table does not exist the insert fails and the request returns a 400.

Identifiers, batch caps, and errors

• Identifiers. schema and table must each match ^[A-Za-z_][A-Za-z0-9_]{0,62}$ (a letter or underscore, then up to 62 letters, digits, or underscores). Uppercase is allowed. Column names follow the same rule. Anything outside it (a dash, a dot inside a segment, a quote) is rejected with a 400, and identifiers are double-quoted before they reach SQL.
• Batch caps. A batch is capped at 10,000 rows and 8 MiB of request body. Exceeding either returns a 413. An empty data (missing, null, or []) returns a 400.

The endpoint serializes writes through the lakehouse writer and retries on a DuckLake commit/write conflict, so concurrent /stream calls to the same table are safe.

Response

A successful write returns 200 with:

{
  "success": true,
  "request_id": "req_0a1b2c3d4e5f",
  "stream_id": "st_9f8e7d6c5b4a",
  "table": "crm.contacts",
  "successful_rows": 3,
  "rejected_rows": 0,
  "snapshot_id": null
}

Examples

The examples use $DEFINITE_API_URL (your deployment, e.g. https://definite.example.com) and $DEFINITE_TOKEN (a session or def_ token).

Append

Insert rows into crm.contacts, creating the table on the first call:

curl -sS "$DEFINITE_API_URL/api/v1/stream" \
  -H "Authorization: Bearer $DEFINITE_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "data": [
      {"id": 1, "name": "Ada Lovelace", "email": "ada@example.com"},
      {"id": 2, "name": "Alan Turing", "email": "alan@example.com"}
    ],
    "config": {"table": "crm.contacts"}
  }'

Merge (upsert)

Upsert on id: an existing contact has the sent columns updated, a new one is inserted, and contacts not in the batch are left untouched. Any column the payload omits (say a created_at or a derived name_lower) keeps its current value on the updated rows — merge only writes the columns you send:

curl -sS "$DEFINITE_API_URL/api/v1/stream" \
  -H "Authorization: Bearer $DEFINITE_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "data": [
      {"id": 2, "name": "Alan Turing", "email": "alan.turing@example.com"},
      {"id": 3, "name": "Grace Hopper", "email": "grace@example.com"}
    ],
    "config": {"table": "crm.contacts", "mode": "merge", "primary_key": ["id"]}
  }'

For a composite key, pass more than one column, e.g. "primary_key": ["org_id", "id"].

Replace

Overwrite the table with a fresh snapshot (every existing row is deleted, then the incoming rows are inserted):

curl -sS "$DEFINITE_API_URL/api/v1/stream" \
  -H "Authorization: Bearer $DEFINITE_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "data": [
      {"id": 1, "name": "Ada Lovelace", "email": "ada@example.com"},
      {"id": 2, "name": "Alan Turing", "email": "alan@example.com"}
    ],
    "config": {"table": "crm.contacts", "mode": "replace"}
  }'

See also

• Event Ingest: append-only product events into a fixed schema
• API reference: auth, the full endpoint surface, and SQL queries
• Permissions: table-level data access