Your browser sucks, JavaScript disabled! Gunnar's console needs JavaScript and WebAssembly. Git over SSH on port 2222 does not.

nordisk/skade

.nornir/streaming-composability-lineage-proposal.md

skade — Streaming Feedback, Zero-Copy Efficiency, a Composable Trait, and OpenLineage

Status: design proposal (no code changed). Author: research agent, 2026-06-17. Scope: skade-katalog (the embedded redb Iceberg catalog) + skade (the Arrow/Parquet data plane) + their integration with nornir-server's gRPC, knut's Flight SQL / Spark Connect, and OpenLineage→OpenMetadata. Decision owner: Rickard. Every claim below is grounded in file:line; the four workstreams are independent enough to land in any order, but a recommended sequencing is in the final section.

Consumer: the streaming-feedback workstream below is the client-side "signal" for korp's ingest profiles — see korp/.nornir/collectors.md §Deployment profiles (esp. P3 korp-landing, Signal vs poll). Spark Connect is pull-only / Structured Streaming is Spark-side polling; skade streaming-on-commit is how a Rust client reacts to new data without polling.

Executive summary

  1. Streaming feedback (replace polling). Every durable catalog mutation already funnels through one chokepoint — Store::group_commit (src/store.rs:233) and record_commit (src/store.rs:111), which advance a monotonic commit_seq. Add a tokio::sync::broadcast of a small CommitEvent { table, snapshot_id, commit_seq, ts } published after the redb commit lands and the L1 mirror is updated. Surface it over nornir-server as a new server-streaming RPC Viz.SubscribeCommits, modelled exactly on the existing Release.Progress stream (nornir-server.rs:917, proto nornir.proto:287). The nornir viz bench_live pane stops polling every 1.5 s (viz/bench_live.rs:46,268) and instead repaints when a bench_telemetry/bench_runs commit event arrives. Late subscribers get a resume cursor (current commit_seq snapshot, then deltas), so this degrades cleanly to polling when no subscriber is attached.

  2. Zero-copy / zero-allocation. The read path is already excellent — recast (skade/src/bridge.rs:101) is genuinely zero-copy on the matching-schema fast path, and L0/L1 hand back Arc<TableMetadata>/Arc<str> (src/meta_cache.rs:75, src/pointer_cache.rs:80). The real, worth-it wins are on the write/commit path: record_commit allocates a fresh key Vec per commit (src/store.rs:99), persist_commit does ~5 String clones per commit (src/catalog.rs:323-337), and read_delta collects whole files into Vec<u8> then re-parses (skade/src/read.rs:165,205). The biggest single read-path win is adopting knut's Arrow C Data Interface zero-copy bridge (graphar-flight/src/export.rs:340) as skade's first-class output so a scan can cross an Arrow-major boundary (57↔56) without a buffer copy. A concrete ranked list (with "not worth it" honesty) is in §2.

  3. A composable skade trait. Introduce a small trait set — IcebergSource (scan → Arrow stream + subscribe() to commit events from §1), IcebergSink (append/commit → returns the CommitEvent), and a Lineage observer (§4) — implemented once for skade::Table. The same traits serve viz, knut-bifrost (which already polls Iceberg snapshots and bridges Arrow zero-copy, knut-bifrost/src/source.rs:651), and a Flight SQL / Arrow Flight front so Spark 4.2 SDP + Iceberg 1.11.0 can read/write skade tables. Version reality: Spark SDP enforces Spark Connect and is Spark-4.1+ / 4.2-preview (knut already speaks it, knut-pipelines/src/client.rs:199); the JVM engine talks to skade as an Iceberg REST catalog (the bench already has a REST shim, bench/src/rest_shim.rs:33) over a shared object store — not via the Rust trait. The Rust trait is for the Rust consumers (viz, knut, Flight).

  4. OpenLineage → OpenMetadata. A skade commit is a natural OpenLineage RunEvent: table = dataset (namespace = "skade.<catalog>.<ns>", name = "<table>"), commit = run (runId = UUIDv7, COMPLETE on success), outputs carries schema + outputStatistics{rowCount} facets, and column-level lineage comes from the composable trait's transform metadata when present. Emit at the same hook as §1 (post-commit), POST JSON to an OpenLineage HTTP endpoint (Marquez or OpenMetadata's OpenLineage ingestion). Sketch payloads in §4.

Single unifying insight: all four workstreams hang off one hook — the post-commit point inside group_commit/atomic_release. Build that hook once (§1) and §3/§4 are mostly wiring.

Workstream 1 — Streaming feedback when data lands

1.1 Current state (file:line)

skade has exactly one durable-commit chokepoint, and it is already instrumented with a monotonic sequence number:

On the consumer side, the nornir viz polls:

And nornir-server already has the streaming primitive we need:

So the streaming alternative is idiomatic for this codebase — we are not inventing a transport, only a new event source and one new RPC.

1.2 Proposed design

(a) skade-katalog: a commit-event broadcast.

Add an optional broadcast sender to Store, published from the same post-commit moment that already updates the L1 mirror. Use tokio::sync::broadcast (not watch): a watch only keeps the latest value and would coalesce a burst of commits into one observed state, which loses per-snapshot deltas that the viz/CDC consumers need. broadcast is lossy under slow consumers (it drops oldest), which is the correct backpressure policy here — combined with the resume cursor below, a lagging subscriber recovers by re-reading from commit_seq.

// src/store.rs — new
/// A durable commit became visible. Cheap to clone (Arc-backed strings).
#[derive(Debug, Clone)]
pub struct CommitEvent {
    pub table_key: Arc<str>,      // "{catalog}\x1f{ns}\x1f{table}" (internal key)
    pub snapshot_id: Option<i64>, // None for snapshot-less commits (rare)
    pub metadata_location: Arc<str>,
    pub commit_seq: u64,          // durable monotonic cursor (META_COMMIT_SEQ)
    pub ts_micros: i64,           // event publish time (post-fsync)
}

pub(crate) struct Store {
    // ...existing fields...
    /// `None` until a subscriber attaches; lazily created so the default
    /// (no-subscriber) path pays nothing but one `Option` check.
    events: Arc<ArcSwapOption<broadcast::Sender<CommitEvent>>>,
}

record_commit already returns the new commit_seq (src/store.rs:130) — thread it out of the CommitFn so group_commit can stamp it onto the event. In group_commit, immediately after self.pointers.insert(...) (src/store.rs:268), if events is Some, let _ = tx.send(event); (a dropped event on no-receiver is fine). Do the same in create_table/register_table/atomic_release.

Public catalog API (on RedbCatalog, mirroring commit_seq):

impl RedbCatalog {
    /// Subscribe to commit events. The returned `commit_seq` is the cursor at
    /// subscribe time: events with `commit_seq > cursor` are guaranteed to flow
    /// through the receiver (subject to broadcast lag). A consumer that wants
    /// "current state then deltas" reads everything up to `cursor` once (e.g.
    /// resolve_many / a table scan), then drains the receiver.
    pub async fn subscribe_commits(&self) -> (u64, broadcast::Receiver<CommitEvent>) { ... }

    /// Replay durable commit events with `commit_seq` in `(after, now]` by
    /// scanning the immutable COMMITS log — the catch-up path for a late or
    /// lagged subscriber. (COMMITS is keyed by snapshot_id, so this adds a
    /// small `commit_seq → (table_key, snapshot_id)` secondary index, work
    /// item below, OR scans COMMITS and filters — acceptable for modest tails.)
    pub async fn commits_since(&self, after: u64) -> Result<Vec<CommitEvent>> { ... }
}

Durability / replay. The broadcast channel is in-memory and lossy by design. Durability is already provided by the COMMITS table + META_COMMIT_SEQ (src/store.rs:77,82). The contract is: the broadcast is a low-latency hint; the durable log is the source of truth. A subscriber that receives RecvError::Lagged(n) calls commits_since(last_seen_seq) to backfill, then resumes the live stream. This is the standard "snapshot + tail" pattern and it is why we expose the cursor from subscribe_commits.

Replay needs a commit_seq-ordered read. COMMITS is keyed by (table_key, snapshot_id_be) (src/store.rs:99), not by commit_seq, so commits_since can't range-scan by sequence directly. Two options: (1) add a second redb table commit_log: u64_be(commit_seq) → (table_key, snapshot_id) appended in record_commit (one extra insert per commit, ~no fsync cost since it's in the same txn), or (2) for small tails, scan COMMITS and filter on a per-entry seq if we also store it. Recommend (1) — it's the clean cursor and it's the same shape the Ragnar StaticIndex could later accelerate.

Backpressure & multi-writer. Multi-writer is already serialized by redb's file lock + group_commit coalescing, so events are naturally totally-ordered by commit_seq. Backpressure: broadcast drops oldest for slow consumers (bounded memory), and the cursor recovers correctness. No writer ever blocks on a slow subscriber — send on a broadcast never awaits.

Degrade to polling. If no one ever calls subscribe_commits, events stays None and the only cost is a relaxed Option load per commit. Consumers that prefer polling keep calling commit_seq() (already exists) as a cheap change detector — the design is strictly additive.

(b) nornir-server: a server-streaming RPC.

Add to nornir.proto, modelled on Release.Progress:

service Viz {
  // ...existing unary RPCs...
  // Server-streaming: push one CommitEvent per table commit landing in the
  // workspace's warehouse. Replaces bench_live's 1.5s poll.
  rpc SubscribeCommits (SubscribeCommitsRequest) returns (stream CommitEvent);
}
message SubscribeCommitsRequest {
  string workspace = 1;
  uint64 after_commit_seq = 2; // 0 = from-now; >0 = replay then tail
  repeated string tables = 3;  // optional filter (empty = all)
}
message CommitEvent {
  string table = 1;            // "ns.table" (de-keyed for the wire)
  int64  snapshot_id = 2;
  uint64 commit_seq = 3;
  int64  ts_micros = 4;
  int64  row_delta = 5;        // see "row delta" note below
}

Handler shape is a copy of nornir-server.rs:917-943: resolve workspace from metadata, call catalog.subscribe_commits() (+ commits_since if after > 0), adapt the broadcast::Receiver into a ReceiverStream/BroadcastStream, map CommitEvent → proto CommitEvent, box it. Auth/interceptor unchanged.

Row delta. The catalog event has snapshot_id but not row count. Two ways to fill row_delta: cheap (read the snapshot summary's added-records from the new metadata, which skade already has parsed in L0) — recommended; or precise (sum record_count of the data files in the snapshot, which read_delta already computes, skade/src/read.rs:204). Use the summary; it's free.

(c) the viz use case, closed.

bench_live (viz/bench_live.rs) replaces its timer with: on pane open, call Viz.SubscribeCommits{ tables: ["bench_telemetry","bench_runs"], after: 0 }; on each event, request_repaint() and re-read just those tables (or, later, carry the delta rows in the event). Mid-run telemetry rows light up the UI within broadcast-latency (sub-ms locally) instead of up to 1.5 s. The same RPC serves the test pane (test_results table). This is the literal use case the prompt names.

1.3 Effort / risk

Workstream 2 — Zero-copy & zero-allocation efficiency review

I read every hot path. skade is already well-tuned; I rank opportunities by real payoff and call out the ones that are not worth it.

2.1 Read path — current state

2.2 Read path — the one real win: a zero-copy Arrow output bridge

read_all/read_delta return Vec<RecordBatch> in skade's Arrow-57 world. The moment a consumer in a different Arrow major needs them — and knut is pinned to Arrow 56 (knut/Cargo.toml:25-28) — a naive bridge copies every buffer. knut solved this with the Arrow C Data Interface zero-copy bridge:

Proposal: make this bridge a first-class, owned part of skade rather than something each consumer re-implements against knut's copy. Add an opt-in arrow-bridge feature to the skade crate exposing to_ffi_batch(&RecordBatch) -> (FFI_ArrowArray, FFI_ArrowSchema) per column (or a RecordBatch-level helper), documented as "cross an Arrow-major boundary zero-copy; transmute on the consumer side per the C ABI." This is the canonical model the prompt asked for, and it makes the §3 Flight path zero-copy end-to-end. Effort: ~0.5 day (lift + test); risk: low but the unsafe transmute must stay gated behind the documented ABI invariant (the SAFETY comment at export.rs:342-346 is the contract to copy).

2.3 Write / commit path — the allocation hotspots (ranked)

  1. persist_commit string churn (highest-frequency, per-commit). src/catalog.rs:323-337 clones the staged location and ident repeatedly: staged_metadata_location is .to_string()'d (:323), then staged_loc = staged_metadata_location.clone() (:337), conflict_ident = ident.clone() (:336), plus base_metadata_location: String flows in. For a high-rate ingest committing every few files (ingest_pipelined, skade/src/write.rs:461), this is a handful of String allocs per commit. Proposed change: move staged_metadata_location/base into the closure by value (they're not needed after), and key the L1 mirror + commit on Arc<str> so the post-commit pointers.insert (src/store.rs:267) doesn't re-Arc::from a fresh &str (src/pointer_cache.rs:94 allocates Arc::from(loc) every insert). Threading an Arc<str> from persist_commit through CommitOutcome to the mirror removes one alloc per commit. Effort: ~0.5 day; payoff: real under high commit rate.

  2. commit_key per-commit Vec (per-commit). src/store.rs:99-105 allocates a fresh Vec<u8> (with_capacity, then two extends) on every record_commit. It's small and short-lived. Proposed change (only if profiling shows it): a SmallVec<[u8; 64]> or a reused thread-local scratch buffer, since group_commit runs commits sequentially inside one txn. Honest verdict: borderline. The redb insert that follows dominates; do this only if a commit-rate bench flags it. List it, don't rush it.

  3. record_commit double-get on META (per-commit). src/store.rs:121-129 opens META and does two get + two insert. Fine — redb in-txn ops are cheap and it's correct. Not worth changing.

  4. read_delta whole-file buffering (per-delta-read, potentially large). skade/src/read.rs:165 (file_io...read().await → whole manifest into a Bytes/Vec) and :205 (whole data file into memory, then ParquetRecordBatchReaderBuilder::try_new(bytes)). For large data files this materializes the entire file before decode. Proposed change: stream via the async Parquet reader (ParquetRecordBatchStreamBuilder over the FileIO input's async read) so only row-group-sized buffers are live. This also aligns read_delta with the windowed reader skade already has (parquet_io.rs:read_row_groups). Effort: ~0.5 day; payoff: memory, not CPU — matters for big CDC deltas, neutral for small ones.

  5. ingest_parallel holds all encoded files in memory (by design). skade/src/write.rs:315-323 collects every encoded Parquet file before writing. This is documented as the trade vs ingest_pipelined (write.rs:388), which already bounds memory via a channel. Not a bug — already has the zero-barrier GATLING alternative. Leave both; the law about GATLING save-stage is satisfied by ingest_pipelined's single sequential writer (write.rs:435-465).

2.4 Not worth it (be honest)

2.5 Summary table

# Path File:line Change Payoff Worth it?
1 commit catalog.rs:323-337, pointer_cache.rs:94 thread Arc<str> loc through CommitOutcome, drop per-commit String/Arc::from per-commit allocs Yes
2 read→export skade (new) ← export.rs:340 first-class Arrow C-Data-Interface bridge (feature arrow-bridge) zero-copy cross-major output Yes
3 CDC read read.rs:165,205 async Parquet stream instead of whole-file Vec<u8> bounded memory on big deltas Yes (memory)
4 commit store.rs:99 SmallVec/scratch for commit_key tiny Only if bench flags
5 read meta_cache.rs:83, static_index.rs:36 — negligible No

Workstream 3 — A composable skade trait (Spark 4.2 + Iceberg 1.11.0 via SDP + structured streaming)

3.1 Current state & version reality (file:line)

The reality gap, stated plainly: Spark 4.2 (JVM) does not link a Rust trait. SDP + structured streaming reach skade through (i) an Iceberg REST catalog skade serves (the bench rest_shim.rs:33 is the seed; Iceberg 1.11's server-side scan planning makes this even more attractive) backed by a shared object store (skade already supports S3/MinIO, skade/src/warehouse.rs:82, skade/Cargo.toml:44), and optionally (ii) Arrow Flight SQL for live push/pull. The Rust trait below is for the Rust consumers (viz, knut-bifrost, a Flight server); Spark interop is a deployment (REST + S3), not a trait impl.

3.2 Proposed trait set

Keep it small, async, and composable. Put it in skade (the data plane), not the catalog crate, so it can name Arrow/DeltaPlan types.

/// A readable Iceberg dataset: scan to Arrow, incremental delta, and a live
/// subscription to commits (from Workstream 1). Object-safe via boxed streams.
#[async_trait]
pub trait IcebergSource: Send + Sync {
    fn ident(&self) -> &TableIdent;
    fn arrow_schema(&self) -> Result<ArrowSchemaRef>;
    fn current_snapshot_id(&self) -> Option<i64>;

    /// Full or projected scan as an Arrow batch stream (back-pressured).
    async fn scan(&self, projection: &[&str])
        -> Result<BoxStream<'static, Result<RecordBatch>>>;

    /// Incremental: only rows added between `from` and `to` (+ the plan).
    async fn scan_delta(&self, from: Option<i64>, to: i64)
        -> Result<(BoxStream<'static, Result<RecordBatch>>, DeltaPlan)>;

    /// Live commit subscription for this table (Workstream 1). Yields the
    /// snapshot id + cursor each time this table advances; the consumer then
    /// calls `scan_delta(last, snapshot_id)`. `None` if the backend can't push.
    async fn subscribe(&self) -> Option<BoxStream<'static, CommitEvent>>;
}

/// A writable Iceberg dataset: append/ingest Arrow, returning the commit event
/// so a caller can chain lineage (Workstream 4) or wake a downstream source.
#[async_trait]
pub trait IcebergSink: Send + Sync {
    fn ident(&self) -> &TableIdent;

    /// Append one snapshot; returns the CommitEvent it produced.
    async fn append(&mut self, batches: &[RecordBatch]) -> Result<CommitEvent>;

    /// Bulk ingest (the GATLING pipeline); returns stats + the final event.
    async fn ingest(&mut self, groups: Vec<Vec<RecordBatch>>, files_per_commit: usize)
        -> Result<(IngestStats, CommitEvent)>;

    /// CDC equality-delete (skade already has this: delete.rs).
    async fn delete_equality(&mut self, keys: &[RecordBatch], equality_cols: &[&str])
        -> Result<CommitEvent>;
}

/// Optional: a transform that carries column-lineage metadata, so a
/// source→sink hop can emit column-level OpenLineage (Workstream 4).
pub trait Transform: Send + Sync {
    fn apply(&self, batch: &RecordBatch) -> Result<RecordBatch>;
    /// output col name -> input col names it derives from (None = opaque).
    fn column_lineage(&self) -> Option<HashMap<String, Vec<String>>>;
}

impl IcebergSource + IcebergSink for skade::Table is a thin wrapper over the existing methods (table.rs:80-204) — scan wraps table.scan().to_arrow() (read.rs:22), append wraps append_with and reads back the new snapshot id for the CommitEvent. This is mechanical discoverability in the spirit of the project's Facet law: every readable/writable surface implements the trait, so a test matrix and a registry can enumerate them.

3.3 How it plugs into SDP + Spark Connect + Flight (and knut)

3.4 Effort / risk

Workstream 4 — OpenLineage → OpenMetadata

4.1 Grounding (web-confirmed spec)

A RunEvent (OpenLineage OpenLineage.json): top-level eventType (START|RUNNING|COMPLETE|ABORT|FAIL|OTHER), eventTime, run { runId: UUID(v7), facets }, job { namespace, name, facets }, inputs[], outputs[] (each a dataset { namespace, name, facets }), producer, schemaURL. Standard facets: schema (SchemaDatasetFacet: column names+types), columnLineage (ColumnLineageDatasetFacet: per-output-field inputFields[]), dataSource, output outputStatistics{rowCount,size}, run nominalTime/parent. Sending at least a START and a terminal (COMPLETE/FAIL) event is required. OpenMetadata, Marquez, DataHub all ingest these via HTTP or Kafka.

4.2 skade → OpenLineage mapping

OpenLineage skade concept Source (file:line)
Dataset namespace "skade://<warehouse_uri>" or "skade.<catalog>" warehouse.rs:57 / catalog.rs:51
Dataset name "<ns>.<table>" TableIdent, catalog.rs
Dataset schema facet table current schema (Arrow/Iceberg) read.rs:15 arrow_schema_of
RunEvent (COMPLETE) a successful commit store.rs:263 post-write.commit()
run.runId UUIDv7 per commit (one append = one run) new
output outputStatistics.rowCount snapshot added-records (or sum data files) snapshot summary / read.rs:204
columnLineage facet Transform::column_lineage() when present §3.2 trait
inputs[] upstream tables in a transform/atomic_release atomic.rs:46 (multi-table)

A commit = a RunEvent. For a plain append, inputs are empty and the output is the appended table (an ingest from outside the lake). For a transform hop (source table → Transform → sink table) inputs = the source dataset(s), outputs = the sink, and columnLineage is filled from Transform::column_lineage(). For atomic_release (multi-table release, atomic.rs:46), emit one run with all N tables in outputs[] — the atomic batch is one logical job run, which is semantically perfect for OpenLineage.

4.3 Emit point & transport

Hook: the exact §1 post-commit point. A Lineage sink trait (mirroring knut's PipelineEventSink, knut-pipelines/src/viewer.rs) subscribes to CommitEvents (it's just another consumer of the broadcast — no extra coupling) and POSTs JSON.

#[async_trait]
pub trait LineageSink: Send + Sync {
    async fn emit(&self, event: &RunEvent) -> Result<()>;
}
/// HTTP transport to Marquez / OpenMetadata OpenLineage endpoint.
pub struct HttpLineageSink { endpoint: String, http: reqwest::Client }

Wire it as a background task: let (cursor, mut rx) = catalog.subscribe_commits(); then for each event build a RunEvent and sink.emit(). Because it rides the broadcast, lineage emission never blocks the writer and a lagged sink backfills via commits_since (same durability story as §1). Transport options: direct HTTP to OpenMetadata's OpenLineage ingestion endpoint, or to Marquez (the reference store) which OpenMetadata can also read. Producer string: "skade/<version>".

4.4 Example payload (a single-table append, COMPLETE)

{
  "eventType": "COMPLETE",
  "eventTime": "2026-06-17T12:00:00.123Z",
  "producer": "https://codeberg.org/nordisk/skade/0.4.10",
  "schemaURL": "https://openlineage.io/spec/1-0-5/OpenLineage.json",
  "run": {
    "runId": "018f...-7b2c",
    "facets": { "nominalTime": { "nominalStartTime": "2026-06-17T12:00:00Z" } }
  },
  "job": { "namespace": "skade", "name": "append:main.bench_telemetry" },
  "inputs": [],
  "outputs": [{
    "namespace": "skade://file:///var/lib/nornir/warehouse",
    "name": "main.bench_telemetry",
    "facets": {
      "schema": { "fields": [
        { "name": "run_id", "type": "long" },
        { "name": "repo",   "type": "string" },
        { "name": "cpu_pct","type": "double" }
      ]},
      "outputStatistics": { "rowCount": 512 }
    }
  }]
}

For a transform hop, inputs[] names the source dataset(s) and the output gains a columnLineage facet: { "fields": { "cpu_pct": { "inputFields": [ { "namespace": "...", "name": "main.raw_samples", "field": "cpu_ns" } ] } } }.

4.5 Effort / risk

Consolidated roadmap

graph TD
  P1["Phase 1 — Commit-event hook (the spine)"] --> P1a["broadcast CommitEvent + commit_log cursor table (skade-katalog)"]
  P1a --> P1b["subscribe_commits / commits_since API"]
  P1b --> P2["Phase 2 — Streaming RPC (viz)"]
  P2 --> P2a["Viz.SubscribeCommits server-stream (model: Release.Progress)"]
  P2a --> P2b["bench_live: drop 1.5s poll, subscribe"]
  P1b --> P4["Phase 4 — OpenLineage (rides the same broadcast)"]
  P4 --> P4a["LineageSink + HttpLineageSink + RunEvent builder"]

  Z["Phase 0 — Zero-copy quick wins (parallel, no deps)"] --> Z1["Arc<str> through CommitOutcome (catalog.rs:323)"]
  Z --> Z2["first-class Arrow C-Data bridge (feature arrow-bridge)"]
  Z --> Z3["read_delta async Parquet stream"]

  P1b --> P3["Phase 3 — Composable trait + transports"]
  Z2 --> P3
  P3 --> P3a["IcebergSource/IcebergSink/Transform + impl for Table"]
  P3a --> P3b["skade-flight (Flight SQL front; knut precedent)"]
  P3a --> P3c["skade-rest (productionize rest_shim; Iceberg 1.11 server-side planning)"]
  P3b --> P3d["Spark 4.2 SDP/structured-streaming via REST+S3, capture via Flight"]

Recommended order:

  1. Phase 0 (zero-copy, parallel, no dependencies): the Arc<str> commit win and the first-class Arrow C-Data bridge. Independent, low-risk, and the bridge unblocks Phase 3's Flight path.
  2. Phase 1 (the spine): the commit-event broadcast + cursor. Everything else hangs off this. Smallest change with the widest leverage.
  3. Phase 2 (viz): the streaming RPC + bench_live rewire. Highest visible payoff, smallest incremental effort once Phase 1 exists. This is the use case the prompt centers on.
  4. Phase 4 (lineage): add the OpenLineage observer — it's just another subscriber to Phase 1's broadcast, so it's cheap once the spine is in.
  5. Phase 3 (composability): the trait set, then the Flight and REST fronts. Largest surface, most external-version risk; do it last and let it consume the bridge (Phase 0) and the subscription (Phase 1).

Estimated total: Phase 0 ~1d, Phase 1 ~1.5d, Phase 2 ~1d, Phase 4 ~1.5d, Phase 3 ~5-7d. The first four phases (~5 days) deliver streaming + lineage + the zero-copy bridge; Phase 3 is the bigger, version-sensitive interop investment.

Cross-cutting LAWs to honor (from project memory):

Open questions for the owner

  1. iceberg-rust version truth. Is iceberg = "0.9.1" (skade/Cargo.toml:23) a real upstream tag, a local fork, or a pre-release? Public latest is 0.8.0 (2026-02). The Flight/REST write surface in Phase 3 depends on which write TransactionActions exist. Please confirm the source of the 0.9.1 pin.
  2. broadcast vs. an explicit ring buffer. tokio::broadcast is lossy under lag (recovered by the cursor). Acceptable, or do you want a bounded durable ring in redb that guarantees no consumer ever needs the catch-up scan? (More work, more I/O; I recommend broadcast + cursor.)
  3. Event granularity for the wire. Should Viz.SubscribeCommits carry the delta rows themselves (server reads read_delta and ships batches) or just the snapshot id (client pulls)? Carrying rows is fewer round-trips but couples the stream to a row format; I lean id-only for the viz, rows for the Flight sink. Your call on the viz.
  4. REST vs Flight as the Spark on-ramp. For Spark 4.2 SDP, do you want skade to serve an Iceberg REST catalog (Spark's native, supports server-side scan planning in 1.11) or push Spark toward Arrow Flight for live capture? They're complementary; which do we build first?
  5. Arrow major. skade is Arrow-57, knut is Arrow-56. The C-Data bridge handles the hop, but should skade also offer a native Arrow-56 build (feature flag) to avoid the bridge in pure-knut deployments, or is the zero-copy bridge always the answer?
  6. Lineage namespace scheme. Is "skade://<warehouse_uri>" the right dataset namespace for OpenMetadata's taste, or do you want it to match an existing OpenMetadata service name so datasets land under the right service node?

Appendix — key file:line index used in this proposal