Module decode 

COLUMNAR DECODE at the source edge: Kafka messages → one Arrow RecordBatch per partition per poll, under the source’s schema and bad-data policy.

The fast path decodes a whole partition’s messages with one arrow-json decoder on a fully nullable schema (a missing field or a mismatched type becomes null, never an error, and unknown fields are ignored). Only when that fails does the slow path run: a syntax-only pass separates the messages that cannot be JSON objects, the survivors decode again as one batch, and the bad ones are dropped and counted, rescued as raw bytes, or fatal — by the source’s policy. Measured (design/2026-09-15-spike-c-json-arrow-bad-data.md): the slow path costs about nothing because the failed first pass only tokenised.

The batch keeps the message order, so a parallel vector of offsets (and the row’s Kafka key) lines up with the rows; dropped messages simply have no row. Their offsets still commit — a parse error is data, not transport.

Structs

Decoded

One partition’s decoded poll: rows in message order, the offset and key of each row.

SourceDecoder

The decoder for one source: holds the resolved schema once it is known.

SourcePolicy

A source’s decode policy: the schema and what to do with data that does not fit it.

Enums

OnBadData

What a source does with a message that is not a JSON object at all.

SchemaMode

How a source’s schema is known.

Constants

META_KEY

The meta column that carries a row’s key: into a sink, the key it is emitted under.

META_OFFSET

META_PARTITION

The meta column that carries a row’s origin: the Kafka partition it came from, or the id of the operator task that produced it. A stateful operator observes event time per origin — the min across origins is its watermark, exactly as one consumer thread’s operator observes per assigned partition.

META_PREFIX

Every meta column starts with this: the engine strips them at the edges.

RESCUED

The column that carries a rescued message’s raw bytes.

Functions

arrow_schema

A schema from declared fields (hosts build SchemaMode::Declared without depending on arrow).

declared_field

A declared column: { name = "price", type = "float" }.

decode_messages

Decode messages under policy. Returns the batch, the indexes of the messages that became rows (in row order), and the indexes of the rescued messages (their rows follow, in order).

decode_schema

The declared schema made fully nullable, plus the rescue column when the policy asks for it.

split_by_partition

The batch split by its __fv_partition column: one sub-batch per origin, in ascending origin order, rows in their original order within each. A batch without the column is one origin, 0.

split_meta

Strip the meta columns back off: the data batch, its offsets and its keys.

with_keys

The batch with keys as its __fv_key meta column (replacing one already there): the key each row is emitted under, or that a downstream operator falls back to.

with_meta

The decoded batch with its offsets and keys appended as meta columns, so batch steps (a filter in particular) keep them aligned with the rows.

with_partition

The batch with partition as its __fv_partition meta column (replacing one already there).

Type Aliases

SharedSchema

A slot for a source’s inferred schema, shared by its tasks (see SourceDecoder::with_shared).