<!-- nornir:class: reasoning -->
# 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:
- **`Store::group_commit`** (`src/store.rs:233-308`) coalesces concurrent table-pointer
advances into one redb write txn + one fsync. After `write.commit()` succeeds
(`src/store.rs:263`) it publishes each op's mirror update to the lock-free L1
pointer cache (`src/store.rs:267-272`) and wakes the caller's `oneshot`
(`src/store.rs:274`). **This is the exact point a commit becomes visible.**
- **`record_commit`** (`src/store.rs:111-131`) runs *inside* that txn, appends the
immutable `(table_key, snapshot_id) → metadata_location` row to `COMMITS`
(`src/store.rs:118`), and bumps `META_COMMIT_SEQ` (`src/store.rs:122`) — a
durable, monotonic, never-reset commit counter. `commit_seq()` reads it back
(`src/catalog.rs:79`).
- The two non-`group_commit` durable paths that also need to fire events:
`create_table` (`src/catalog.rs:767-779`), `register_table`
(`src/catalog.rs:916-927`), `atomic_release` / `atomic_release_raw`
(`src/atomic.rs:121-139`, `src/atomic.rs:259-274`). All call `record_commit` and
then the L1 mirror write-through, so they share the same "event-worthy" moment.
- **`persist_commit`** (`src/catalog.rs:317-369`) is what `commit_table` and
`update_table` route through, and it ends in `group_commit` — so wiring the event
into `group_commit` covers single-table commits for free.
On the **consumer** side, the nornir viz polls:
- `viz/bench_live.rs:46` — `const RELOAD_EVERY: Duration = Duration::from_millis(1500);`
- `viz/bench_live.rs:268-276` — the draw loop re-reads on a timer: local mode reads
the `bench_telemetry` + `bench_runs` warehouse tables, remote mode calls the
unary `Viz.BenchTelemetry` RPC (`nornir.proto:799-846`, all unary today), then
`request_repaint_after(400ms)`.
And nornir-server already has the streaming primitive we need:
- **`Release.Progress`** server-streaming RPC — handler `nornir-server.rs:917-943`,
proto `nornir.proto:287` (`rpc Progress (Empty) returns (stream ProgressEvent);`).
It builds a `tokio::sync::mpsc::channel(64)`, spawns a tailer, and returns
`ReceiverStream::new(rx)` boxed as the response stream. Auth is a bearer-token
interceptor (`nornir-server.rs:2916`); workspace routing is the `nornir-workspace`
metadata header.
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`.
```rust
// 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`):
```rust
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`:
```proto
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
- **skade-katalog broadcast + cursor:** ~1 day. Low risk (additive, lossy channel,
no hot-path cost when unsubscribed). The `commit_log` secondary table is the only
on-disk change; it's append-only in the existing txn.
- **nornir-server RPC + viz rewire:** ~1 day. Low risk; `Release.Progress` is the
proven template.
- **Tests (LAW: inject + assert real values):** a test that commits N tables, has a
subscriber, and asserts it observes N events in `commit_seq` order with the right
`snapshot_id`s; a lag/replay test that drops the receiver, commits, then
`commits_since` returns the missed events. These go in `tests/commit_log.rs` /
a new `tests/events.rs`, and feed `functional_status` rows under
`--features testmatrix` (matching `src/store.rs:434`).
---
## 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
- **L0 metadata cache** hands back `Arc<TableMetadata>` on a hit
(`src/meta_cache.rs:75`) — zero-copy, ~100-300 ns, single-flight on miss. The
only allocs are on a *miss* (`fileio.clone()` + `loc.to_string()`,
`src/meta_cache.rs:83-84`), explicitly justified as dwarfed by the I/O. **Leave it.**
- **L1 pointer mirror** returns `Arc<str>` clones (`src/pointer_cache.rs:53,80`) and
uses `imbl::HashMap` O(1) structural-sharing clones on write
(`src/pointer_cache.rs:111`). **Leave it.**
- **Ragnar `StaticIndex`** stores `Arc<[Entry]>` / `Arc<[i64]>`
(`src/static_index.rs:49-50`); lookups return refs. But `resolve_location_at`
and `resolve_many` then `.to_string()` the `metadata_location`
(`src/catalog.rs:157,230,253`). That `String` is then handed to the L0 cache
which keys by `String` anyway, so the alloc is unavoidable *given moka's
owned-key API*. **Marginal — skip.**
- **`recast`** (`skade/src/bridge.rs:101-110`) is the model citizen: it
`Arc::clone`s columns whose type already matches the target and only invokes
`arrow_cast::cast` on genuine type changes. On the common path (caller built the
batch against the table's field-id schema) it is **fully zero-copy** — it
re-stamps the schema over the same buffers. **Exemplary; leave it.**
### 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:
- `graphar-flight/src/export.rs:329-364` — `bridge_56_to_57(batch) -> RecordBatch`:
`arrow_array::ffi::to_ffi(&col.to_data())` exports to the stable `#[repr(C)]`
`FFI_ArrowArray`/`FFI_ArrowSchema`, then `std::mem::transmute` to the *other*
arrow major's identically-laid-out FFI structs (the C ABI is fixed across
versions), then `from_ffi` imports — **buffer pointers transferred, zero copy**,
with the `release` callback travelling along so memory frees exactly once.
- The reverse `batch_57_to_56` lives in `knut-bifrost/src/source.rs:728-757` and is
already how knut ingests skade Iceberg reads into the graph
(`knut-bifrost/src/source.rs:651,685`).
**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)
- Eliminating the L0 miss-path clones (`meta_cache.rs:83`) — dwarfed by I/O.
- `StaticIndex` `Box<str>` → `Arc<str>` (`static_index.rs:36-37`) — built once per
compaction, read-only after; the `Box<str>` is fine.
- Trying to avoid `Table::build()`'s per-call `ObjectCache` alloc — it's
iceberg-rust-internal (`plan.md:105`), not skade's to fix; skade already routes
metadata-only callers around it via `resolve_metadata` (`src/catalog.rs:99`).
### 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)
- skade is **already** a source/sink, just without a unifying trait: read =
`Table::read`/`read_columns`/`read_delta` (`skade/src/table.rs:156-182`),
write = `Table::append`/`ingest*` (`skade/src/table.rs:80-153`), incremental =
`read_delta` + `DeltaPlan` (`skade/src/read.rs:105`). The catalog is a real
`iceberg::Catalog` (`src/catalog.rs:408`) plus the multi-table `atomic_release`
(`src/atomic.rs:46`) and a raw `commit_table` for REST callers
(`src/catalog.rs:272`).
- **Versions skade targets today:** `iceberg = "0.9.1"`, `arrow-* = "57"`,
`parquet = "57"`, `datafusion = "52.2"` (`skade/Cargo.toml:23-29,60`), tables
written as **FormatVersion::V3** (`skade/src/warehouse.rs:19`).
- **Versions knut targets today:** Arrow/Parquet **56** (`knut/Cargo.toml:25-28`),
Spark **4.2.0-preview5** + Iceberg-Java **1.11.0** + Iceberg-Spark runtime **4.1**
(`knut/containers/spark-iceberg/Containerfile:16-19`), Spark Connect via vendored
protos + tonic 0.14 (`knut-pipelines/build.rs:56-61`, `knut-pipelines/Cargo.toml:51`).
- **Web-confirmed version facts (2026):**
- **Iceberg 1.11.0** released **2026-05-19**; stabilizes V3 (deletion vectors via
Roaring bitmaps, Variant, native geospatial, ns timestamps), adds **server-side
scan planning** in the REST catalog, defaults to **Spark 4.1 / Flink 2.1**.
- **Spark 4.2** is in **preview4/preview5**; **SDP requires Spark Connect** ("API
mode enforced to Spark Connect, requires a remote origin"). SDP datasets/flows
are the declarative graph Spark executes.
- **iceberg-rust** public latest is **0.8.0** (2026-02); schema-evolution
`TransactionAction`s only partially exposed. skade's `"0.9.1"` pin is **ahead of
the public crate** — it resolves a local/sibling build (the workspace
`[patch.crates-io]` for iceberg-datafusion at `skade/Cargo.toml:103` confirms
local patching is in play). **Flag:** confirm whether iceberg-rust 0.9.1 is a
real upstream tag or a local fork; the trait must not assume APIs the public
crate lacks.
**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.
```rust
/// 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)
- **knut-bifrost** already consumes exactly this shape: `TableSource::fetch_new()
-> Vec<RecordBatch>` (`knut-bifrost/src/source.rs:18-25`) polls Iceberg snapshots
and bridges Arrow zero-copy (`source.rs:651,685,728`). Re-express
`IcebergSource::subscribe()` so bifrost replaces its poll loop with the §1
push — same trait, now event-driven.
- **Flight SQL / Arrow Flight front (new, optional crate `skade-flight`).** knut's
`CypherFlightService` (`graphar-flight/src/service.rs:113`, server
`server.rs:186`) is the precedent: a tonic Flight service that streams
`RecordBatch`es. A `skade-flight` service implements `do_get` = `IcebergSource::scan`
/ `scan_delta`, and `do_put` = `IcebergSink::append` (knut already does
`do_put_statement_update` graph-ingest, `service.rs:376`). Because skade owns the
§2.2 C-Data-Interface bridge, batches cross to Arrow-56 consumers zero-copy and to
Arrow-Flight IPC without an extra copy. **This is the composable hub: one trait,
three transports (in-process Rust, Flight, REST).**
- **Spark 4.2 SDP / structured streaming.** SDP defines `Dataset`/`Flow` graphs
(knut models them: `DataflowGraph`/`Dataset`/`Flow`, `knut-pipelines/src/model.rs:88-200`)
that Spark executes via Spark Connect (`knut-pipelines/src/client.rs:199-222`).
Spark reads/writes Iceberg through its **Iceberg-Spark 4.1 runtime** against a
**catalog**. skade participates by **serving an Iceberg REST catalog** (productionize
`bench/src/rest_shim.rs` into a `skade-rest` crate; Iceberg 1.11 server-side scan
planning is a natural extension point) over **shared S3/MinIO**
(`skade/src/warehouse.rs:82`). A Spark SDP `Dataset { format: "iceberg" }` that
reads/writes a skade table then works unchanged — and structured streaming's
Iceberg source consumes the same snapshots skade's `read_delta` already produces.
knut-pipelines, which today *drops* result batches (`knut-pipelines/src/client.rs:473`),
could instead point a Flow's output at a skade Flight sink for live capture.
### 3.4 Effort / risk
- **Trait + `impl for Table`:** ~1 day, low risk (wraps existing methods).
- **`skade-flight` crate:** ~2-3 days, medium risk (Flight protocol surface; but
knut's service is a close template).
- **`skade-rest` (productionized shim):** ~2-3 days, medium risk (Iceberg REST
conformance for the endpoints Spark exercises; shim already covers config/ns/
table CRUD + commit, `rest_shim.rs:35-46`). Server-side scan planning is a
follow-on.
- **Version risk (the real one):** confirm the iceberg-rust 0.9.1 pin vs public
0.8.0; confirm Spark 4.2 GA timing (preview today). The trait itself is
version-agnostic; the Flight/REST fronts must track Iceberg-rust's evolving write
API (deletion vectors, schema evolution actions land across 0.9→0.10).
---
## 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 `CommitEvent`s
(it's just another consumer of the broadcast — no extra coupling) and POSTs JSON.
```rust
#[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)
```json
{
"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
- **`LineageSink` + `HttpLineageSink` + RunEvent builder:** ~1.5 days. Low risk
(additive observer of §1; no hot-path coupling). Needs `reqwest` (optional
feature `lineage`) — keep it off the default build like `sql`.
- **Column lineage:** only as good as `Transform::column_lineage()`; for plain
appends/ingests it's correctly omitted (spec allows). Marquez/OpenMetadata both
accept dataset-level-only events.
- **Tests:** commit a table with a subscribed `LineageSink` capturing to a `Vec`,
assert a START+COMPLETE pair with the right dataset name and `rowCount`
(inject-and-assert per the LAW); a transform test asserting the `columnLineage`
facet.
---
## Consolidated roadmap
```mermaid
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):**
- Every change adds a real inject-and-assert `#[test]` to the matrix
(`tests/*.rs` + `--features testmatrix` `functional_status` rows, cf.
`src/store.rs:434`).
- Each feature ships `.nornir` design + guide docs (this doc is the design; guides
follow per phase).
- Telemetry/events persist to the warehouse, not `/tmp`; the durable `COMMITS` +
`commit_log` tables already satisfy this — the broadcast is the *hint*, the log
is the *truth*.
- GATLING zero-copy on produce→save is already met by `ingest_pipelined`
(`write.rs:435`); the Arrow bridge extends zero-copy to the *output* edge.
---
## 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
`TransactionAction`s 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
- Commit chokepoint: `src/store.rs:233` (`group_commit`), `:111` (`record_commit`),
`:99` (`commit_key`), `:77/:82` (`COMMITS`/`META_COMMIT_SEQ`).
- Commit API: `src/catalog.rs:79` (`commit_seq`), `:272` (`commit_table`),
`:317` (`persist_commit`), `:767`/`:916` (create/register), `src/atomic.rs:46`/`:157`.
- Read accel: `src/meta_cache.rs:75` (L0), `src/pointer_cache.rs:80` (L1),
`src/static_index.rs:42` (Ragnar), `.nornir/plan.md:13-117`.
- Data plane: `skade/src/read.rs:22/105/165/205`, `skade/src/write.rs:158/272/388`,
`skade/src/bridge.rs:101` (zero-copy recast), `skade/src/table.rs:80-204`,
`skade/src/warehouse.rs:19/57/82`, `skade/Cargo.toml:23-29/60/103`.
- REST shim: `bench/src/rest_shim.rs:33-46/207`.
- nornir streaming precedent: `nornir-server.rs:917` (`Release.Progress`), `:2916`
(auth), `proto/nornir.proto:287`; viz poll `viz/bench_live.rs:46/268`.
- knut zero-copy bridge: `graphar-flight/src/export.rs:329-364` (`bridge_56_to_57`),
`knut-bifrost/src/source.rs:651/685/728`; Flight `graphar-flight/src/service.rs:113`,
`server.rs:186`; knut Arrow 56 `knut/Cargo.toml:25-28`.
- knut-pipelines (SDP/Spark Connect): `client.rs:199/367/473`, `model.rs:88-200`,
`progress.rs:21-208`, `build.rs:56-61`, `containers/spark-iceberg/Containerfile:16-19`.
```