nordisk/knut

Þund — the Arrow-native streaming dataflow engine — design

Class: design · Function axis: f0 (author over one IR → run on a pluggable backend) · f1 authoring · f2 the IR · f3 ExecBackend + the two backends. Reuse: thund.uses = [pipelines] — SparkBackend reuses the knut-pipelines Spark Connect client, and the IR is a strict superset of knut_pipelines::DataflowGraph. Strict: this file states only the shape we run. Every why — the landscape, the anchor decisions, the Spark 4.1-vs-4.2 version reality, and the honest status — lives in thund-reasoning.md.

f0 — the whole (author over one IR → run on a pluggable backend)

Þund authors a batch + streaming pipeline once, as one Arrow-typed IR, and runs it on a pluggable execution backend — the native Arrow/DataFusion engine or a lowering to Spark Declarative Pipelines (SDP). The IR is the contract; front-ends and back-ends are interchangeable around it. Three moves:

1. Author on any surface (f1) — each lowers to one ir::Pipeline (knut-thund/src/lib.rs:Pipeline).
2. The IR (f2) validates, orders, and lowers: Pipeline::validate / topo_order / to_sdp (knut-thund/src/ir/pipeline.rs).
3. Run through an ExecBackend (f3): check rejects an un-runnable pipeline up front; run lowers + starts it, streaming RunEvents that fold onto the knut-pipelines viewer (knut-thund/src/backend/mod.rs:ExecBackend).

The one invariant: a pipeline drawn in facett, exported to RON, hand-tuned in Rust, and run on either backend is the same graph — because all four are the same serde-round-tripping Pipeline value.

f1 — authoring: four surfaces over one IR

knut-thund/src/authoring/mod.rs. All four emit the identical Pipeline:

The SQL surface: extend, don't fork

Built the way DataFusion's own DFParser extends sqlparser-rs — wrap a Parser, intercept the statements the dialect does not know, delegate the rest. The dependency is sqlparser 0.62 only, not the datafusion tree; it is the version datafusion 54 already resolves, so sql + native link one AST rather than two.

The one piece of genuinely new SDP syntax, STREAM(t), needs no grammar fork. Stock sqlparser already parses FROM STREAM(t) as a table-valued function and FROM STREAM t as a table with an alias, so both arrive as TableFactor::Table named STREAM and are reinterpreted in place (sql.rs:rewrite_stream) — mutating name/args/alias rather than rebuilding the variant, which keeps it immune to sqlparser adding fields. The streaming-ness lands on the Flow, where it belongs: it is a property of the edge, not of the SELECT.

reads — and therefore the DAG — is extracted by walking the parsed query's table factors, with CTE names excluded (they are query-local, not graph nodes).

Two accepted forms exceed what Spark's SDP will take, both already modelled by the IR: CONSTRAINT … EXPECT … ON VIOLATION (Spark: PARSE_SYNTAX_ERROR near 'EXPECT' — a Databricks-DLT extension) and non-identity PARTITIONED BY transforms. Authoring is not gated on the backend; Capabilities.partition_transforms is what refuses at check time.

f2 — the IR (Pipeline = Datasets + Flows)

knut-thund/src/ir. A Pipeline (ir/pipeline.rs:Pipeline) is datasets (nodes — what data should exist) and flows (edges — how each dataset is computed): the asset-graph model, so the DAG is inferred from declared inputs/outputs. The (flow.reads → flow.target) pairs are the edges.

Datasets (nodes)

Dataset { name, output_type, materialization, schema, format, partition_cols } (ir/dataset.rs:Dataset).

Dataset::to_sdp renders the schema to an SDP schema string (ir/schema.rs:DatasetSchema::to_sdp_schema_string); the native backend rebuilds a real arrow_schema::Schema (DatasetSchema::to_arrow, feature native).

Flows (edges) — batch + streaming + CDC

Flow { name, target, reads, kind, query, projection, filter, expectations } (ir/flow.rs:Flow). FlowKind (ir/flow.rs:FlowKind) is the switch:

• Batch — a bounded SELECT recomputing the target.

• Streaming { source, watermark, trigger, output_mode, window } — an unbounded query with full event-time semantics (ir/streaming.rs):

• Cdc { CdcSpec } — APPLY CHANGES INTO: keys, sequence_by, apply_as_deletes, SCD Type1/Type2 (ir/flow.rs:CdcSpec, ScdType; mirrors AutoCdcFlowDetails).

FlowKind::is_unbounded marks the streaming/CDC path (ir/flow.rs:FlowKind::is_unbounded).

Multi-flow fan-in (union append). More than one flow may name the same target — SDP's canonical "one streaming table, N append flows" shape, where each flow reads its own independently-authored source and their rows union into one asset. The IR always permitted it (flow names are unique, targets need not be; validate + topo_order accept a shared target), but the native executor used to materialize per flow (deregister + re-register the target), so the last flow silently clobbered every earlier flow's rows. The executor now unions them: once a prior flow has produced a target in this run, the next flow's fresh output is spliced onto the accumulated rows and re-materialised (backend/native.rs:union_append), so downstream flows and the file sink see every append flow's rows — order-preserving in flow-declaration order. Each append flow's expectations still apply to its own rows (evaluated inside its run_flow, before the splice), matching SDP per-flow semantics. Loud gap (L4): an append flow whose output schema differs from the accumulated one fails the run rather than silently dropping rows or emitting a mixed-schema table — a union is only meaningful over a common schema. Additive: a single flow per target (the common case) never enters the union path and is byte-for-byte unchanged. Covered by tests/native_union_append.rs (two_append_flows_union_into_one_target, three_way_fan_in_accumulates_all, downstream_flow_sees_full_union, schema_mismatched_append_flow_fails_loudly, single_flow_per_target_unchanged).

Each append flow reads its own source; union_append splices prior ++ fresh and re-materialises, so the target, downstream flows, and the sink see every flow's rows (schema-mismatched appends fail loudly).

Typed projection / filter — the introspectable pushdown node. A flow's transform was, until now, only the opaque SQL query string: the engine could not read it to prune columns or push a filter into the source scan — it handed the whole string to DataFusion (or Spark) and hoped. Flow.projection: Vec<String> (a select column list) and Flow.filter: Option<String> (a simple predicate) are structured nodes the planner introspects directly (ir/flow.rs:Flow, builders with_projection / with_filter, predicate has_pushdown). They LOWER to real scan pushdown on both backends:

• Native — a projection/filter flow with no query reads its single reads[0] input directly and applies the typed nodes via the DataFusion DataFrame API (ctx.table(input).filter(expr).select_columns(cols), backend/native.rs:run_flow). DataFusion's optimizer then prunes columns AND pushes the predicate into the source scan — for a parquet file the physical plan shows DataSourceExec: … projection=[customer, amount], predicate=amount@2 > 100, pruning_predicate=…, i.e. id/note are never read and non-matching row-groups are skipped. The proof is captured on the run (backend/native.rs:Pushdown, NativeRun::pushdown(flow)): projected columns, the pushed filter, and the optimized physical-plan text — so a caller/test can assert the pruning reached the source, not a post-scan ProjectionExec.
• Spark — Flow::to_sdp synthesizes the equivalent SELECT <cols> FROM <input> [WHERE <pred>] (ir/flow.rs:Flow::synth_query) when the flow has a typed node and no explicit query, so Catalyst does its own column pruning + filter pushdown from the generated SQL.

Additive (L2): the opaque-SQL path is byte-for-byte unchanged; a flow with a query still runs it (projection/filter, if also present, compose on top of the query result). A flow with neither a query nor a typed node stays a structure-only edge. Streaming/CDC flows are still query-driven; the typed single-input scan-pushdown path is the batch case (where it is the real query optimization win). Covered by tests/native_projection_pushdown.rs (projection_prunes_columns_at_scan, filter_drops_nonmatching_rows, projection_and_filter_compose, typed_pushdown_lowers_to_spark_select) + lib.rs:dsl_roundtrips_typed_projection_filter.

The typed node lowers to a DataFusion scan that reads only [customer, amount] of only the amount > 100 rows (id/note pruned, non-matching row-groups skipped), and to the equivalent Spark SELECT … WHERE … — real pushdown from structure the planner can read, not an opaque SQL string.

Expectations (data quality)

Expectation { name, constraint, on_violation } with OnViolation = Warn | Drop | Fail (ir/expectation.rs:Expectation, OnViolation) — the analogue of Dagster's blocking asset checks and SDP/DLT CONSTRAINT … EXPECT … ON VIOLATION.

Validation & lowering

Pipeline::validate (ir/pipeline.rs:Pipeline::validate) enforces unique dataset/flow names, every flow target resolving to a declared dataset (reads may be external), and an acyclic in-graph DAG. topo_order (ir/pipeline.rs:Pipeline::topo_order) is a deterministic Kahn ordering; is_streaming (ir/pipeline.rs:Pipeline::is_streaming) is true if any flow is unbounded. to_sdp (ir/pipeline.rs:Pipeline::to_sdp) lowers the graph shape to a knut_pipelines::DataflowGraph; the streaming knobs and expectations SDP cannot express are dropped here and surfaced by the backend capability report — the shape always lowers faithfully.

The honest drop report — SdpLoweringReport. to_sdp dropping a knob used to be silent: a watermark/window streaming flow lowers to an SDP flow with the query but no event-time semantics, and ExecBackend::check does not refuse it (SDP accepts a streaming table and runs it on Spark's defaults — the silent-acceptance gap). SparkBackend::lowering_report(&Pipeline) (backend/spark.rs:SparkBackend::lowering_report) makes it legible: an SdpLoweringReport (backend/mod.rs:SdpLoweringReport) with one SdpFlowDrop per lossy flow, naming exactly which knob is dropped — watermark, window, non-default trigger, non-append output mode, expectations, and CDC (the whole flow on a 4.1 target; the SCD-Type2 history on a 4.2 target, where Type1 lowers losslessly). is_lossless() is the gate a caller/test asserts. This is the machine-readable, per-pipeline complement to the static Capabilities, and the honest half of the Iceberg↔Spark streaming gap: what this report lists as dropped is precisely what the native backend runs. Proven end-to-end over one IR by tests/iceberg_spark_gap.rs (Spark accepts + reports the drop; native runs the watermark late-drop + window close on the same Pipeline).

f3 — ExecBackend + the two backends

knut-thund/src/backend/mod.rs. The seam that makes the central promise work:

trait ExecBackend {
    type Run: RunHandle;
    fn capabilities(&self) -> Capabilities;       // what it can express/run
    fn check(&self, p: &Pipeline) -> Result<()>;  // up-front reject (default impl)
    fn run(&self, p: &Pipeline) -> Result<Self::Run>;  // lower + start; streams RunEvents
}

• Capabilities { name, batch, streaming, event_time, cdc, expectations, output_modes } is reported per backend (backend/mod.rs:Capabilities).
• The default check (backend/mod.rs:ExecBackend::check) validates the IR against those caps before execution — a pipeline relying on a feature the backend lacks is rejected up front, not mid-run.
• RunEvent / RunPhase (backend/mod.rs:RunEvent, RunPhase) are shaped to fold straight onto knut_pipelines::PipelineRunEvent, so a native run and an SDP run render identically in facett. RunHandle drains them (backend/mod.rs:RunHandle::poll_events / cancel).

Two backends ship:

• NativeBackend (backend/native.rs:NativeBackend::capabilities / run) — lowers each flow to a DataFusion LogicalPlan, wraps unbounded sources in streaming operators, attaches the watermark tracker + window operators + checkpoint injector, and runs RecordBatch streams on tokio; state is incremental Parquet on object storage (with_state_root), transport is Arrow IPC/Flight. Scaffolded — see reasoning · status.

Feeding it data — three seams, all bounded-testable in-process:

with_file_input is the on-disk sibling of with_input: a plain batch flow whose query reads name scans the local file directly (parquet/csv), with no in-memory pre-load and no live infra. It reuses the same register_named_file helper the streaming file-source path (SourceSpec::Batch/FileDrop) already used, so files register exactly once, the same way. This is the first native path a consumer can run end-to-end on a laptop — see the batch_file_native example + tests/native_file_source.rs.

Landing the result — the file sink (write side):

with_file_output is the write-side sibling of with_file_input: after the topo-ordered dataflow runs, each requested dataset's run.outputs batches are written to a single local file (parent dir auto-created). It closes the on-disk loop — read a file, run a SELECT … GROUP BY, land the result as a file the next tool can pick up — with no external system and no catalog. The write is observable from the return value: NativeRun::sink(dataset) reports the resolved path, format, row count, and partition_cols (a SinkWrite), and a dataset that produced nothing writes no file and reports nothing (honest, not a silent empty file). Additive: declaring a sink never changes the in-memory NativeRun::output path. See tests/native_file_sink.rs (read → SQL → write → re-ingest → identical rows).

Declarative file sink — Dataset::with_path. The write side used to be imperative-only (the path/format lived on the with_file_output builder, not in the IR), so a pipeline loaded from RON could not say where its outputs land — an asymmetry with the source side, which the streaming SourceSpec already expresses declaratively. Dataset now carries an optional path (round-trips in RON, skip_serializing_if = "Option::is_none"): a produced dataset that names a path is written after the run through the same write_file_sink (single file, or a Hive directory when it also declares partition_cols), reusing Dataset.format or inferring from the extension. Exec step 4 first drains the imperative seed_file_outputs, then (step 4b) writes any IR-declared sink whose dataset is not already covered imperatively — so with_file_output wins over a declared path (no double write). This lets a .ron spec round-trip a full read→SQL→write pipeline with no imperative wiring. Covered by tests/native_file_sink.rs (declarative_sink_writes_file_without_imperative_output, declarative_sink_infers_csv_from_extension, imperative_sink_overrides_declarative_path, declarative_partitioned_sink_writes_hive_directory) + the RON round-trip dsl_roundtrips_declarative_file_sink.

Partitioned sink — honoring Dataset.partition_cols. When the sink dataset declares partition columns (Dataset::with_partition_cols(["region"])), the write lands a Hive-partitioned directory — region=US/…/part.parquet, region=EU/… — via DataFusion's partitioned writer (DataFrame::write_parquet / write_csv with with_partition_by), instead of a single file. The IR field was previously inert on the native backend; the sink now consumes it. Every declared partition column must exist in the produced schema — a missing one fails the run loudly (mirroring the file-source loud-gap law), never a silent unpartitioned dump. Additive: an empty partition_cols (the default) takes the single-file path unchanged. Covered by partitioned_parquet_sink_writes_hive_directory (region=US/+region=EU/, read back with region from the path) and partition_col_absent_from_schema_fails_loudly.

Declarative write mode — Dataset.write_mode (Append vs Overwrite). The file sink writes fresh part-files on every run, so a re-run of a partitioned sink into the same directory used to accumulate — DataFusion emits new part-files under generated names and the old ones stayed, leaving two (or N) generations of rows under one path. Dataset now carries a typed write_mode: WriteMode { Append, Overwrite } (ir/dataset.rs):

clear_sink_target runs at the top of write_file_sink (before both the single-file and partitioned branches), so Overwrite gives fresh-dir semantics uniformly; a single file is truncated by the writer either way, so the distinction is observable on a partitioned/directory sink, where part-files would otherwise pile up. Authored with Dataset::with_write_mode(mode) or the Dataset::overwrite() shorthand; the mode is read at both sink call sites (an imperative with_file_output honors a write_mode declared on the IR dataset). Additive (L2): write_mode is #[serde(default)] → Append and skip_serializing_if = "WriteMode::is_append", so existing RON specs deserialize unchanged, keep appending, and the default serializes to nothing (byte-identical wire). Not lowered to SDP: write_mode governs the native on-disk file sink (path) — a native-backend concept that, like path itself, is not part of the SDP declarative graph; SDP already expresses overwrite-vs-append through the output type / materialization (a MaterializedView is fully recomputed each run — overwrite semantics — while a Table fed by a streaming flow appends), both of which to_sdp already lowers. Covered by write_mode_overwrite_holds_one_generation (two runs → 3 rows, not 6), write_mode_append_accumulates_generations (two runs → 6 rows, each group twice), write_mode_default_is_append, and the RON round-trip dsl_roundtrips_write_mode. RED-when-broken: neutralizing clear_sink_target reverts the Overwrite sink to accumulate (6 rows) and fails the overwrite test while Append stays green.

The same flow lands 3 rows/run. Append (default) keeps the dir, so a re-run piles a second generation of part-files under one path (6 rows); Overwrite runs clear_sink_target (remove_dir_all) first, so a re-run holds one generation (3 rows). Native file-sink only; serde-default Append keeps existing specs byte-identical.

Partitioned source — reading a Hive directory back (the round-trip's read side). with_partitioned_file_input(name, path, format, partition_cols) registers the partitioned directory the sink produced and recovers each partition column from its col=val/… path segments via DataFusion's table_partition_cols (values come back as Utf8 — standard Hive semantics), so a flow can SELECT region, … over a directory where region lives only in the path, not the data files. This closes the partitioned round trip through thund's OWN source path — write region=US/… with the partitioned sink, read it straight back with region reconstructed — rather than a hand-rolled directory walk. It shares the same register_named_file helper as the plain source (an empty partition_cols is exactly with_file_input), so files still register exactly once. Covered by partitioned_parquet_source_round_trip and partitioned_csv_source_round_trip (write partitioned → read back → region recovered from the path, exact rows; RED-when-broken — without the wiring the SELECT region fails to plan).

• SparkBackend (backend/spark.rs:SparkBackend::run) — lowers the IR via Pipeline::to_sdp and drives a Spark Connect server through the existing knut-pipelines client; it runs no engine of its own. spark_4_2() (backend/spark.rs:SparkBackend::spark_4_2) flips on sinks/persistent views/CDC. The SDP version reality is in reasoning · Spark 4.1 vs 4.2.

See also

• pipelines — the SDP/Spark Connect viewer Þund reuses and lowers onto (DataflowGraph/Dataset/Flow, the four-command dance).
• bifrost — knut's other runtime (table→graph CDC fabric); Þund's FileDrop/Cdc sources share its lineage.
• thund-guide.md — how to author + run a pipeline.
• thund-reasoning.md — the landscape, the why, the status.