# CDC · streaming · lineage — design
> **Class:** `design` · **Function:** `cdc-stream` · **Uses:** `store`,
> `commit`, `read-path`, `catalog`. Only the shape that runs; the *why* +
> the ecosystem survey (Fluss/Paimon/Hudi/Iceberg-CDC) is in
> [`cdc-streaming-lineage.md`](cdc-streaming-lineage.md) and
> [`streaming-composability-lineage-proposal.md`](streaming-composability-lineage-proposal.md);
> the underlying `read_delta` diff is in [`incremental-read.md`](incremental-read.md).
skade turns one durable commit chokepoint into a **change-data-capture stream, a
push consume API, and a lineage feed** — four thin layers stacked on the redb
catalog, each additive and independently feature-gated. Nothing here is a new
scan path: it is re-tagging and broadcasting over the primitives that already
exist (`group_commit`, `record_commit`, `read_delta`).
```
catalog commit (group_commit) src/store.rs
│ one redb txn, one fsync
├─▶ TABLES pointer advance + L1 mirror (source of truth)
├─▶ COMMIT_LOG row (durable, commit_seq-keyed) ← durable replay
└─▶ CommitEvent ─broadcast──────────────┐ ← sub-ms hint
▼
Phase 1 read_changelog(from,to) ── _change_type-tagged windows skade/src/read.rs
▼
Phase 2 ChangeStream::changes(from) ── live BoxStream, exactly-once cursor skade/src/stream.rs
▼
Phase 3 LineageSink ── WarehouseLineageSink / HttpLineageSink skade/src/lineage.rs
Phase 4 Table::lookup(key, as_of) ── embedded delta-join probe skade/src/read.rs
```
## f0 — why this is cheap
Every catalog mutation already funnels through **one** durable chokepoint with a
monotonic, gapless cursor: `group_commit` (`store.rs:371`) coalesces concurrent
writes into a single redb transaction, and `record_commit` (`store.rs:211`)
bumps `META_COMMIT_SEQ` (`store.rs:169`) and appends the immutable `COMMITS` log
inside that same txn. The streaming spine is two additions to that chokepoint —
a durable seq-keyed index and an in-memory broadcast — so it costs **no extra
fsync** and the default (poll-only) build pays nothing.
## f1 — Phase 0: the commit-broadcast spine (`src/store.rs`, `src/catalog.rs`)
The keystone every later layer subscribes to. A commit publishes a
[`CommitEvent`] (`store.rs:39`) *after* the redb txn lands and the L1 mirror is
updated:
| field | meaning |
|---|---|
| `table_key` | internal `"{catalog}\x1f{ns}\x1f{table}"` (compute via `commit_table_key`) |
| `snapshot_id` | `Some` for a data commit, `None` for a create / metadata-only pointer move |
| `metadata_location` | where the pointer now points |
| `commit_seq` | durable, gapless, monotonic cursor — the exactly-once resume key |
| `ts_micros` | event time, stamped in the txn |
Two durable/transient halves, written in the **same** write txn as the pointer:
- **`COMMIT_LOG`** (`store.rs:164`) — a `commit_seq: u64 → JSON CommitLogRow`
index. `COMMITS` (`store.rs:157`) is keyed by `(table_key, snapshot_id)` and
cannot be range-scanned by sequence; this second index can, and it includes
snapshot-less commits so `commit_seq` accounting stays gapless. This is the
**source of truth**.
- **`broadcast::Sender<CommitEvent>`** (`store.rs:282`, capacity
`COMMIT_EVENT_BUFFER` = 1024, `store.rs:63`) — the **low-latency hint**.
`Store::publish` (`store.rs:358`) / the `group_commit` tail (`store.rs:413`)
send after the commit is durable + mirror-visible. A send with no receivers is
a cheap no-op, so the poll-only default is free. A lagging subscriber that
overruns the ring gets `RecvError::Lagged` and recovers from the durable log.
Every pointer-advancing path publishes: `group_commit` (single-table),
`atomic_release` / `atomic_release_raw` (`atomic.rs`, one event per advanced
table), and the non-`group_commit` `create_table` / `register_table` paths
(`catalog.rs`, fire their own event). `record_commit` now returns
`RecordedCommit { seq, event }` (`store.rs:197`); the `CommitOutcome`
(`store.rs:103`) carries the event to publish post-commit.
Public consumer surface on `RedbCatalog`:
| method | source | behaviour |
|---|---|---|
| `subscribe_commits()` | `catalog.rs:108` | `(cursor, Receiver)`; subscribes **before** reading the cursor, so a commit racing the subscribe is delivered (and de-duped by `> cursor`) rather than lost |
| `commits_since(after)` | `catalog.rs:121` | durable replay: range-scan `COMMIT_LOG` for `commit_seq > after`, oldest-first, gapless — the catch-up path after a `Lagged` |
| `commit_table_key(ident)` | `catalog.rs:94` | the `table_key` for a `TableIdent`, so a consumer filters the broadcast without reaching into the crate-private key encoding |
| `commit_seq()` | `catalog.rs:79` | the current global cursor |
## f2 — Phase 1: the changelog encoder (`skade/src/read.rs`)
A **re-tagging layer** over `read_delta` + `read_equality_deletes`, not a new
scan. `read_changelog(table, from, to)` (`read.rs:850`) returns a
[`ChangelogBatch`] (`read.rs:788`) whose every row carries a `_change_type`
`Utf8` op tag.
- **Vocabulary** — the `change_type` module (`read.rs:725`):
`INSERT` / `UPDATE_BEFORE` / `UPDATE_AFTER` / `DELETE`, and the sentinel column
name `_change_type`. Byte-for-byte knut-bifrost's `mapping.rs` strings (and
Iceberg's `create_changelog_view` vocabulary), so a skade changelog and a
Flink/Spark/bifrost consumer agree with **zero translation**.
- **Tagging** — `read_delta`'s appended data files become `INSERT` rows;
`read_equality_deletes`'s identities become `DELETE` rows widened to the full
table schema (nulls in the non-identity columns, via `changelog_schema`
`read.rs:809` making every base field nullable + appending the non-null
`_change_type`).
- **Full-reload honesty** — when the window hit an overwrite or a position
delete (`DeltaPlan::needs_full_reload`), it re-reads the whole `to` snapshot
as a pure `INSERT` stream (exactly bifrost's rule) instead of an unsound
additive delta.
- **`TableKind`** (`read.rs:750`) — derived from the schema's Iceberg
identifier-field-ids: `Log` (no PK) vs `PrimaryKey(ids)` (upsert). This is
Fluss/Paimon's LogTable-vs-PrimaryKeyTable distinction, on **one** Iceberg
table, with no RocksDB/LSM.
- **`.collapsed()`** (`read.rs:1307`) — net-change over the window, joining
inserts↔deletes on the delete files' identity columns: on a `Log` table a
matched insert+delete **cancels**; on a `PrimaryKey` table a matched
delete+insert folds into `UPDATE_BEFORE` (old) + `UPDATE_AFTER` (new). Each
row's identity key is built **once** into a `Vec<BatchKeys>` (one packed
`(buf: Vec<u8>, offs)` per batch) and reused for both the match set and the
re-tag pass. The keys are built by a **typed** encoder (`ColEnc` / `BatchKeys`,
`read.rs`): it reads each identity column's native array values straight into
one contiguous per-batch byte buffer as a self-delimiting token (presence byte
+ native LE bytes for fixed types, `u32`-length-prefixed bytes for strings) —
**no** per-row `arrow_cast::display::array_value_to_string` format and **no**
per-row `String` alloc (one buffer per batch, not one String per row). The
encoding is a proper length-delimited serialization, so it is strictly
**injective** (distinct identity tuples never collide — the old `\x1f`-join
could, on embedded-`\x1f` strings; guard `read::tests::typed_row_keys_are_injective`).
The match set is a **zero-copy** `HashMap<&[u8], u8>` borrowing those keys —
INSERT sets bit 0, DELETE sets bit 1, "matched" iff state `0b11`. All three
phases now run on the ONE fork-join engine (gatling): key build + the
**per-batch state fold** (each batch folds into a private local map, then the
partials are OR-merged — OR is commutative/idempotent, so the merged state is
order-independent and byte-identical to the old single serial fold) + the
re-tag/`filter` pass. Cycle 1 (`eb68f3e`) fanned the key-build + re-tag passes
(collapse ~4.5ms → ~2.2ms). Cycle 2 typed the key builder and parallelized the
fold: `examples/cdc_mor_bench.rs` clean interleaved A/B ~10.3ms → ~7.0ms median
(~1.47×, 11.6M → 17M rows/s), cores_busy held ~2.0/12 — byte-identical (the
bench's `content_sig` id-set assert gates every iter; `CDC_CORES_FLOOR`=1.3 is
the RED-when-broken saturation floor).
Cycle 3 (target b) then attacked `read_changelog` itself: `read_delta`'s
**manifest walk** (`read.rs:982`) was an async-serial `for mf in
manifest_list.entries()` loop that read **and** `Manifest::parse_avro`-decoded
every wanted manifest one at a time — the avro parse + entry classification is
pure CPU and ran single-threaded (~1.0 core). Now the wanted manifests' bytes
are async-read (genuine I/O, kept async) into a vec, then the parse + classify
fans **per-manifest** across all cores on the ONE fork-join engine (gatling,
`gatling_map_owned`), each producing a private `ManifestScan` partial that is
merged order-independently (added/delete files sorted+deduped after; a boolean
OR on `needs_full_reload`) — byte-identical to the serial plan. `cdc_mor_bench`
read_changelog median **~28-47ms → ~6.9ms** (contended 12-core box; ~4M →
~17M rows/s), read cores_busy **~1.0 → ~5.5** (`CDC_READ_CORES_FLOOR`=1.5 is
the new RED-when-broken read-path floor; a per-iter raw `content_sig` assert
gates byte-identity every read iter).
Cycle 4 (target c) then took the **point lookup** (`lookup_gatling`, f5) — see
the perf note there. The recorded "bloom-probe loop" target was **invalidated
by measurement** (the per-row-group bloom test is ~31µs, ~3% of a lookup, far
too fine to fan); the real single-CPU costs were the serial per-manifest
`parse_avro` walk and the whole-file data reads, both now fixed.
- **`.project_columns()`** (`read.rs:1033`) — column pruning (the `_change_type`
sentinel is always retained) — the Fluss columnar win, free from Arrow
projection. Prune **after** collapse (collapse needs the identity columns).
`Table::read_changelog` (`table.rs:407`) is the handle convenience.
## f3 — Phase 2: push consume, `ChangeStream` (`skade/src/stream.rs`, feature `stream`)
Inverts CDC from **poll** (`commit_seq()` then `read_delta`) to **push**.
`ChangeStream::changes(from)` (`stream.rs:67`, impl `stream.rs:115`) returns a
`'static BoxStream<Result<ChangelogBatch>>`:
1. subscribe to `subscribe_commits()` **before** reading the current snapshot
(no gap);
2. `from = Some(s)` first yields a catch-up window `read_changelog(s, current)`,
then tails; `from = None` tails only future commits;
3. on each broadcast event, filter to this table (`commit_table_key`), skip
snapshot-less / already-seen events, and yield
`read_changelog(last, event.snapshot_id)`.
A producer→consumer hop is **sub-millisecond and in-process** — no Kafka, no
network.
- **Resilience** — a `RecvError::Lagged` recovers by re-reading to the table's
*current* snapshot in one window (`stream.rs:174`); the durable catalog is the
truth. `Table::changes_polling(from, interval)` (`stream.rs:204`) is the same
stream with **no** broadcast — a pure timer poll, for a consumer that doesn't
want the push path or whose producer runs in another process.
- **Exactly-once** — [`ConsumerCursor`] (`stream.rs:40`) `{ table, snapshot_id,
commit_seq }`. A window's `to_snapshot` is the resume key: persist it, restart
with `changes(cursor.snapshot_id)`, and the first window covers only rows
*strictly after* it (`read_delta`'s semantics), so nothing is delivered twice.
`ConsumerCursor::after(table, batch)` (`stream.rs:51`) mints the next cursor.
## f4 — Phase 3: lineage emission (`skade/src/lineage.rs`, `skade/src/warehouse.rs`)
The event model + sink trait for "who did what, to which tables, between which
systems, when". The **types are always compiled** (serde + async-trait, already
deps); only the automatic emit hook is feature-gated.
- **[`LineageEvent`]** (`lineage.rs:114`) — `event_id`, `actor`, `operation`
([`Operation`]: `Append`/`Delete`/`Overwrite`/`Create`/`Release`, mirroring
skade snapshot ops + knut `_change_type`), `inputs: Vec<DatasetRef>`,
`output: DatasetRef`, `ts_micros`, `commit_seq`. `DatasetRef` / `SystemRef`
carry the cross-system axis (`None` system = "within this warehouse").
- **[`LineageSink`]** (`lineage.rs:219`) — one `async emit`. Best-effort by
contract: a sink error must never fail a commit. Implementors:
- `CapturingSink` (`lineage.rs:229`) — in-memory `Vec`, tests + a viewer's
live tail.
- **`WarehouseLineageSink`** (`lineage.rs:418`) — historizes each event as a
row in a reserved `lineage_events` Iceberg table (`LINEAGE_EVENTS_TABLE`,
`lineage.rs:292`) via `lineage_events_schema()` (`lineage.rs:298`, the
13-column reserved schema) + `LineageEvent::to_arrow_row`/`from_arrow_row`.
skade writing skade — so the lineage feed itself gets time-travel,
`read_changelog`, and the CDC stream for free. A **self-reference guard**
(`is_self_reference`, `lineage.rs:444`) skips writes to `lineage_events`
itself, so appending a lineage row never recurses.
- **`HttpLineageSink`** (`lineage.rs:500`, feature `lineage-http`) — see f6.
- **Emit hooks** (feature `lineage`, default off) — `Table::emit_lineage`
(`table.rs:72`) stamps the actor (`Table::with_actor`, `table.rs:62`) and the
catalog `commit_seq`, and routes: `append`/`ingest`/`ingest_parallel`/
`ingest_pipelined` → one summarizing `Append`; `delete_equality`
(`delete.rs`) → `Delete`; `Warehouse::atomic_release_with_lineage`
(`warehouse.rs:294`) → one `Release` event naming all N tables via
`emit_release` (`lineage.rs:467`). Attach a sink with
`Table::with_lineage_sink` (`table.rs:105`); `Warehouse::lineage_sink`
(`warehouse.rs:273`) ensures the reserved table exists and hands back the
bound sink.
## f5 — Phase 4: the point lookup (`skade/src/read.rs`) + the eq-delete field-id fix
`lookup(table, key, as_of)` (`read.rs:470`, `Table::lookup` `table.rs:455`)
resolves the single current row whose equality-key column(s) match, pinned to a
snapshot (`as_of = Some`, time-travel) or the latest (`None`). It runs through
the engine's **merge-on-read** path, so an upsert (delete-old + append-new)
resolves to the *new* row and a deleted key resolves to `None`; the key is both
pushed down (file/row-group pruning) and applied as an exact row filter. This is
the Fluss/Flink **delta-join** idea — "externalize the join state, probe the
latest at emit" — done **embedded**: skade's catalog *is* the point-lookup
store, with no external state service.
The lookup surfaced a latent bug it also fixes: `delete_equality`
(`delete.rs:40`) now **stamps each equality-delete parquet column with its
Iceberg field-id** (`PARQUET_FIELD_ID_META_KEY`). Without it, a merge-on-read
scan over an eq-delete table called `arrow_schema_to_schema` on a file whose
columns lacked ids and errored `"Field id not found in metadata"` — a panic deep
in the vendored engine's delete loader. Writing the ids at the write side is both
the correct Iceberg on-disk form and the fix; skade is the canonical publisher of
the fork, so it lands here.
**Cycle-4 perf (bloom fast-path `lookup_gatling`).** The append-only single-
schema no-delete fast path was profiled (`examples/lookup_bloom_bench.rs`). The
recorded target — the per-row-group **bloom-probe loop** — turned out to be a red
herring: the `Sbbf` read + hash-check is ~31µs for 200 row groups (~3% of a
lookup), and fanning it *regressed* by ~150× (a fresh gatling scope per lookup
dwarfs µs of work). The genuine single-CPU costs, and their byte-identical fixes:
- **Manifest walk** — `Manifest::parse_avro` is ~325µs of pure CPU **per
manifest**, run serially, so a table with N appends paid ~N·325µs (>10ms at
N=32). Now the manifest bytes are async-read (~8µs each) and the
`parse_avro` + entry classification fans **per manifest** across all cores on
the ONE fork-join engine (gatling), merged order-independently (file lists
concatenated then sorted+deduped; a declining manifest ⇒ decline) — exactly the
cycle-3 `read_delta` shape.
- **Whole-file data reads** — each candidate file was slurped whole (~380µs) even
to match nothing. Now the manifest's per-file **min/max bounds prune** whole
files the key can't be in *before any read* (`scalar_out_of_file_bounds`, keyed
on `field_id_by_name`; conservative — only an affirmative type-matched
out-of-range proof skips a file), so an id-clustered append set reads exactly
the one file whose range covers the key (or zero, out of range).
A/B on the light (contended 12-core) bench, 200 probes over 32 appends:
**~2288ms → ~455ms (~5×)**, cores_busy **0.93 → ~7.5** (`LOOKUP_CORES_FLOOR`=1.3
is the RED-when-broken manifest-fan floor, asserted at ≥8 manifests). 1-file case
~175ms → ~158ms (pruning still trims the out-of-range probes). Byte-identity is
asserted every probe in the bench, plus guards `lookup_prunes_out_of_range_files_via_bounds`
(pruning skips the read: `row_groups_seen == 0`) and
`lookup_gatling_multi_manifest_byte_identical` (the parallel manifest merge
resolves the exact row per file). Authoritative HEAVY numbers are still owed on a
quiet box.
**Cycle-5 perf (target d — cross-probe caching + byte-range reads).** Cycle 4
fanned the manifest `parse_avro` across cores, but a delta-join emit loop probing
many keys against ONE snapshot still re-read + re-parsed *every* manifest **and**
slurped *every* surviving data file whole **per probe**. The light bench's
telemetry confirmed the manifest re-parse dominated (462ms/200 probes at
cores_busy ~7.4 — the fan was firing on every probe). Both are now amortised, all
byte-identical:
- **Cross-probe cache** (`LookupCache`, held on the `Table` handle, keyed by
*immutable* ids so it never yields a stale answer): the parsed manifest set —
every live data file with its recorded min/max bounds — is parsed **once per
snapshot** (keyed by `snapshot_id`; on a hit the manifest list isn't even
loaded) and reused by every later probe, which then applies only the pure-CPU
min/max bound pruning over the cached bounds. Each data file's parquet **footer**
is parsed once (keyed by file path — Iceberg files are write-once). A commit
advances the snapshot id, so the manifest set auto-invalidates by key miss.
- **Byte-range reads** (respecting the object-store/`FileRead` seam, reusing the
vendored `ArrowFileReader` as the `AsyncFileReader`): with the footer already
parsed, the async `ParquetRecordBatchStreamBuilder` reads only the bloom regions
+ only the surviving row group's column chunks — never the whole file (the S3
win: a range GET, not a full object GET). A `CountingReader` decorator meters
every fetched range into `LOOKUP_BYTES_READ` (`pub fn lookup_bytes_read`).
A/B on the light (contended 12-core) bench, 200 probes over 32 appends:
**~463ms → ~5.7ms (~81×)**, 432 → **35 276 lookups/s**, **~22.9 KB read/probe**
(a whole-file slurp is ~100 KB+/probe). cores_busy legitimately drops 7.4 → ~0.9
— the parallel parse now runs once on the cold probe, not per probe. Guard swap:
the RED-when-broken invariant is now a **bytes-read-per-probe ceiling**
(`LOOKUP_BYTES_CEIL`=49152, proven to fire) — a regression to a whole-file slurp
or a defeated cache blows it; the old `LOOKUP_CORES_FLOOR` is retained but OFF by
default (the hot loop is no longer parse-bound). Byte-identity is asserted every
probe in the bench; the unit guards (`lookup_gatling_skips_row_group_via_bloom`,
`lookup_prunes_out_of_range_files_via_bounds`,
`lookup_gatling_multi_manifest_byte_identical`) all still pass unchanged.
Authoritative HEAVY numbers are still owed on a quiet box.
**Cycle-6 perf (target e — fd handle reuse + coalesced bloom reads).** Cycle 5
killed the manifest re-parse and the whole-file slurp, leaving a hot probe's
residual cost as **per-probe fd setup**. First *verified* (the recorded discipline)
via new opens/probe + gets/probe telemetry on the light bench that the residual is
genuinely setup / round-trip count, **not** I/O: a probe reads only ~23 KB yet
issued **~0.75 fd opens + ~6.9 byte-range reads per probe**, and on the local
backend each `FileRead::read` is a seek+read syscall (~few µs) — so ~9.6 syscalls
≈ the whole ~27 µs/probe. Two byte-identical fixes:
- **fd handle reuse** — the open `FileRead`/fd is cached on the `LookupCache`
keyed by *immutable* file path (Iceberg files are write-once, so an fd never goes
stale — no snapshot invalidation, same rule as the footer cache) and reused by
every later probe: a probe builds a fresh `ArrowFileReader` over a cheap `Arc`
clone (`CountingFileRead`) instead of `reader()`-ing a new fd. opens/probe → ~0.
- **coalesced bloom reads** — instead of `get_row_group_column_bloom_filter` one
row group at a time (one round-trip each), every row group's key-column bloom
region (`bloom_filter_offset..+length`, both recorded by the writer) is fetched
in ONE merged `get_byte_ranges` — the vendored reader coalesces the ranges into a
single fd read — and each buffer is reconstructed with `Sbbf::from_bytes` (the
identical filter ⇒ identical prune verdict; a foreign file with no recorded bloom
length falls back to the per-row-group path). Counting moved to the **`FileRead`
seam** (`CountingFileRead`) so the GET metric reflects real post-merge round-trips.
- **write-side enabler** — bloom filters are now written **contiguously at the file
End** (`BloomFilterPosition::End`, set in `WriteProps::to_writer_properties` when
any bloom column is enabled) instead of the parquet default `AfterRowGroup`.
Interleaved blooms would coalesce *across* the intervening row-group data and pull
~the whole file (measured: bytes/probe 22.9 KB → 113 KB); packed at the End they
merge into one small read. Pure physical-layout change — bloom content + lookup
result unchanged.
A/B on the light (contended 12-core) bench, 200 probes over 32 appends:
**~5.9ms → ~4.15ms (~1.42×)**, ~33.8k → **~47k lookups/s**; **opens/probe 0.75 →
0.000**, **gets/probe 6.86 → 1.555 (~4.4× fewer fd round-trips)**, **bytes/probe
22 911 → 22 911 (byte-for-byte identical fetch)**. The modest wall-time gain is
expected on a local page-cache bench (each read is a cheap syscall); the headline
is the round-trip collapse, which on a real object store (each range GET = a
network RTT) is the dominant latency. Byte-identity is asserted every probe in the
bench; the RED-when-broken guards are extended with a **per-probe GETs ceiling**
(`LOOKUP_GETS_CEIL_X1000`=3500, proven to fire — a lost-coalescing regression hits
~6.9) and a **per-probe opens ceiling** (`LOOKUP_OPENS_CEIL_X1000`=250, proven to
fire — a defeated fd cache hits ~0.75), alongside the retained cycle-5 bytes
ceiling. Unit guards (`lookup_gatling_skips_row_group_via_bloom`,
`lookup_prunes_out_of_range_files_via_bounds`,
`lookup_gatling_multi_manifest_byte_identical`, `table_lookup_uses_bloom_fast_path`)
all pass unchanged. Authoritative HEAVY numbers are still owed on a quiet box.
**Cycle-7 outcome (target f — predicate pushdown evaluated, NOT shipped; the point
lookup is essentially optimal).** With fd opens + bloom round-trips gone, a present
probe was down to ~2 fd reads and the recorded hypothesis was that the pure-CPU
decode+scan of the one bloom-surviving ~1000-row row group now dominates, to be cut
by pushing the equality predicate into the reader (parquet `RowFilter` /
`RowSelection`) so only the matching row materializes. Both halves were **measured**
on the light bench (200 probes / 32 appends) with a new `LOOKUP_ROWS_DECODED`
metric + a decode-vs-read split timer:
- **Verified decode dominates** — a present probe's ~23µs splits ~2µs bloom fetch +
~14-17µs decode+scan of the one surviving row group (rows_decoded/probe **782.5** =
the whole ~1000-row group per present probe), + fixed parquet stream setup. So the
residual is genuinely decode/scan, not the 2 reads.
- **But the pushdown is a net LOSS, proven by A/B** — a per-key-column `RowFilter`
(reusing `scalar_eq_at`, byte-identical) *did* collapse rows-materialized 782.5 →
**~0.8**, yet the mandatory cost is reading+decoding the ONE bloom-surviving row
group's **key column** — a probabilistic bloom hit MUST be confirmed by decoding
the key column and locating the row — which a `RowFilter` cannot remove; it only
skips the (already cheap, on narrow rows) *non-key* materialization. Net on the
representative narrow workload: **median 3.95ms → 4.5ms (~13-15% WORSE)** — the
filter's two-phase (predicate-cols then output-cols) machinery costs more than it
saves — and **gets/probe 1.56 → 2.31** (the split read adds a fd round-trip),
directly against the cycle-6 object-store round-trip doctrine. On an artificially
wide payload (400-byte non-key column) it wins only ~15% while still adding the
round-trip and barely cutting bytes (the column chunk is the I/O granule, read
whole regardless). So the pushdown was **reverted**; the path keeps its single
full-row-group decode.
- **Kept** — the `LOOKUP_ROWS_DECODED` telemetry + `lookup_rows_decoded()` are
permanent, and the bench gains a **RED-when-broken rows-decoded-per-probe ceiling**
(`LOOKUP_ROWS_CEIL`, default 2×`rg`, proven to fire): a present probe decodes ~one
row group, so a defeated bloom / bound prune (decoding many row groups or a whole
file) blows past it by orders of magnitude — the direct pruning-still-works guard,
complementing the cycle-5/6 bytes/gets/opens ceilings (all unchanged, still green).
**The point-lookup path is now essentially optimal**: a present probe = ~2 fd reads
+ the irreducible decode of the one bloom-surviving row group's key column. Further
gains need a *stronger index* (a sorted-key + page-index binary search, or an exact
secondary index) that avoids the full key-column decode — a bigger design change, not
a reader tweak. The optimize loop should **pivot to a different skade hot path** (or
await the #88 HEAVY sweep on an idle box for the authoritative numbers).
## f6 — the OpenLineage HTTP sink (`skade/src/lineage.rs`, feature `lineage-http`)
`HttpLineageSink` (`lineage.rs:500`) maps each `LineageEvent` to an OpenLineage
**`RunEvent`** (`run_event`, `lineage.rs:548`, public so it is testable without a
live POST) and POSTs it JSON to a collector (Marquez/OpenLineage), so skade
commits land in the same lineage graph as Spark/Airflow runs. `eventType` is
`COMPLETE` (a skade commit is a completed run); `run.runId` is a deterministic
UUID from the event id (`run_id_uuid`, two FNV-1a passes, no `uuid` crate);
`eventTime` is RFC-3339 from `ts_micros` (`rfc3339_from_micros`, civil-from-days,
no time-zone crate); datasets are versioned by snapshot id; `commit_seq` rides as
a run facet. The blocking `ureq` call is fenced onto `tokio::task::spawn_blocking`
so it never stalls the async runtime. Its own feature keeps `ureq` + `serde_json`
out of every other build; enabling it does **not** imply the `lineage` emit hook.
## The comparison — and skade's edge
| capability | Fluss | Paimon | Hudi | Iceberg (Spark CDC) | **skade** |
|---|---|---|---|---|---|
| changelog verbs (`INSERT`/`UPDATE_*`/`DELETE`) | ✅ | ✅ | ✅ | ✅ (`create_changelog_view`) | ✅ same strings (`change_type`) |
| Log vs PrimaryKey table | ✅ (2 table types) | ✅ (2 table types) | ✅ (COW/MOR) | one table type | ✅ derived from identifier-field-ids, **one** table |
| push consume | Flink source | Flink/Spark | Flink/Spark | Spark micro-batch | in-process `broadcast`, **sub-ms**, `ChangeStream` |
| exactly-once cursor | Flink checkpoint | offset | commit time | snapshot id | `commit_seq` / `to_snapshot`, `ConsumerCursor` |
| point lookup / delta-join | external state svc | LSM lookup | index | — | **embedded** MOR probe (`lookup`) |
| lineage | — | — | — | OpenLineage (Spark integration) | native emit + `lineage_events` table + OpenLineage POST |
| runtime | JVM + coordinator | JVM | JVM/Spark | JVM/Spark | **pure Rust, in-process, one file** |
| state store | RocksDB/LSM | LSM | index files | — | the redb catalog itself |
**skade's edge:** the "log" and the "table" are the *same* embedded, single-file
ACID Iceberg catalog. There is no separate log service (Fluss's coordinator, a
Kafka/Debezium tier) and no separate state store (RocksDB/LSM) — the durable
`commit_seq` *is* the log offset, the redb pointer *is* the current-state store,
and the broadcast is an in-process sub-ms hint rather than a network hop.
Everything is pure Rust, runs inside the consumer's process, and ships as one
file on disk.
## Public API surface + feature flags
**`skade-katalog` (always on):** `CommitEvent`, `COMMIT_EVENT_BUFFER`;
`RedbCatalog::{subscribe_commits, commits_since, commit_table_key, commit_seq}`.
**`skade` (types always on):** `read_changelog`, `ChangelogBatch`,
`TableKind`, `change_type`, `lookup`, `DeltaPlan`, `EqualityDeleteFile`,
`Scalar`; `Table::{read_changelog, read_delta, read_equality_deletes, lookup}`;
lineage `LineageEvent`, `LineageSink`, `CapturingSink`, `DatasetRef`,
`SystemRef`, `Operation`, `WarehouseLineageSink`, `lineage_events_schema`,
`emit_release`, `LINEAGE_EVENTS_TABLE`; `Warehouse::lineage_sink`.
| feature | default | adds |
|---|---|---|
| `stream` | off | `ChangeStream`, `ConsumerCursor`, `Table::{changes, changes_polling}` |
| `lineage` | off | the automatic emit hooks on `append`/`ingest*`/`delete_equality`; `Table::{with_lineage_sink, with_actor}`, `Warehouse::atomic_release_with_lineage` |
| `lineage-http` | off | `HttpLineageSink` (OpenLineage POST); does **not** imply `lineage` |
All three are purely additive; the default build compiles none of them and pays
nothing on the write path beyond the one durable `COMMIT_LOG` row (no extra
fsync).