The consumer loop, and where a crash lands

Declare a consumer and the node runs it: a topic of this store read into a table exactly once, or a Kafka topic read in at least once — what to read, where it lands, under which group, and the guarantee, stated where you configure it.

Ingestion is declared, not scripted. One statement says what to read, where the records land and under which consumer group; the node runs it, starts it again after a restart, and reports what it is doing. There is no separate connector to deploy and no process to remember to start.

There are two sources, and they differ in the one thing that matters most:

	reads	guarantee	runs in
`DEFINE TOPIC CONSUMER`	a topic of this store	exactly once into the store	every build, the published image included
`DEFINE KAFKA CONSUMER`	a Kafka topic	at least once, made safe by the record identity	a build with `--features kafka`

Reading a topic into a table

The topic, a group that reads it, and the table its messages land in:

DEFINE TOPIC order_events;
DEFINE TOPIC order_events_dead;
DEFINE TABLE order_rows (total int);
DEFINE GROUP 'into-rows' ON TOPIC order_events ACK DEADLINE 30s IN FLIGHT 64
    DELIVERIES 3 DEAD LETTER TO order_events_dead;

DEFINE TOPIC CONSUMER events_in
    FROM order_events
    GROUP 'into-rows'
    INTO order_rows
    IDENTITY order_id
    MAP amount AS total
    ON FAILURE quarantine
    PARALLELISM 2;

From then on, a message appended to order_events —

CREATE order_events:'o-1001' = { order_id: 1001, amount: 40 };

— becomes the record order_rows:1001 holding { total: 40 }, within a second, without anybody reading the topic by hand.

Messages published before the consumer was declared are not skipped: a new group starts at the topic's oldest message, so the consumer takes the backlog first, in order, and then keeps up.

The guarantee: exactly once into this store

The node reads the topic as a member of the group, writes the records, and acknowledges the messages in one transaction. They commit together or not at all:

the node dies, or the commit is refused	what you get
before the commit	nothing written, nothing acknowledged, nothing handed out — the messages are read again
after the commit	records written and messages acknowledged — they are never read again

There is no step in between, which is why this source can promise what a Kafka consumer cannot: each message is applied to the store once. IDENTITY is still required — two messages carrying one identity land as one record holding the later one, and moving the group back with ALTER GROUP … START AT replays into the same records rather than beside them. What the guarantee does not cover is anything outside the store: a record written here once is not an email sent once.

One consequence to plan for: the topic and the table must be in the same database, because a transaction is one database's. Declaring one across two is refused, naming both.

The clauses

clause	what it says
`FROM order_events`	the topic, in the destination's database
`GROUP 'into-rows'`	the group it reads as — declared beforehand with `DEFINE GROUP`; its deadline, width (`IN FLIGHT`) and dead letter are what the consumer reads under
`INTO order_rows`	a table that already exists
`IDENTITY order_id`	the message field holding the record's id: a whole number or a non-empty string
`MAP amount AS total, …`	which message fields become which record fields; a field nobody named does not land
`ON FAILURE stop` or `quarantine`	what happens to a message that cannot be applied — required, no default
`PARALLELISM n`	how many members this node runs; left out, one

IDENTITY, MAP and the refusals around them behave exactly as they do for a Kafka consumer. There is no FORMAT: a topic's message is already a value.

A message that cannot be applied

ON FAILURE stop halts the consumer at that message. Nothing of the batch it was in is written — not even the read, so the message is still unread — and the reason stays in INFO FOR TOPIC CONSUMER until you fix the cause and declare the consumer again.
ON FAILURE quarantine hands the message back to the group in the same transaction that writes the rest of the batch, so good messages keep flowing. The group counts the delivery, and past its DELIVERIES the message is appended to its DEAD LETTER TO topic, where you can read it. That is why quarantine is refused on a group without both clauses: without them a bad message would be handed out forever.

A write the table refuses — a type or an assertion — counts as a message that cannot be applied: the consumer narrows to one message at a time to find which one, and the policy decides it.

How it runs

As tasks on the node's runtime, in every build. No broker, no C toolchain, no --features flag.
Declared means running. DEFINE and DROP TOPIC CONSUMER take effect within a second, without a restart; a node started later starts it from the catalog.
On a cluster, the declaration replicates and every node runs it; a node that may not write the table waits and tries again, so the node that can makes progress and consumption moves with the leadership.
With its declarer's authority, checked again every batch: revoking the declarer's read on the topic or write on the table stops it at its next batch, with the refusal in its report. Declaring one needs manage, read and write where it reaches.
At shutdown each member finishes the batch it is in and returns, at the same moment the node stops accepting connections.

Seeing what it is doing

INFO FOR TOPIC CONSUMER events_in;
INFO FOR TOPIC order_events;

INFO FOR TOPIC CONSUMER answers in the same three groups a Kafka consumer's report does — declared, running on this node (applied, quarantined, halted, last_error) and guarantees — and INFO FOR TOPIC lists the topic consumers reading it under ingested_by, each with its group, the table it writes (or hidden, when you may not read that table) and whether it is running here. The console's Topics screen shows the same list under Into tables. The group's own numbers — lag, in flight, redelivered, dead lettered — are in INFO FOR TOPIC as for any group.

Stopping one

DROP TOPIC CONSUMER events_in;

Its members finish their batch and stop within a second. The group and its position stay: declaring the consumer again resumes where the group stands.

A topic named consumer is still a topic — DEFINE TOPIC consumer, DROP TOPIC consumer and INFO FOR TOPIC consumer mean it, because what follows the word tells the two apart.

Reading a Kafka topic into a table

When the messages start in Kafka, a Kafka consumer brings them in. It is the same declaration with a broker as the source, and one difference follows from that: the broker's offset is a second commit in a second system, so this source is at least once, made safe by the record identity.

The table a Kafka consumer writes into:

DEFINE TABLE orders (total int, placed_at datetime);

DEFINE KAFKA CONSUMER orders_in
    FROM 'broker-1:9092', 'broker-2:9092'
    TOPIC 'orders'
    GROUP 'shop-orders'
    FORMAT json
    INTO orders
    IDENTITY order_id
    MAP amount AS total, placed.at AS placed_at
    ON FAILURE quarantine
    PARALLELISM 2;

Every clause except PARALLELISM is required, and they are read in that order.

A group the broker has never seen starts at the oldest message still on the topic, so what was published before the declaration is ingested too. A group that has committed an offset resumes from it.

The delivery guarantee, before anything else

At-least-once, with idempotent application by record identity. Not exactly-once, and the difference matters when you are deciding what to put in the topic.

A consumer performs two commits into two different systems — this store's transaction, and the broker's offset — and no transaction spans both. So there is one step where a crash is visible, and the order of those two commits decides what a crash there costs you.

die anywhere before step 3 nothing written · offset unmoved a clean redelivery

die inside the window data written · offset unmoved a duplicate, never a loss

the same message arrives again — and lands on the same record, because its identity came from its own payload

The store commit comes first, which chooses duplicates over loss. A duplicate is something a record identity absorbs; a loss is not.

the node dies here	what you get
after polling a batch	nothing written, offset unmoved → clean redelivery
after parsing it	as above
after the store commit	data present, offset unmoved → duplicate on restart
after the offset commit	consistent

That is what IDENTITY is for. Every record is written at the identity taken from its message, and the write replaces whatever was there — so a redelivered message converges to one record rather than making a second. Without a declared identity there would be no way to make redelivery safe, which is why the clause is required rather than optional.

INFO FOR KAFKA CONSUMER repeats all of this in its own output. That is deliberate: a guarantee documented away from the point of configuration is one that will be misread.

How it runs

The statement is a declaration, and it is stored in the catalog like a table or a user. Nothing about running it lives in a file beside the database.

The declaration replicates; the running does not. Every node that receives it starts the same consumer in the same group, and the broker then spreads the topic's partitions across them — which is the behaviour you want from a cluster, and it is why a group name that the database derived for itself would be a bug. What each node is doing — running or not, how far it has got, its last error — is that machine's own business and is never written down. A stored "running" flag outlives the thread it describes: a node killed mid-batch would come back claiming to run a consumer that is not there.

Each consumer is an OS thread, and PARALLELISM says how many. The node serves on an async runtime, but a consumer is a loop that blocks on a broker and then blocks on a commit, which is what a thread is for — and a consumer kept off the runtime cannot hold one of its workers while a broker is slow.

The loop is the four steps in the diagram above, and it takes a batch rather than a message at a time because a batch is one transaction. A batch is not atomic and does not need to be: if a node dies with half of one applied, the offset has not moved, so the whole batch comes back and the applied half is written again to the same identities.

At boot, and after a restart, the node reads the catalog and starts what it finds. A declaration whose destination has been dropped, or whose broker cannot be reached, is logged and skipped — one broken consumer must not stop a database from answering the queries that have nothing to do with it.

At shutdown, consumers stop taking new messages at the same moment the node stops accepting new connections, and each one finishes the batch it is already holding before it returns. That ordering matters: a consumer still writing while the node has told its load balancer it is not ready is a writer nobody is waiting for.

The clauses

`FROM` and `TOPIC` — where the messages are

FROM 'broker-1:9092', 'broker-2:9092'
TOPIC 'orders'

Addresses as written. Whether one resolves is a question for the moment the consumer dials it, not for the moment you declare it — refusing an unreachable address here would make the statement's success depend on the network being up while you typed it.

`GROUP` — which consumer group

GROUP 'shop-orders'

Declared, never derived. The group is a broker-side identity, and reading positions are tracked per group. If the database derived it from the node's own identifier, every node in a cluster would form its own group and every node would consume every message — a bug that cannot appear on one machine.

Every node that receives this declaration joins the same group, and the broker spreads the topic's partitions across them. That is the behaviour you want, and it is why the declaration replicates.

`FORMAT` — how a message becomes fields

FORMAT json

json today. A format this build does not read is refused where you wrote it, rather than becoming a consumer that starts and then fails on its first message.

Two things the JSON reader decides, because they are decisions rather than translations:

A number written without a fraction or an exponent is read as a whole number. A millisecond timestamp read as a double comes back rounded, and nothing anywhere would say so.
JSON's null becomes this store's null, which is present and holding nothing — not none, which is absent. A producer that wrote the key meant the first one.

`INTO` — where the records land

INTO orders
INTO shop.orders

A table that must already exist. It is resolved when you declare the consumer, so a consumer can never be reading a topic into a table that is not there yet.

You need authority over that table. Declaring a consumer is administering — it points a background writer at somebody's data — so it needs an owner, and the destination is checked against your tenancy and your grants exactly as a write to it would be.

`IDENTITY` and `MAP` — what the record is

IDENTITY order_id
MAP amount AS total, placed.at AS placed_at

IDENTITY names the message field carrying the record's id. It must be a whole number or a non-empty string. A float is refused rather than truncated, because 1.0 and 1 would become the same record and 1.5 would become 1.

MAP reads in the direction SELECT already reads: the message's field, then what the record calls it. The left side may be a path, so a nested payload works; the right side is a plain name, so the record stays flat.

A message field nobody named does not land. There is no schema inference here — not because inference is hard, but because an inferred mapping is right in your test and wrong on the day a producer adds a field. The consequence is worth stating positively: a producer adding a field is a non-event, and a field you want has to be named.

A mapped field the message does not carry is left absent, not written as null. The producer did not send it, and that is the true statement.

Mapping two message fields onto one record field is refused: which one wins would depend on which was applied last, and there is no order there that is not arbitrary.

`ON FAILURE` — a message that cannot be applied

ON FAILURE stop
ON FAILURE quarantine

Required, with no default, because a default here is a decision about data loss taken by whoever did not type the clause.

stop halts this consumer and records why. Use it when a message you cannot apply is a thing you want to look at before anything else moves.
quarantine retries a small number of times, then parks the message and carries on. Use it when one bad payload must not hold up a partition.

There is no skip mode. The failure it would cause is the one this design is most careful about: one malformed message blocking a consumer, the stalled consumer triggering a group rebalance, and the rebalance degrading every other consumer in the group. One bad payload should cost one record.

`PARALLELISM` — how many consumers this declaration runs

PARALLELISM 2

The one optional clause. Leaving it out reads as one, not as "decide for me". Zero is refused: a consumer declared to run nothing is a consumer you believe is consuming.

Each one is an OS thread of its own, beside the runtime the node serves on: a consumer is a loop that blocks on a broker and then blocks on a commit, which is what a thread is for.

Seeing what it is doing

INFO FOR KAFKA CONSUMERS;
INFO FOR KAFKA CONSUMER orders_in;

The listing says which consumers exist and which this node is running. The single report answers in three named groups:

declared — what a backup carries and what every node agrees on.
running — this node only: whether it is running here, how many records it has applied, how many it has quarantined, the last error it hit, and where it had reached in each partition.
guarantees — the delivery semantics, in the place you are configuring them.
quarantine — the messages ON FAILURE quarantine parked, each with its partition, offset, reason, payload and at (from 0.26.0-beta).

running is empty on a node that has just restarted and has not started it yet, and it is empty again the moment it stops. Nothing about it is stored: a persisted "running" flag outlives the thread it describes, and the next process to read it would report a consumer that is not there.

last_error is kept even after the consumer recovers. An error that cleared itself is the one nothing else will ever mention.

A parked message is kept in the store, unlike running. It is written in the transaction that writes its batch, so it is there after any restart and on every node, up to the newest 1 000 per consumer; before 0.26.0-beta a quarantine was a count that ended with the process. A payload was on its way into the destination table, and asking about the consumer needs only manage, so the payload is shown to a user who may read that table whole — anyone else sees each entry with its payload withheld, saying so. DROP KAFKA CONSUMER removes what it parked along with the declaration.

Both refusals — no exactly-once, no schema inference — are in that output too.

Stopping one

DROP KAFKA CONSUMER orders_in;

Forgets the declaration and frees the name. The node stops the consumer the next time it reads the catalog.

What a shutdown does

A consumer told to stop finishes the batch it is already holding — writing it and committing its offset — and then returns. It stops taking new messages at the same moment the node stops accepting new connections, so it is not still writing while the node has been telling a load balancer it is not ready.

A second interrupt exits immediately and joins nothing. That is safe here for the same reason the ordering above is: a consumer killed mid-batch has either not written, or has written and not committed its offset, and the second case is a redelivery.

Which build can run one

The broker client compiles a C library, so an ordinary build of this database does not carry it — a cargo build needs no C toolchain. The published container image is such a build: it stores a DEFINE KAFKA CONSUMER and does not run it. A topic consumer runs in every build, the image included.

cargo build --features kafka

A build without it still parses DEFINE KAFKA CONSUMER and still stores the declaration; what it cannot do is carry one out. It says so at startup, and INFO FOR KAFKA CONSUMER reports nothing running here.

What this is not

Not a transformation language. MAP renames and flattens. If your payload needs reshaping, reshape it before the topic or after the table.
Not exactly-once. Said twice on purpose.
Not a schema. The mapping is the whole schema, and it is the one you wrote.

Stream ingestion

Reading a topic into a table

The guarantee: exactly once into this store

The clauses

A message that cannot be applied

How it runs

Seeing what it is doing

Stopping one

Reading a Kafka topic into a table

The delivery guarantee, before anything else

How it runs

The clauses

FROM and TOPIC — where the messages are

GROUP — which consumer group

FORMAT — how a message becomes fields

INTO — where the records land

IDENTITY and MAP — what the record is

ON FAILURE — a message that cannot be applied

PARALLELISM — how many consumers this declaration runs

Seeing what it is doing

Stopping one

What a shutdown does

Which build can run one

What this is not

`FROM` and `TOPIC` — where the messages are

`GROUP` — which consumer group

`FORMAT` — how a message becomes fields

`INTO` — where the records land

`IDENTITY` and `MAP` — what the record is

`ON FAILURE` — a message that cannot be applied

`PARALLELISM` — how many consumers this declaration runs