nordisk/knut

Submit models: streaming, "Rust connect", and SDP

Decision doc. Can you hand knut a Rust function, walk away, and have it keep running — the spark-submit shape? Or is SDP (Spark Declarative Pipelines) the road there?

Every claim below is anchored to file:line. Nothing here was built or benched for this doc; it is a read of the shipped code.

1 — The direct answer, in five sentences

No — not today, on any path. knut has a real streaming engine and a real "insert a Rust function" seam, but they are the same process as your client: the Transform trait (crates/knut-bifrost/src/transform.rs:26) runs in-process, and the Iceberg→FalkorDB sync loop (crates/knut-bifrost/src/listener.rs:372, watch_iceberg_delta) is an async fn your binary owns. If you go through Spark instead, start_run (crates/knut-pipelines/src/client.rs:373) opens an ExecutePlan gRPC stream and drains it to completion (:394-407) while run_on_spark (crates/knut-thund/src/backend/spark.rs:225) blocks on it — kill the client and the run's event stream dies with it, because an SDP dataflow graph is Connect-session-scoped. Your Rust code never reaches the JVM in any case: there are zero UDF registrations anywhere in the workspace (grep -rn 'ScalarUDF\|create_udf\|register_udf' crates/ → no hits), and Spark Connect UDFs are JVM/Python-serialized, so a Rust fn cannot cross that boundary. What actually happens if you try: you either write your own main that holds the loop open forever (works, but that is the client), or you get a Spark run that is alive exactly as long as your terminal is.

2 — The three models, compared honestly

(i) knut's own Rust engine — knut-thund micro-batch / DataFusion

Real engine, honestly built: watermark tracker with explicit allowed_lateness and idle_timeout (backend/native.rs:1572), tumbling/hopping/session windows that close once and evict (:2002), incremental aggregation (:3003), and barrier/epoch checkpoint+resume against an object-store state_root (backend/checkpoint.rs). 25 tests in tests/streaming_microbatch.rs (1365 lines) assert late-drop, window close, session gaps and checkpoint resume against real DataFusion — this is not a smoke suite.

The gap is intake, not semantics. Sources are pre-materialised: with_stream_input(name, micro_batches: Vec<Vec<RecordBatch>>) (native.rs:196) means you build the micro-batches. Kafka and CDC sources are explicitly deferred — native.rs:1231 matches SourceSpec::Kafka { .. } | SourceSpec::Cdc { .. } and skips the flow, emitting "deferred: live/unbounded source has no in-process data" (:1276); the test unseeded_kafka_flow_still_defers (tests/streaming_microbatch.rs:271) pins that as intended. There is no Kafka client dependency in the whole workspace. Batch and FileDrop sources do resolve to real object-store scans (native.rs:818-819).

No server, no daemon, no binary, no scheduler. Detachment: not applicable — there is no client to leave.

(ii) Spark Connect from Rust

Two consumers: the thund spark backend (lowers the Þund IR to an SDP dataflow graph, backend/spark.rs), and knut-bifrost's SparkConnectSource (issues Spark SQL: REFRESH TABLE, {table}.snapshots, system.create_changelog_view, crates/knut-bifrost/src/source.rs:200-287).

If the client disconnects: the run stops being observable and, for SDP, the graph is gone — define_graph (client.rs:354, wrapping create_dataflow_graph at :260) and start_run (:373) are session-scoped commands on one Connect session. There is no re-attach handle in the proto knut vendors.

Also worth knowing before you lean on this: capabilities() (spark.rs:76-95) honestly declares event_time=false, expectations=false, incremental_state=false, output_modes=["append"]. So a watermarked, windowed pipeline — the interesting case, and the one the thund guide leads with — cannot be lowered onto Spark. And spark.rs:275-277: with no pipeline.storage set, the run silently degrades to dry (validate-only) and still reports success events. Do not mistake a green dry run for work done.

(iii) spark-submit / SDP

SDP GA'd in Spark 4.1.0 (SPARK-51727, donated by Databricks at the 2025 Data+AI Summit) — not 4.0; knut's own comment (spark.rs:29-33) is right that pipelines.proto is absent from branch-4.0. Our pinned line is 4.1.2 (Iceberg 1.11.0 publishes no iceberg-spark-runtime-4.2), so yes, our line ships SDP.

What SDP actually is: declarative dataset definitions (materialized views, streaming tables) with automatic dependency ordering and managed checkpointing. Authored in Python or SQL only — .py or .sql files. There is no plugin ABI and no Rust surface; SDP will never run your Rust function.

On detachment: spark-pipelines run "launches an execution of a pipeline and monitors its progress until it completes." It is built on spark-submit and accepts --remote sc://…, but neither 4.1 nor 4.2 documents a detached/background mode. 4.2 adds external sinks (create_sink, Python-only) and injects spark into the module namespace — neither changes the lifecycle.

So SDP is orthogonal to the question. It is a better way to declare datasets, not a way to detach Rust.

Sources: Spark 4.1.0 SDP guide, 4.2.0 SDP guide, SPARK-51727, Spark 4.1.0 release.

3 — The honest gap, and the four candidate fixes

The gap is one sentence: jera can already detach a process or a container, but knut ships no headless Iceberg→FalkorDB binary to put in one.

crates/knut-bifrost/src/bin/bifrost.rs — the only shipped daemon-shaped binary, with JSON logs, graceful shutdown and a Prometheus /metrics endpoint — imports source::FileSource (:29) and constructs only FileSource::new(&files) (:230). Grep it for IcebergSource|SparkConnectSource|skade: zero hits. There is no --warehouse, no --iceberg, no --spark flag. The Iceberg lanes are library-only.

Candidate A — jera (edda) as the detach mechanism, knut as the executor

Smallest real path. jera has a redb-backed durable JobStore (edda/crates/jera/src/jobs.rs:597-772), a gRPC agent transport, resource-floor placement and kill semantics (agent.rs:761-836). It is a genuinely good spark-submit analogue and it already works — dwarves' runa runner is the proof (dwarves/crates/dwarves/src/runa_exec.rs, mapping each runa::Step onto a WorkUnit).

The limit, stated plainly: WorkPayload (jera/src/agent.rs:143-150) is exactly Process(ProcessSpec) | Container(ContainerRun) | Vm(VmBoot). You ship a spec that names an already-built binary or image. The code does not cross the wire. So "submit my Rust function" degrades to "build it, then submit a spec". For a fixed pipeline that is entirely fine — and it is what spark-submit does too (you submit a jar).

Cost: add an --iceberg <warehouse> / --spark <uri> arm to bifrost run over the already- generic Listener<Src, Snk> (listener.rs:55/72) — the plumbing exists, this is a clap enum plus one constructor. Then containerise and dispatch as WorkPayload::Container. Days, not weeks.

Candidate B — a knut server exposing a submit RPC

Reimplements jera's ledger, placement, and kill semantics inside knut. A trap. We would own a second job scheduler with worse durability than the one already shipping. Reject unless knut needs a submit surface that is meaningfully different from "run this unit" — and it does not.

Candidate C — DataFusion UDFs

Cheap and useful, but it does not answer the question. A ScalarUDF registered in the native backend runs in your own process — the same place Transform already runs. It buys expressiveness inside thund; it buys zero detachment. Worth doing on its own merits, not as an answer to this.

Candidate D — Arrow FFI into Spark (Rust UDF on the JVM)

The real trap. This is a JNI/Arrow-C-Data-Interface bridge, a native library shipped to every executor, a serialization contract with Connect's UDF protocol, and a permanent maintenance liability against a Spark version we do not control. Enormous cost to reach a place we do not want to be (our Rust running inside a JVM we boot only for catalog reach). Reject.

The one that gets closest to the literal ask

Ship the detached bifrost daemon (A), then let it load a wasm transform at runtime. bifrost already hosts wasm (src/wasm.rs, wasmi, Arrow IPC in/out) and Arrow-Flight DoExchange plugins (src/flight.rs). A wasm module is shippable, sandboxed code that already speaks Arrow. "Insert a Rust function and leave" becomes: compile to wasm, hand the bytes to the running daemon. That is a plugin-load path away, not an engine away — most of it is built.

4 — Recommendation

Take Candidate A now: wire the Iceberg source into the bifrost CLI, containerise it, and dispatch it through jera. Do not pursue SDP for this question, and do not build a knut server. Keep C (DataFusion UDFs) as an independent expressiveness item. Keep the wasm plugin-load as the phase-2 that delivers the literal "insert a Rust function" ask.

Strongest counter-argument, stated

"That is not submitting Rust at all — it is building a container and running it. Spark lets you write a job and submit it in one motion; you are proposing a build step, an image push, and a spec. That is a worse developer loop, and you are conceding the actual feature."

Answer

Half conceded, half not. Conceded: the code does not cross the wire, and for genuinely ad-hoc logic that is a real ergonomic loss versus a scripting language.

Not conceded, on three counts. First, spark-submit does the same thing — you build a jar or a wheel and submit a reference to it; the illusion of "submitting code" comes from Python being interpreted, and SDP's own answer (.py/.sql files on a path) is exactly a build-and- reference model. Second, the alternative that avoids the build step does not exist for us: SDP will not take Rust, and Arrow-FFI-into-JVM is a multi-month liability against a Spark line we pin only because Iceberg publishes a runtime for it. Third, jera's job model is better than what we would be trading for — durable redb rows, resource-floor placement, fleet- exclusive gating, cross-machine dispatch, working kill. Spark gives none of that outside a cluster manager.

And the ergonomic loss is temporary: once the daemon is detached and running, wasm plugin-load restores the one-motion loop for the cases that need it, without a JVM anywhere.

Net: we are not conceding the feature. We are getting the durable half now, from parts that already work, and the code-mobility half later from a host bifrost already has — instead of paying for a JVM bridge to reach a lifecycle model that is no better than jera's.

First concrete step

Add the Iceberg arm to bifrost run: --warehouse <dir> constructing IcebergSource::new(dir) in crates/knut-bifrost/src/bin/bifrost.rs alongside the existing FileSource (:230), feature-gated on skade. Listener is already generic over TableSource, so this is a source enum, not a rewrite. One caveat to carry into the design: the skade warehouse is redb-backed and single-open, and IcebergSource reopens it on every poll (source.rs:460) — a live external writer holding the catalog will block the daemon. That makes the Spark-free lane a hand-off lane (writer commits and closes, poller opens), which is exactly how tests/delta_iceberg_falkordb.rs:68 stages it, and it must be stated in the flag's help text.