nordisk/knut

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

Class: guide · audience: a techie authoring and running a Þund pipeline. Reuse: thund.uses = [pipelines] — the Spark backend drives the knut-pipelines Spark Connect client. The what is in thund-design.md; the why in thund-reasoning.md. This file is the workflow.

f0 — zero to a running pipeline (the whole workflow)

Four moves take a data flow from nothing to a checked, runnable pipeline:

1. Build the graph. Pipeline::new(name) then chain .with_dataset(..) (the output nodes) and .with_flow(..) (how each is computed). The IR types are the builder — there is no separate builder type.
2. Validate. p.validate()? — unique names, resolvable targets, acyclic. p.topo_order() gives the deterministic build order.
3. Pick a backend. NativeBackend::new() (feature native) for local Arrow runs, or SparkBackend::new("sc://host:15002") (feature spark) to lower to Spark. Call backend.check(&p)? to reject anything the backend can't express before touching the network.
4. Run. backend.run(&p)? returns a RunHandle; drain poll_events() for RunEvents that render in the knut-pipelines facett viewer.

The same Pipeline also serializes to RON (feature dsl) and is what the facett graph editor edits — one value, four surfaces.

f1 — build the graph (typed Rust builder)

The canonical form. A batch pipeline of two datasets, one flow:

use knut_thund::ir::{Pipeline, Dataset, Flow, OutputType, DatasetSchema};

let p = Pipeline::new("daily_sales")
    .with_dataset(Dataset::new("raw_orders", OutputType::Table))
    .with_dataset(
        Dataset::new("clean_orders", OutputType::MaterializedView)
            .incremental()                                  // reconciliation knob
            .with_schema(
                DatasetSchema::new()
                    .field("order_id", "Int64", false)      // Arrow display type
                    .field("amount",   "Float64", false),
            ),
    )
    .with_flow(
        Flow::batch("f_clean", "clean_orders", ["raw_orders"])
            .with_query("SELECT * FROM raw_orders WHERE amount > 0"),
    );

p.validate()?;                                              // unique/resolvable/acyclic
assert_eq!(p.topo_order().unwrap(),
           vec!["raw_orders".to_string(), "clean_orders".to_string()]);

Dataset::new / .incremental() / .with_schema() are in knut-thund/src/ir/dataset.rs; Flow::batch / .with_query() in knut-thund/src/ir/flow.rs; the schema builder in knut-thund/src/ir/schema.rs.

A streaming flow

Streaming lives on the flow, not the dataset kind. A Table fed by a Flow::streaming is a streaming table:

use knut_thund::ir::{Flow, SourceSpec, Watermark, Expectation, OnViolation};

let f = Flow::streaming(
        "f_clicks", "click_counts",
        SourceSpec::Kafka {                                 // unbounded source
            bootstrap: "localhost:9092".into(),
            topic: "clicks".into(),
            format: "json".into(),
        },
    )
    .with_watermark(Watermark {                             // explicit allowed-lateness knob
        event_time_column: "ts".into(),
        allowed_lateness_ms: 5_000,
        idle_timeout_ms: 30_000,
    })
    .with_query("SELECT window_end, count(*) n FROM clicks GROUP BY TUMBLE(ts, '10s')")
    .expect(Expectation::new("n_positive", "n > 0").on(OnViolation::Drop));

Flow::streaming / .with_watermark() / .expect() are in knut-thund/src/ir/flow.rs; SourceSpec / Watermark (and Trigger, OutputMode, WindowSpec) in knut-thund/src/ir/streaming.rs; Expectation / OnViolation in knut-thund/src/ir/expectation.rs. Because the source is unbounded, p.is_streaming() is now true.

f2 — the RON DSL round-trip (feature dsl)

The DSL is not a second model — it is the same IR as text, so it round-trips losslessly:

use knut_thund::authoring::dsl;

let ron   = dsl::to_ron(&p)?;        // pretty RON — config-as-code / LLM / warehouse
let back  = dsl::from_ron(&ron)?;    // parses AND validates
assert_eq!(p, back);                 // lossless: it's a serde view of the IR

to_ron / from_ron are in knut-thund/src/authoring/dsl.rs (from_ron validates before returning). Use this for anything a human or an LLM should edit, and for warehouse-historized definitions.

f2b — the SQL surface (feature sql)

SDP-style DDL, parsed into the same Pipeline. Reach for it when the pipeline is already written as SQL, or when you want a definition a SQL-literate colleague can read without knowing Rust or RON.

use knut_thund::authoring::sql;

let p    = sql::from_sql("medallion", script)?;   // parses AND validates
let text = sql::to_sql(&p)?;                      // renders back, topo-ordered

The grammar, verified statement-by-statement against a real Spark 4.1.3:

CREATE MATERIALIZED VIEW name
  [ ( col TYPE [NOT NULL], …
    [, CONSTRAINT c EXPECT (expr) [ON VIOLATION {DROP ROW | FAIL UPDATE}]] ) ]
  [ PARTITIONED BY (col | months(col) | bucket(n, col) | truncate(n, col)) ]
  [ COMMENT 'text' ]
  [ TBLPROPERTIES ('k' = 'v', …) ]
  AS SELECT …;

CREATE [TEMPORARY] VIEW name AS SELECT …;
CREATE STREAMING TABLE name [ (…) ] [clauses] [ AS SELECT … FROM STREAM(src) ];
CREATE FLOW name AS INSERT INTO target [BY NAME] SELECT … FROM STREAM(src);

Three things worth knowing:

Statement order does not matter. The DAG comes from each flow's reads, extracted from the parsed SELECT, so a query may reference a dataset defined further down the file. CTE names are excluded — they are local to the query, not nodes in the graph.

STREAM(t) is erased. It marks the Flow streaming and then disappears; the query a backend receives is plain portable SQL naming t directly. Both FROM STREAM(t) and the bare FROM STREAM t work. Streaming-ness is a property of the edge, never of the SELECT.

Two forms are accepted here that Spark rejects — both already in the IR: CONSTRAINT … EXPECT (Spark: PARSE_SYNTAX_ERROR near 'EXPECT'; a Databricks-DLT extension) and non-identity PARTITIONED BY transforms. Authoring them is fine — check is what tells you a backend cannot run them.

SQL and RON are two views of one value, so they interchange freely:

let p = sql::from_sql("medallion", script)?;
assert_eq!(p, dsl::from_ron(&dsl::to_ron(&p)?)?);   // bit-identical

TBLPROPERTIES lands in Dataset::properties (a BTreeMap, so RON stays diff-stable) and lowers onto DefineOutput.TableDetails.table_properties — it reaches Spark rather than being parsed and dropped.

Scaling is linear — examples/medallion_sql.rs sizes a bronze/silver/gold graph by argv and reports each stage: ~40–50k statements/s single-threaded, 6001 datasets parsed in 135 ms, topological sort 6.9 ms.

cargo run -p knut-thund --features sql,dsl --release --example medallion_sql -- 500

f3 — choose a backend, check, run

Both backends implement ExecBackend (knut-thund/src/backend/mod.rs); the choice is one line. check is the guardrail — it compares the IR to the backend's Capabilities and refuses up front, the way DataFusion's SanityCheckPlan refuses an un-runnable plan.

use knut_thund::backend::{ExecBackend, native::NativeBackend, spark::SparkBackend};

// Local Arrow/DataFusion run — accepts the whole IR:
let native = NativeBackend::new();                          // .with_state_root("s3://…") for durable state
native.check(&p)?;                                          // passes: native caps cover everything
let mut run = native.run(&p)?;                              // streams RunEvents (scaffold today)
for ev in run.poll_events()? { /* fold onto the viewer */ }

// Lower to Spark Declarative Pipelines over Spark Connect:
let spark = SparkBackend::new("sc://spark:15002").spark_4_2();  // 4.2 → sinks/persistent views/CDC
match spark.check(&p) {
    Ok(()) => { spark.run(&p)?; }
    Err(e) => eprintln!("not expressible on Spark: {e}"),   // e.g. expectations (OSS 4.1)
}

NativeBackend::new / with_state_root are in knut-thund/src/backend/native.rs; SparkBackend::new / spark_4_2 in knut-thund/src/backend/spark.rs.

Run it on your laptop — read a local file, get a table back

The shortest real run needs no Spark, no Kafka, no container — just a CSV (or parquet) on disk. with_file_input(table, path, format) registers the file under a table name your batch query reads:

use knut_thund::backend::{ExecBackend, RunHandle, native::NativeBackend};
use knut_thund::ir::{Dataset, Flow, OutputType, Pipeline};

let p = Pipeline::new("orders_rollup")
    .with_dataset(Dataset::new("by_customer", OutputType::MaterializedView))
    .with_flow(
        Flow::batch("agg", "by_customer", ["orders"])
            .with_query("SELECT customer, SUM(amount) AS total FROM orders GROUP BY customer"),
    );

let mut run = NativeBackend::new()
    .with_file_input("orders", "orders.csv", "csv")   // ← the local file becomes table `orders`
    .run(&p)?;
for ev in run.poll_events()? { println!("{}", ev.message); }
let table = run.output("by_customer").expect("materialised output");   // Vec<RecordBatch>

Runnable as-is: cargo run -p knut-thund --example batch_file_native --features native (writes a sample orders.csv, runs the rollup, prints the result table). The same path is asserted end-to-end in tests/native_file_source.rs (CSV and parquet). with_file_input is the on-disk sibling of with_input (in-memory RecordBatches) and with_stream_input (synthetic micro-batch streams).

Keep the result on disk — the file sink

run.output("by_customer") hands you the result in memory. To keep it as a file the next tool can pick up, add with_file_output(dataset, path, format) — the write-side sibling of with_file_input. After the run, the dataset's rows are written to a single local csv/parquet file (its parent directory is created if missing):

let run = NativeBackend::new()
    .with_file_input("orders", "orders.csv", "csv")
    .with_file_output("by_customer", "out/rollup.parquet", "parquet")   // ← land the result on disk
    .run(&p)?;

let sink = run.sink("by_customer").expect("sink written");   // &SinkWrite
println!("wrote {} row(s) -> {} ({})", sink.rows, sink.path, sink.format);

That closes the on-disk round trip: read orders.csv → run SQL → write rollup.parquet, no external system, no catalog. The write is observable without re-reading the file — run.sink(dataset) returns a SinkWrite { path, format, rows }; a sink pointed at a dataset the pipeline never produces writes no file and returns None (honest, not a silent empty file). Declaring a sink never changes the in-memory output path (fully additive). Asserted end-to-end in tests/native_file_sink.rs (read → SQL → write → re-ingest → identical rows, for both csv and parquet).

Or name the sink IN the IR — Dataset::with_path

with_file_output is imperative (on the backend builder). To let the pipeline itself say where an output lands — so a .ron spec carries the destination and round-trips — put the path on the dataset:

let p = Pipeline::new("orders_rollup")
    .with_dataset(
        Dataset::new("by_customer", OutputType::Sink)
            .with_path("out/rollup.parquet")        // ← the sink destination, in the IR
            .with_format("parquet"),                // optional — else inferred from the extension
    )
    .with_flow(Flow::batch("agg", "by_customer", ["orders"]).with_query(
        "SELECT customer, SUM(amount) AS total FROM orders GROUP BY customer"));

// No .with_file_output needed — the backend writes the declared path after the run:
let run = NativeBackend::new().with_file_input("orders", "orders.csv", "csv").run(&p)?;
let sink = run.sink("by_customer").expect("declared sink written");

format is optional: with none, a .csv extension picks the csv writer, else parquet. partition_cols on the dataset still drive a Hive-partitioned directory. If a dataset has both a declared path and an imperative with_file_output, the imperative one wins (no double write). Asserted in native_file_sink.rs (declarative_*) and round-tripped by dsl_roundtrips_declarative_file_sink.

Re-running a sink — overwrite vs append (Dataset::overwrite)

By default a re-run appends: a partitioned sink writes fresh part-files each run, so running the same pipeline twice into the same directory leaves two generations of rows under one path (the pre-existing behaviour — kept as the default so nothing changes silently). To make a re-run replace the prior output instead, declare Overwrite and the backend clears the destination first:

Dataset::new("by_region", OutputType::Sink)
    .with_path("out/by_region")
    .with_partition_cols(["region"])
    .overwrite();   // ← clear the dir first: a re-run holds ONE generation
    //  .with_write_mode(WriteMode::Append)  is the (default) accumulate mode

Overwrite wipes the target dir/file (remove_dir_all / remove_file) before writing, so re-running a full-refresh load leaves exactly one generation. A single-file sink is truncated by the writer either way, so the difference is visible on a partitioned/directory sink (where part-files would otherwise pile up). The mode round-trips in RON and defaults to Append (existing specs are unaffected). Asserted by write_mode_overwrite_holds_one_generation (two runs → 3 rows) vs write_mode_append_accumulates_generations (two runs → 6 rows) and round-tripped by dsl_roundtrips_write_mode.

Combine several sources into one table — union append (fan-in)

Point more than one flow at the same target and their rows union into one asset — SDP's "one table, N append flows" pattern. Each flow reads its own source and defines its own SELECT; you do not write a UNION by hand, and the sources need not be co-located:

let p = Pipeline::new("all_sales")
    .with_dataset(Dataset::new("all_sales", OutputType::Table))
    .with_flow(Flow::batch("from_east", "all_sales", ["east"]).with_query("SELECT region, amount FROM east"))
    .with_flow(Flow::batch("from_west", "all_sales", ["west"]).with_query("SELECT region, amount FROM west"));

let run = NativeBackend::new()
    .with_input("east", east_batches)      // each append flow reads its OWN source
    .with_input("west", west_batches)
    .run(&p)?;

assert_eq!(run.row_count("all_sales"), east_rows + west_rows);   // union, not clobber

Every append flow's rows land in all_sales (order-preserving, in flow-declaration order), and any downstream flow reading all_sales — or a file sink on it — sees the full union. Each flow's own expectations apply to its own rows (SDP per-flow semantics). The append flows must agree on output schema: a mismatched one fails the run loudly rather than silently dropping rows or mixing schemas. A single flow per target is unchanged (no union path). Asserted in tests/native_union_append.rs.

Read only the columns/rows you need — typed projection + filter (pushdown)

Instead of an opaque SELECT string, declare a typed projection (the columns to keep) and/or filter (a predicate) on the flow. The engine reads these directly and pushes them into the source scan — only those columns are read off disk, and non-matching rows are pruned at the source, not after a full read:

let p = Pipeline::new("big_orders")
    .with_dataset(Dataset::new("big", OutputType::MaterializedView))
    .with_flow(
        // no .with_query(..) — the typed nodes ARE the transform
        Flow::batch("pick", "big", ["orders"])
            .with_filter("amount > 100")            // rows: pushed into the scan predicate
            .with_projection(["customer", "amount"]) // columns: pruned at the scan
    );

let run = NativeBackend::new()
    .with_file_input("orders", "orders.parquet", "parquet")
    .run(&p)?;

// The output carries ONLY the projected columns …
assert_eq!(run.row_count("big"), 2);               // only amount > 100 rows survive
// … and the pushdown is provable from the physical plan:
let pd = run.pushdown("pick").unwrap();
assert!(pd.physical_plan.contains("projection=[customer, amount]"));
assert!(pd.physical_plan.contains("predicate="));  // filter reached the parquet scan

The flow reads exactly one input (reads) when it has no query. It also lowers to Spark: to_sdp synthesizes `SELECT customer, amount FROM orders WHERE amount

100, so Catalyst does its own pruning/pushdown. Combine a typed node **with** a with_query(..)and it applies on top of the query result (native-only refinement). Asserted intests/native_projection_pushdown.rs`.

What check will reject, per backend

That last row is the one people trip on, so it is worth being exact. No shipped backend runs a non-identity partition transform, and the two say no for different reasons:

• Spark cannot. PartitionHelper#applyPartitioning maps Seq[Transform] -> Seq[String] and accepts only IdentityTransform; CLUSTER BY dies in the same place. Unchanged in 4.2.
• native does not yet. The file sink maps each partition_cols entry to a literal Hive path segment, so months(event_ts) would be looked up as a column of that name and fail at write time.

Either way check refuses locally, before a graph is created or a file is written, instead of letting it surface as an opaque JVM error mid-run. The fix in both cases is the same: materialize the value as a column and partition by that.

-- rejected
PARTITIONED BY (months(event_ts))
-- accepted, same physical layout
CAST(DATE_TRUNC('MONTH', event_ts) AS DATE) AS event_month  …  PARTITIONED BY (event_month)

One caveat if you reach for Iceberg partition evolution to work around this: do not run ALTER TABLE … ADD PARTITION FIELD months(event_ts) against an SDP-managed table. The ALTER succeeds, and every subsequent run then fails permanently with CANNOT_UPDATE_PARTITION_COLUMNS. Recover with DROP PARTITION FIELD.

Pipeline::to_sdp (knut-thund/src/ir/pipeline.rs) lowers the graph shape faithfully even when a knob degrades, so the Spark viewer still renders the DAG. The version reality behind that table is in thund-reasoning.md.

f4 — incremental streaming: watermarks, checkpoints, recovery

Everything above runs on the recompute default: each trigger re-runs the flow's SQL over all accumulated rows. That is correct and at-least-once, but it grows without bound and cannot recover. The native backend also carries an opt-in incremental mode — event-time watermarks, window close + eviction, durable barrier checkpoints, crash recovery, and per-(window, group) incremental aggregation — selected with one builder call. It never replaces recompute: when a query shape or source is unsupported it degrades to recompute with an honest event, so nothing you author can silently break. The full shape, the guarantees, and their honest limits are in thund-streaming-state-design.md.

Turn it on

use knut_thund::backend::native::{NativeBackend, StreamingMode};

let native = NativeBackend::new()
    .with_streaming_mode(StreamingMode::Incremental)   // opt in; default is Recompute
    .with_state_root("file:///var/lib/thund/state");   // durable checkpoints (file:// or s3://)

let run = native.run(&p)?;                              // same ExecBackend::run as f3

StreamingMode::default() is Recompute, and NativeBackend::default() is unchanged — every existing caller keeps the recompute path with no edit. with_streaming_mode / with_state_root / with_max_triggers are in knut-thund/src/backend/native.rs; the checkpoint store is knut-thund/src/backend/checkpoint.rs::ParquetStateStore.

What the watermark drives (needs a .with_watermark(..) on the flow)

• Late-row drop. After the combined watermark W advances, rows with event_time_column < W are dropped before the operator sees them and counted — NativeRun::late_rows_dropped(flow_target).
• Window close + evict. A tumbling/hopping/session window closes once W ≥ window.end + allowed_lateness_ms; its aggregate is emitted once and its keyed state is evicted — this is what bounds memory. Count via NativeRun::windows_closed(flow_target).

Checkpoints and recovery

One epoch = one trigger. On each trigger the operator state, the watermark W + per-input maxima, the consumed source offset, the late-drop/closed counters, and the emitted-window tags are written under {state_root}/{pipeline}/{flow_target}/epoch=NNNNNN/, the JSON manifest being the atomic commit point. NativeRun::checkpoints_committed(flow_target) reports how many committed.

Recovery is automatic: run the same pipeline again against the same source and a state_root that already holds a committed epoch, and it restores state + W + offset and resumes from the committed offset (no double-work, no lost on-time rows). NativeRun::resumed_from_epoch(flow_target) is Some(epoch) on a resumed run, None on a cold start. To simulate a crash in a test or a staged restart, cap a run with .with_max_triggers(Some(n)): it checkpoints each of n triggers then stops, and the next run resumes.

// Run 1: process 2 triggers, then "crash".
let run1 = NativeBackend::new()
    .with_streaming_mode(StreamingMode::Incremental)
    .with_state_root("file:///tmp/state")
    .with_max_triggers(Some(2))
    .run(&p)?;
assert_eq!(run1.resumed_from_epoch("win_out"), None);      // cold start

// Run 2: same source + state_root → resumes and finishes.
let run2 = NativeBackend::new()
    .with_streaming_mode(StreamingMode::Incremental)
    .with_state_root("file:///tmp/state")
    .run(&p)?;
assert_eq!(run2.resumed_from_epoch("win_out"), Some(1));   // resumed

Output is at-least-once with idempotent replay: an idempotent/keyed sink (the _gar_id MERGE law, bifrost-design.md) absorbs a replayed mid-crash trigger, making incremental mode effectively exactly-once; a plain append sink stays at-least-once.

Incremental aggregation (no per-window recompute)

For a supported windowed aggregate — an Aggregate directly over a TableScan, group columns bare Int64/Utf8, aggregates COUNT/SUM/MIN/MAX/AVG over Int64, tumbling or hopping window, a pure pass-through top projection — incremental mode maintains one accumulator per (window, group) instead of buffering rows, so the working set is bounded by group cardinality × open windows, not row count. Anything outside that shape (a WHERE, a computing projection like SUM(x)+1, a session window, float source columns, DISTINCT) falls back to recompute-per-window and stays correct. NativeRun::used_incremental_agg(flow_target) tells you which path ran.

The green oracle. The incremental result equals the recompute result equals the batch path over the same rows — the equality asserted across tests/streaming_microbatch.rs. Recompute stays the correctness reference, so incremental mode is always a performance/durability win, never a semantics change.

Running the native test lane

Every native_*.rs and streaming_microbatch.rs test in crates/knut-thund/tests/ opens with a crate-level #![cfg(feature = "native")]. native is off by default (it pulls the DataFusion tree), so:

cargo test -p knut-thund            # 5 files × "running 0 tests" — green, and meaningless
cargo test -p knut-thund --features native   # the real lane: 50 tests + 13 unit tests

The first form is not a pass, it is a silence. Everything the design and guide docs claim is "asserted end-to-end in tests/native_*.rs" is only asserted under the second. CI runs the second explicitly (.github/workflows/ci.yml, the "Test knut-thund native (DataFusion) lane" step) precisely so the gap cannot reopen unnoticed. None of these tests is #[ignore]d and none needs a container, a network or any external infra — if one is red, it is red for real.

Look & feel

Runs render through the existing knut-pipelines facett viewer for free: RunEvent/RunPhase (knut-thund/src/backend/mod.rs) are shaped to fold onto knut_pipelines::PipelineRunEvent, so a native run and an SDP run look identical. Don't build a second viewer — emit events and let the pipeline viewer paint them.