nordisk/knut

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

Class: reasoning. This file holds the why, the trade-offs, and the alternatives we rejected. The descriptive shape we actually run is in thund-design.md, which by rule mentions none of this. If you want to know "why not X?" — X is here. So is the honest status. Reuse: thund.uses = [pipelines] — the Spark backend reuses the knut-pipelines Spark Connect client; the IR is a strict superset of knut_pipelines::DataflowGraph.

Þund (Old Norse Þund, "the swelling/roaring river") is the river that flows in front of Valhöll — fitting for a streaming dataflow engine. It is the "Airflow killer" half of the pipelines story; knut-pipelines is the viewer, which Þund reuses and lowers onto. It sits alongside Bifröst (the table→graph fabric) as knut's second runtime.

The landscape — what we read, and what we steal from each

A wide read of the dataflow/orchestration world. The recurring finding: almost nobody is Arrow-native end to end — most engines touch Arrow only at the edges and pay a conversion tax. Only Arroyo and DataFusion carry RecordBatch internally, and each leaves a gap the other fills. That gap is Þund's opening.

The anchor: Arroyo's strategy on DataFusion

Þund's native backend reuses DataFusion's planner + physical operators + expression evaluator and wraps the unbounded ones in streaming operator versions — exactly Arroyo's approach, which is the only proven way to get a Rust+Arrow streaming engine without re-writing query kernels. We carry DataFusion's orthogonal Boundedness × EmissionType so the planner mechanically rejects unrunnable streaming plans up front (its SanityCheckPlan rule), surfaced in Þund as ExecBackend::check (knut-thund/src/backend/mod.rs:ExecBackend::check). Concretely:

• From Arroyo: disaggregated state as incremental Parquet on object storage (S3), not RocksDB; the dual append-only vs updating SQL semantics; a single self-contained binary that scales unchanged to a cluster. We improve on it with an explicit allowed-lateness knob (Arroyo drops late rows silently) — Watermark::allowed_lateness_ms (knut-thund/src/ir/streaming.rs:Watermark) — and a real programmatic authoring API (Arroyo is SQL-only).
• From DataFusion: the ExecutionPlan/SendableRecordBatchStream backbone and Flight shuffle — and we add what it deliberately omits (event-time watermarks, windowing, barrier checkpointing) as first-class.
• From Materialize/RisingWave: (data,time,diff) collections + arrangements backed by Arrow buffers; S3-backed LSM state; Chandy-Lamport barrier checkpointing for exactly-once; arrow-udf over Flight for polyglot UDFs.
• From Dremio: the end-to-end Arrow invariant and Flight as the default transport (knut already has the Flight stack in graphar-flight).
• From Dagster: the asset model as the primary abstraction (below).
• From NiFi: per-edge back-pressure (count + byte thresholds) — the exact shape of nordisk's GATLING slow-sink law.

Why all-three-over-one-IR (why we do not pick one authoring model)

The brief's hypothesis was to make all three authoring surfaces lower to one IR, rather than choose a single model. That is the right call because each surface serves a different user without forking the model, and the IR stays the single source of truth:

1. Typed Rust builder — the canonical, compile-time-typed form. The IR types are the builder (Pipeline::new(..).with_dataset(..).with_flow(..)), so no separate builder type is needed and there is nothing to keep in sync.
2. Thin declarative DSL (RON) — authoring::dsl (feature dsl) is a pure serde (de)serialization of the same IR. It is config-as-code, the LLM-authoring surface, and what gets warehouse-historized (defs + runs). Because it round-trips losslessly through the IR's serde impls, there is no second model and therefore no drift — the law nordisk keeps relearning.
3. facett graph editor — the visual node-graph editor reuses the existing knut-pipelines PipelineView; it edits the IR's (dataset, flow) graph directly and emits the same Pipeline. CLI/Rust/DSL/visual parity is thus structural, not bolted on.

Rejected: pick one surface. Rust-only excludes config/visual users and LLM authoring; DSL-only loses compile-time types; visual-only has no diffable text. Over-one-IR keeps every strength and a single source of truth — a pipeline can be drawn in facett, exported to RON, hand-tuned in Rust, and run on either backend, unchanged. The cost is that the IR must be a faithful superset of everything the surfaces express; we pay that willingly because it is the whole thesis.

Why an asset graph, not a task graph

The IR is datasets (nodes — what data should exist) and flows (edges — how each dataset is computed), Dagster's software-defined-asset model: declare the artifact, infer the DAG from declared inputs/outputs, then recompute only what is stale (reconciliation), not blind task-order re-execution. The Materialization::{Full, Incremental} knob (knut-thund/src/ir/dataset.rs) is exactly the reconciliation lever. A task graph encodes order and is blind to the data it touches, so it can only re-run; an asset graph knows the data and can reconcile. This is the single deepest reason Þund is not another orchestrator.

Why it kills Airflow

Airflow's DAG encodes task order and is blind to the data it touches, so it can only re-execute, never reconcile; in Airflow 3 even "assets" are trigger signals, not data. Data flows through the metadata DB (XCom), unfit for DataFrames. The scheduler re-parses every DAG every loop. It is Python-only. Þund inverts every one of those:

• Asset graph, not task graph (Dagster) — declare the artifact, infer the DAG, recompute only what's stale. Incremental reconciliation, not blind re-run.
• Arrow end-to-end (Dremio) — one columnar format storage→exec→wire, Flight transport, zero-copy. No XCom-through-the-DB.
• Streaming first-class — watermarks/triggers/output-modes/windowing/CDC in the IR, run natively or lowered to SDP. No orchestrator does this.
• Pluggable backend — dev locally on the native engine, run on a cluster via Spark-SDP, unchanged. No cold-JIT-cluster wait for the inner loop (the ADF anti-pattern).
• Parity over one IR — Rust + DSL + visual + CLI all produce the same pipeline (the nordisk parity law), and every component emits its state as readable data for the screenshot-proof viz test matrix.
• Data-quality + provenance built in — expectations as blocking checks, runs historized to the warehouse as an append-only event log (Temporal/NiFi lineage), back-pressure per edge (NiFi / the GATLING law).

Spark Connect / SDP — the version reality (4.1 vs 4.2)

From primary-source research, the facts that gate the Spark backend (knut-thund/src/backend/spark.rs). These are the reason SparkBackend's capabilities look the way they do, and why they are pinned to a Spark tag:

• SDP GA'd in Apache Spark 4.1.0 (released 2025-12-16), under SPARK-51727 — not 4.0 (4.0.0 = 2025-05-23 does not contain it). Any "Spark 4.0 SDP" claim is wrong about GA.
• The wire target is spark/connect/pipelines.proto at sql/connect/common/src/main/protobuf/spark/connect/ — absent from branch-4.0. There is no new gRPC service: every pipeline op rides on SparkConnectService.ExecutePlan, wrapped in Command.pipeline_command (field 19). The four-command dance is CreateDataflowGraph → DefineOutput per dataset → DefineFlow per flow → StartRun (then read the streamed PipelineEventResult). This is exactly what knut-pipelines already implements, which is why SparkBackend reuses it rather than reimplementing.
• What requires 4.2+: external sinks (create_sink) and persistent views (SPARK-53651); AutoCdcFlowDetails (CDC) is 4.2-era on master. SparkBackend::spark_4_2() (knut-thund/src/backend/spark.rs:SparkBackend::spark_4_2) flips those capabilities on.
• Expectations are documented on Databricks/Lakeflow but not yet fully upstreamed into OSS Spark 4.1 — hence expectations: false by default (knut-thund/src/backend/spark.rs:SparkBackend::capabilities) and the planner refuses to lower an expectation-bearing pipeline onto Spark (it keeps it in the IR; the native backend runs it).
• Pin to a tag. DefineOutput/DefineFlow field names drift between 4.1.0, 4.1.2 and 4.2 master — always generate tonic stubs from the proto at the exact target Spark tag.
• Arrow on the wire: Spark Connect transports results as Arrow-encoded batches over the ExecutePlanResponse stream — so even the Spark path stays Arrow at the edge, keeping the end-to-end-Arrow invariant intact.

Rejected: expectations as a lowered SDP feature (for now)

We could try to lower Expectations onto whatever SDP-adjacent constraint support exists. We don't: since expectations are not fully upstreamed in OSS 4.1, SparkBackend::capabilities reports expectations: false and the default check refuses the pipeline up front rather than lowering something that will silently no-op. The expectation stays in the IR; the native backend (knut-thund/src/backend/native.rs) runs it as a filter/abort operator on the RecordBatch stream. Better to reject loudly than to lower a lie — the SanityCheckPlan discipline applied to a capability gap.

The Iceberg↔Spark streaming GAP — and how we close it honestly

The whole point of the two-backend seam is that the one IR carries streaming knobs SDP-over-Connect cannot express, and that we neither fake them on Spark nor lose them silently. There are three honest moves, all landed:

1. The native backend RUNS them. NativeBackend executes the event-time watermark (late-row drop, counted), the window close/emit/evict (tumbling/hopping/session), incremental accumulators (COUNT/SUM/MIN/MAX/AVG), barrier checkpoints + crash recovery, and CDC SCD Type1/Type2 — all on real DataFusion, proven equivalent to a recompute oracle (tests/streaming_microbatch.rs, 25 tests).
2. The Spark path capability-reports the drop — per pipeline, per flow. SparkBackend::lowering_report(&pipeline) returns an SdpLoweringReport (backend/mod.rs) that names exactly which knob is dropped on which flow: watermark, window, non-default trigger, non-append output mode, expectations, and CDC (whole-flow on a 4.1 target; the SCD-Type2 history on a 4.2 target). This is the machine-readable half of what Capabilities states statically — and it exposes the silent-acceptance gap: check does not refuse a watermark/window streaming flow (SDP accepts it and runs it on Spark defaults), so without the report the loss would be invisible. Now it is not.
3. The proof ties both ends over one IR. tests/iceberg_spark_gap.rs: one streaming IR → Spark check accepts it, lowering_report lists the watermark+window drops, and the SAME IR on the native backend drops the late row + closes the windows. What SDP drops, Native runs — red-when-broken on either side.

The deeper Iceberg-side gap (a true incremental append scan between two snapshots, the Spark/Flink streaming-read primitive) is missing from rust-iceberg upstream too (apache/iceberg-rust #1469, closed not-planned) and from our vendored skade/vendor/iceberg-arrow58 fork (it has single-snapshot time-travel snapshot_id() + hidden-partition transforms + writes, but no from_snapshot → to_snapshot range scan). Java's BaseIncrementalAppendScan is the model; the fork plan is in skade's .nornir. Until it lands, thund reads Iceberg via a full/single-snapshot scan and does incremental/CDC one layer up (bifrost's _change_type changelog + thund's CDC operator).

Status — honest

• IR (datasets/flows/expectations, batch + streaming, Arrow-typed) — complete and validated (validate / topo_order / to_sdp / SparkBackend::lowering_report, knut-thund/src/ir/pipeline.rs).
• Authoring: typed Rust builder (the IR itself) + RON DSL (dsl, lossless round-trip) + SQL (sql, SDP-style DDL via an sqlparser extension) + the facett graph editor (reuses knut-pipelines PipelineView / facett_adapter). All four lower to the same Pipeline.
• ExecBackend trait + Capabilities / RunEvent + SdpLoweringReport. NativeBackend genuinely RUNS batch + streaming (watermark/window/CDC/ incremental-agg/checkpoint-recovery) on DataFusion — no longer a stub. SparkBackend lowers to SDP + drives a real Spark Connect server under the spark feature (knut-pipelines grpc); with no server it returns an honest transport error after proving the lowering. spark_4_2() flips sinks / persistent views / CDC on.
• Tests: the in-crate suite + native_*/streaming_microbatch/ iceberg_spark_gap/sql_frontend integration lanes are green under --features native,spark,dsl,sql (the CI native lane covers the gated files). One IR proven on both backends (iceberg_spark_gap.rs).

Build-out still ahead (honestly not yet wired): the S3-Parquet state backend for a distributed run (checkpoints are local-fs today), Flight shuffle across a cluster, warehouse historization of defs + runs, and the Iceberg incremental append-scan source (blocked on the iceberg-arrow58 fork above).

See also

• pipelines — the SDP/Spark Connect viewer Þund reuses and lowers onto (the four-command dance, DataflowGraph).
• bifrost — knut's table→graph CDC fabric; Þund's FileDrop/Cdc sources share its lineage.
• thund-design.md — the chosen shape, tersely.
• thund-guide.md — how to author + run.