<!-- nornir:class: design -->
# Doris sink — Bifröst gains a table sink
> **Class:** `design` · **Function:** `doris-sink` · **Uses:** `cdc`, `bifrost`,
> `skade`. This is the delivery half of the CDC story: skade computes the diff
> (its `cdc-stream-design.md`), bifröst already streams it
> ([`cdc-design.md`](cdc-design.md), [`bifrost-design.md`](bifrost-design.md)),
> and this adds the one genuinely missing piece — a **table**-shaped sink.
>
> **Read f0 before f1.** The premise this was originally specced against does not
> survive contact with the Doris version actually deployed.
## f0 — the gap this closes, and how much smaller it turned out to be
### The measurement
On the `red` bench cluster a 2 000 000 000-row person table exists twice: once as
Doris-native rows (`fake.personer2g`, 81.6 GB incl. replicas), once as Iceberg
zstd-parquet on MinIO (`iceberg.bench.doris2g`, 23 GiB). The same point lookup
(`WHERE personnummer = <x>`), measured server-side:
| where the rows live | point lookup | note |
|---|---|---|
| Iceberg on MinIO, Doris file cache **off** | **41 838 ms** | instrumented: 20.69 GB read at 495 MB/s — 100 % disk-bound |
| Iceberg on MinIO, Doris file cache warm | **1 517 ms** | the recorded number; a later run was reported at 962 ms |
| Doris **native** storage | **24 ms** | 34 ms without the inverted index |
Source: `ubm-play-red-apps/docs/benchmarks.md`. Three corrections to the earlier
draft of this spec, which must not be repeated:
* **The inverted index is worth ~30 %, not 46×.** The gap comes from native
storage plus predicate pushdown plus zone maps; `idx_pnr` moved 34 ms → 24 ms.
* **The MinIO volume is not NVMe.** One Scaleway block volume, `ROTA=1`, ~495 MB/s
ceiling. The Doris *file cache* is on BE local disk; the bench does not
characterise that disk, so neither does this document.
* **962 ms is not in the cited source.** The bench records the warm number as
1 517 ms. Both are reported here rather than quietly picking one.
Getting the 24 ms path on lake-sourced data means physically copying rows into the
engine. So far, so uncontroversial.
### The premise that broke
The original argument was: *Doris async materialized views over an Iceberg
catalog support only `COMPLETE` refresh — partition-level refresh is Hive-only —
so every cycle rewrites all 2 B rows to absorb a day's appends. Therefore build a
CDC delivery sink.*
That was true for Doris **2.1 and 3.0**. It is **not true for the cluster in
question**. `ubm-play-red-apps/versions.env` pins `doris_version=4.0.8`, and
Doris's async-MV support matrix records partitioned refresh over **Iceberg as
"Supported in 3.1"** (alongside Paimon and Hudi). So the deployed engine can, in
principle, refresh a materialized view partition-by-partition rather than whole.
What is *still* true, and what actually leaves a gap:
* **Partitioned MV refresh over Iceberg requires the Iceberg table to be
partitioned**, on a **single** partition column, with only limited partition
evolution tolerated. `iceberg.bench.doris2g` is **not partitioned** — the bench
doc says so explicitly ("Getting real pruning on Iceberg needs *partitioning* …
Not yet done"). On the table as it exists today, `COMPLETE` is the only
available refresh. That is a property of the *table*, not of Doris.
* **Auto-trigger is not supported for any external catalog**, so even a
partitioned MV refreshes on a schedule, not on change. Freshness is bounded by
the schedule either way.
* Doris's data-lake metadata cache can leave an MV believing it is in sync after
an external `INSERT OVERWRITE` that does not change the metadata it watches.
**So the honest gap is narrower than advertised**: not "Doris cannot do this",
but "Doris can do this only for partitioned Iceberg tables, only on a schedule,
and the table we measured is not partitioned". The cheapest fix for the measured
case may well be *partition the Iceberg table*, not *build a daemon*. That
possibility is now the first item in f8, and it may be the whole answer.
The case that survives regardless: a `TableSink` is engine-generic, handles
deletes, and is not restricted to one partition column. If bifröst is to deliver
CDC anywhere other than a graph, the seam has to exist. That — not the 46× — is
the justification for building it.
## f1 — where this belongs, and why not the other three candidates
The earlier draft put this in **skade**, as a new `ChangelogSink` trait plus a
new `skade-streamer` daemon. That was wrong, and the evidence is mechanical.
### It belongs in `knut-bifrost`
Bifröst *is* the CDC streamer. The daemon the earlier draft proposed already
exists here, under another name:
| the daemon needed | what bifröst already has |
|---|---|
| poll a table for new snapshots | `Listener::run` / `run_with_backoff` (`listener.rs:317,335`), with exponential backoff on idle rounds |
| read the CDC window | `IcebergSource::delta_from_table` → `skade::Table::read_changelog` (`source.rs:767`); `SkadeCdc` (`cdc/skade.rs:87`) |
| split inserts from deletes | `partition_change_rows` (`transform.rs:215`) — routes all four `_change_type` tags through the one `cdc::ChangeOp` vocabulary |
| durable per-table cursor | `SyncState` (file or Valkey) + `TableState::snapshot_id` (`state.rs:26,72,89`) |
| apply, *then* advance the cursor | `listener.rs:298` → `listener.rs:303`, comment: *"Only after the sink succeeded"* |
| metrics / Prometheus / structured logs | `metrics.rs`, `metrics_http.rs`, `tracing` targets `bifrost::*` |
| a CLI driver | `bin/bifrost` behind feature `cli` |
Bifröst also already carries the two dependencies a Doris sink needs — `reqwest`
(optional, feature `airflow`) and `skade 0.5.3` (optional, feature `skade`) — so
the Doris feature pulls **no new dependency tree** into the workspace.
### Why not skade
skade already owns the delta primitives and should keep owning them:
`Table::read_delta` / `read_changelog` (`skade/src/read.rs:2750,3045`),
`ChangelogBatch::collapsed`, `Table::read_equality_deletes`,
`DeltaPlan::needs_full_reload` (`read.rs:2728`), `ConsumerCursor`
(`skade/src/stream.rs:40`), `Table::changes_polling`. Those are *lake* concerns.
A Stream Load client is not. It is an HTTP client for one specific MPP engine,
and skade's own charter ("add to skade without destroying skade") keeps the data
plane free of engine knowledge. skade has no HTTP client at all; its existing
`LineageSink` is an **OpenLineage metadata** sink and is unrelated to a data
sink. Putting an engine-specific loader there would be the first crack.
### Why not knut-thund
Þund has a sink notion — `OutputType::Sink` plus a `format` string plus `knut.*`
keys in `Dataset.properties` — and even a working non-file external sink
(FalkorDB). But:
* Þund's own precedent **delegates the wire protocol**:
`backend/native.rs::write_graph_sinks` maps IR → `GraphMutation` and hands the
protocol to `knut_bifrost::sink::FalkorDbSink`. The parallel move puts the
Stream Load client in bifröst, not in thund.
* The dependency direction forbids the alternative anyway. `knut-pipelines`
depends on `knut-bifrost`, and thund depends on `knut-pipelines`, so **bifröst
cannot depend on thund**; thund → bifröst is the only legal edge.
* Þund's sinks write *after the whole dataflow has materialised*, from an
in-memory `Vec<RecordBatch>` per dataset. Wrong shape for a windowed,
cursor-driven bulk load.
* Þund has **no HTTP client anywhere** in its tree, and its own gap doc
(`thund-vs-spark-gap.md` §4 item 7) explicitly deprioritises connector breadth.
A thund-side `format = "doris"` dataset remains possible later, and should be
built the way the graph sink was: IR mapping in thund, protocol in bifröst.
### Why not a new crate
Same grounds on which `graphar-falkordb` is a separate crate but `FalkorDbSink`
is not: the sink is the *adapter* and belongs beside its siblings. A separate
crate would need its own copy of the listener contract, the `_change_type`
vocabulary and the error type — exactly the duplication the workspace charter
forbids. If the Stream Load client ever grows client-side concerns of its own (a
BE topology cache, connection pooling, a non-HTTP protocol), splitting a
`knut-doris` out then is a mechanical move. Today's code does not justify it.
### What this means for the hypothesis on the table
The working hypothesis was: *delta computation belongs in skade where it already
lives; streaming/orchestration already exists in knut-bifrost; only a
table-shaped sink is genuinely new, and it belongs in knut.* That survives. Two
corrections to it:
1. **Bifröst does not currently refuse a full reload — it silently performs one.**
`source.rs:801-821` logs `full_reload = plan.needs_full_reload` and passes the
whole-snapshot re-read straight through to the sink. Correct for FalkorDB
(`MERGE` is idempotent, the graph is small); catastrophic for Doris. The
refusal is **new behaviour**, not inherited. See f6.
2. **The refusal cannot live where the earlier draft put it.** See f6 — as
specified, it would have paid for the full read before refusing to write it.
## f2 — the seam: `TableSink`, a sibling of `GraphSink`
`GraphSink` is `async fn apply(&mut self, &GraphMutation) -> Result<SinkReport>`
(`sink.rs:19-22`). It cannot be generalised into service here, and the reason is
not stylistic: `GraphMutation` is `Vec<VertexSet>` + `Vec<EdgeSet>` +
`Vec<VertexDelete>`, and every one of those carries `label`, `id_column`,
`src_label`/`dst_label`/`edge_type`, `edge_key_props` (`mutation.rs`). A Doris
load has none of those, and `SinkReport` counts `vertices_written` /
`edges_written` / `deleted`. Making the graph fields optional to accommodate a
table produces a type whose majority of fields are meaningless to either caller —
strictly worse than two honest traits.
So: a **sibling**, and a module split so two sink families never accumulate in
one file.
```
src/sink/
mod.rs re-exports; the shared row-rendering helper stays private
graph.rs GraphSink + FalkorDbSink (moved verbatim from sink.rs)
table.rs TableSink + TableWindow + LoadAck (no new dependency)
doris.rs DorisSink + the Stream Load client (feature `doris`)
```
The trait:
```rust
#[async_trait]
pub trait TableSink: Send {
/// Fail fast on a schema the sink cannot accept. Called once before the
/// first window; a mismatch here refuses to start rather than half-loading.
async fn probe(&mut self, schema: &Schema) -> Result<()>;
/// Apply exactly one CDC window. Idempotent by contract — the same window
/// may arrive twice after a crash between the load and the cursor commit.
async fn apply(&mut self, window: &TableWindow<'_>) -> Result<LoadAck>;
}
```
`TableWindow` is the window, not a mutation: the source table, the snapshot pair
that identifies it, and the already-partitioned rows.
```rust
pub struct TableWindow<'a> {
pub table: &'a str,
pub snapshots: SnapshotWindow, // { from: Option<i64>, to: i64 }
pub upserts: &'a [RecordBatch],
pub deletes: &'a [RecordBatch],
}
```
`upserts` / `deletes` come from the existing `partition_change_rows`
(`transform.rs:215`), reused rather than reimplemented. That is what makes the
four-tag mapping correct for free: `INSERT` / `UPDATE_AFTER` / folded `UPDATE` →
upserts, `DELETE` → deletes, `UPDATE_BEFORE` dropped (its paired `UPDATE_AFTER`
fully overwrites the row), unknown tag → hard error.
## f3 — the ordering that must not be swapped, enforced by the type system
The rule: **commit the cursor strictly after the sink acknowledges.**
Crash between load and commit ⇒ the window is replayed ⇒ the label is
byte-identical ⇒ Doris rejects the duplicate ⇒ the replay is a no-op. That is
at-least-once delivery made idempotent by the sink.
Commit first, and a crash before the load loses the window **silently**. No
error, no retry, no lag metric — the rows are simply not there. It is the worst
class of bug this design can have.
`listener.rs` gets this right today, but by *comment* (`listener.rs:301`). A
comment is not a mechanism. The table path encodes it in types:
```rust
/// Proof that a window was durably accepted by a sink. Its only constructor is
/// private to `sink::table`, and `TableSink::apply` implementations are the only
/// callers — so a `LoadAck` cannot exist unless a load actually acked.
pub struct LoadAck { /* labels, to_snapshot, rows, deletes, seal: Seal */ }
/// The single choke point. The ONLY function that advances a table-delivery
/// cursor, and it cannot be reached without a `LoadAck`.
pub async fn deliver_window<S: TableSink>(
sink: &mut S,
cursor: &mut SyncState,
window: &TableWindow<'_>,
) -> Result<LoadAck>;
```
The cursor's new value is read off `LoadAck::to_snapshot`, never off
`TableWindow`. A later edit that wants to advance the cursor early must first
invent a `LoadAck`, which it cannot do from outside the module. That is the
entire point of the seal.
## f4 — idempotent replay via Doris's own label
Doris Stream Load takes a caller-supplied `label` header and rejects a load
carrying a label it has already seen. **Measured on the live cluster (Doris
4.0.8)**, not assumed: three rows loaded three times into `sltest.t` —
* load #1, label `skade-sltest:t-1042-1043` → `Status: Success`, `NumberLoadedRows: 3`
* load #2, *identical* label → `Status: Label Already Exists`, `NumberLoadedRows: 0`,
`[LABEL_ALREADY_EXISTS] ... Label [skade-sltest:t-1042-1043] has already been
used, relate to txn [862], status [VISIBLE]`
* load #3, *different* label, same three rows → `Success`, 3 rows. Final count 6.
So a replay under the same label is a genuine no-op, and a different label does
**not** deduplicate — the whole argument rests on the label being *derived*.
One trap in that result: the rejection arrives as **HTTP 200** with the verdict
in the JSON `Status` field. A client that treats 200 as success would count a
rejected replay as a successful load and report rows that never landed. The
result parser keys on `Status`, never on the HTTP code.
Be precise about what that buys, because the earlier draft overclaimed. Doris's
own documentation calls this **At-Most-Once**, not exactly-once: *"Data with the
same label cannot be imported more than once … so that duplicate requests for the
same batch are accepted only once."* Combined with a retry loop and a cursor that
only advances on success, at-most-once per label plus at-least-once delivery is
what yields effectively-once. Doris reserves the term "exactly-once" for its
**two-phase commit** mode (`two_phase_commit: true`, then `txn_operation:
commit|abort` against `/_stream_load_2pc`). This design deliberately does *not*
use 2PC: it adds an in-flight transaction to reason about across a crash, and the
label-derivation trick already gives a correct replay. 2PC is the documented
escalation if the caveats below ever bite.
The label is **derived**, never generated:
```
skade-<db>:<table>-<from>-<to>-<chunk> e.g. skade-fake:personer2g-1042-1043-0
skade-<db>:<table>-init-<to>-<chunk> when from is None (first run)
```
**A `.` is invalid and the earlier draft of this spec was wrong to use one.**
Doris validates the label against `^[\-_A-Za-z0-9:\p{L}]{1,128}$` and rejects
anything else outright:
```
[ANALYSIS_ERROR] ... Label format error. regex: ^[\-_A-Za-z0-9:\p{L}]{1,128}$
```
Hence `:` between database and table — it is in the permitted set. The 128-char
cap is enforced by the builder, and a unit test asserts every generated label
matches the regex.
Derived from the snapshot pair means a replay of the same window produces the
same label with no state kept anywhere. The `<chunk>` suffix exists because a
window may exceed one Stream Load (f5); chunk *k* of a given window always gets
the same label, so a partial replay re-sends only what was missing.
Three caveats that belong in the design because each can bite:
* **Labels expire, sooner than the obvious config says.** Doris documents
`label_keep_max_second` (default 3 days) on the Stream Load page, but a Stream
Load is governed by `streaming_label_keep_max_second`, whose default is
**43 200 s = 12 hours**. After expiry, *"the removed labels can be reused"* —
a replay is no longer deduplicated and would double-load. A crash replay
happens within seconds, so the window is narrow; a job that has been down for
more than 12 hours and resumes mid-window is the case to worry about. Do not
paper over it.
* **A CANCELLED load frees its label immediately.** Dedup is not absolute.
* **Sanitising is not enough on its own — it has to be injective.** Folding every
disallowed character to `_` makes `personer.2g` and `personer_2g` render the
same label, and a shared label between two tables is a silently skipped load.
The builder therefore appends a hash of the original whenever a character
actually had to be replaced, leaving well-behaved names untouched and readable.
Truncation for the 128-char cap works the same way and never touches the
snapshot or chunk fields. This was found by the collision test, not by
inspection.
## f5 — the Doris sink specifically
### Topology, and the error that looks like auth and is not
Stream Load answers on the **FE** (`http_port`, default 8030) at
`PUT /api/{db}/{table}/_stream_load`, and the FE replies **307** with a
`Location` naming a **BE** (`webserver_port`, default 8040) chosen round-robin,
which does the actual work. (The `307` status is stated outright in the Doris 4.x
key-features page; the per-version manuals say only "HTTP redirect". `red` runs
4.0.8, so the 4.x wording is the one that applies.)
Consequences:
* the loader must reach the FE **and every BE it can be redirected to**;
* the redirect `Location` carries the BE's *configured* address, which on a
cluster with split internal/external addressing is not necessarily routable
from where the loader runs;
* credentials must survive the redirect (`curl` needs `--location-trusted`; a
client that re-authenticates on redirect will appear to fail authentication).
**And the failure mode is a hang, not an error.** An earlier draft of this spec
claimed an unreachable BE "fails in a way that reads like an auth error". That is
wrong, and it was measured: with the FE at `172.16.12.13:8030` answering
`/api/health` with 200 and the BEs on `:8040` unreachable, the Stream Load
**hangs after the 307 until something times out**. Nothing in the FE's health
looks wrong while this happens.
The practical consequence is a design requirement, not a documentation note:
without an explicit connect and request timeout on the redirected hop, one
routing mistake stalls the stream **forever, with no error at all**. So the sink
* disables automatic redirect following and re-issues the `Location` itself,
* puts a hard connect timeout (10 s) and request timeout (300 s) on every hop,
* and reports the stall as what it is, naming the BE and the timeouts.
Doris also permits PUTing directly to a known-healthy BE, which is the escape
hatch when FE-issued addresses are unroutable.
### `Expect: 100-continue` is mandatory
Also measured: omit it and Doris answers
`{"status":"FAILED","msg":"There is no 100-continue header"}`. This is worth its
own heading because of how it presents — `curl` sends the header for large
bodies and omits it for small ones, so a client that does not set it explicitly
sees **small windows fail while large ones succeed**. That looks like data
corruption long before it looks like a missing header. The sink sets it
unconditionally on every load, and a test asserts it is present on every PUT,
deliberately using a one-row window.
Placement, in order of preference:
1. **Co-located with the Doris BEs.** The 307 hop stays local.
2. **Its own small host in the dataplattform zone.** One extra hop, cleaner
lifecycle; preferred once several tables stream and load is spiky.
Either way the process needs: MinIO/S3, the skade catalog (redb file or
`skade-katalog serve`), Doris FE `:8030`, and every BE `:8040`.
### The write belongs to Doris
The sink ships rows over HTTP. Doris's BE decides encoding, compaction and
indexing on its own storage. bifröst never writes into a Doris data directory and
must never be handed a path into one — the moment one program writes into
another's storage directory you own both forever. That coupling argument is the
whole reason this is a network protocol and not a file drop.
### Table models and what a `DELETE` row means
| mode | Doris table model | `DELETE` rows |
|---|---|---|
| `append` | duplicate key | **asserted absent** — the window fails if any appear |
| `upsert` | unique key + `__DORIS_DELETE_SIGN__` | applied, delete-sign column set to 1 |
`append` is not merely an optimisation. On an append-only source an unexpected
delete means the premise of the deployment is wrong, and a loud failure beats a
silent divergence between lake and engine.
Mechanics, corrected from the earlier draft:
* The hidden column is spelled `__DORIS_DELETE_SIGN__` (two underscores each
side), `TINYINT`, default 0. Doris FE automatically adds
`__DORIS_DELETE_SIGN__ != true` to every query plan.
* It is a **Unique Key** feature. It is *not* gated on
`enable_unique_key_merge_on_write`; that association was invented by the
earlier draft. Aggregate/Duplicate key tables must use `DELETE` statements
instead — which is precisely why `append` mode refuses delete rows rather than
trying.
* Stream Load marks deletes via the **`merge_type`** header (`APPEND` default /
`DELETE` / `MERGE` with a `delete:` condition) or by naming
`__DORIS_DELETE_SIGN__` in `hidden_columns` / `columns`. `DELETE ON` is Broker
Load / Routine Load syntax and does not apply here.
* If a sequence column is configured, a delete only takes effect when the
sequence value is ≥ the current one — otherwise it *silently* does not apply.
The sink does not configure sequence columns and documents that it does not.
### Payload and chunking
`TableWindow` carries Arrow. Stream Load's `format` header accepts `csv`, `json`,
`parquet`, `orc` and `arrow`. The implementation ships **JSON**
(`format: json`, `strip_outer_array: true`) because it needs no dependency
bifröst does not already have and is unambiguous about nulls and types.
**Parquet** and **arrow** are the throughput follow-ups, and are the reason the
encoder is a small enum rather than a hard-coded function. Which wins is an
empirical question (f8), not an assumption.
Chunking is not optional. The BE caps a single Stream Load at
`streaming_load_max_mb` (default **10 240 MB = 10 GB**), and Doris's own guidance
is that Stream Load *"is generally suitable for loading data less than a few GB"*
and that a single file should stay under 10 GB — a dropped connection restarts
the whole request. So the sink splits a window into chunks under a configured
byte budget (default well below the cap), each with its own derived label, and
only issues a `LoadAck` when **every** chunk has acked. A partially-loaded window
leaves the cursor where it was and replays; the already-loaded chunks are
rejected by label and the missing ones go through.
### Schema
`probe()` compares the Arrow schema against the Doris table's columns and
**refuses to start** on a mismatch. Discovering a column-order or type difference
mid-stream, after a partial load, is far worse than refusing to boot.
## f6 — `needs_full_reload`: refuse, and refuse in the right place
`DeltaPlan::needs_full_reload` is set when a commit in the window rewrites rows in
place — an `Overwrite`/`Replace` operation, or a position-delete file — a change
the additive append + equality-delete delta cannot express.
The rule: **refuse the window, emit `functional_status(..., ok = false)`, require
explicit operator action.** Never fall back silently. On a 2 B-row table a silent
fallback turns a streaming job into an 81 GB rewrite at whatever hour the writer
happened to run.
The earlier draft put this check on the `ChangelogBatch` handed to the sink. That
does not work, and the reason is worth stating precisely because it is not
obvious:
> `skade::Table::read_changelog` **already** handles a full-reload window by
> re-reading the whole `to` snapshot as a pure `INSERT` stream
> (`skade/src/read.rs:3045-3062`). By the time a sink sees the batch, the
> expensive read has already happened. A sink-side refusal would refuse to
> *write* an 81 GB read it had already paid for.
The correct place is a **pre-check on `read_delta`**, before `read_changelog` is
called at all:
```
plan = table.read_delta(from, to) # cheap: snapshot-lineage walk only
if plan.needs_full_reload: REFUSE # read.rs:2793 returns (vec![], plan)
# without touching a manifest or a
# parquet file
else: batch = table.read_changelog(from, to)
```
`read_delta` returns early with an empty batch and the flag set, so the pre-check
costs a metadata walk. It also exposes a second hazard: a driver that used
`read_delta` alone on such a window would apply **zero rows** and believe it had
succeeded. The refusal is what turns that silence into an error.
Operator resolution is an explicit one-shot reload, never an automatic one.
## f7 — delivery as a service, without two service definitions
Two audiences:
* **Inside the workspace** this is a declared **Skidbladnir** service, with
`jera` owning the process/VM/container decision.
* **Outside it**, someone who `cargo install`s the bifröst binary has no
Skidbladnir and never will, and needs the binary to install itself.
These do not conflict, and the workspace has already solved it once. From
[`flight-graph-server-design.md`](flight-graph-server-design.md) §N5: the
`knut-flight` service is delivered **two ways from ONE `ComponentSpec`** —
Skidbladnir's `SystemdInstaller` renders the unit, its `KubernetesRenderer`
renders a Deployment from the same spec, *"so the Runtime targets never drift"*.
knut's own seam audit ([`consolidation-seam-eval.md`](consolidation-seam-eval.md)
§2) anticipates exactly this case: *"if knut ever ships its own long-running
service … it should declare a `skidbladnir::ServiceSpec` then."*
So the answer is not "write a unit file inside the binary". It is:
* **One `ServiceSpec`**, declared in bifröst, is the single source of truth.
* `bifrost install` renders that spec through `skidbladnir::SystemdInstaller` —
the same code path Skidbladnir uses. `skidbladnir` is on crates.io (0.1.14), so
an outside user gets the identical renderer, not a copy of it.
* A test asserts the `ExecStart` produced by the two entry points is
byte-identical. The failure mode being designed against is not "two files", it
is "two files that drift".
Install semantics:
* `install` writes the unit **and** a 0600 `EnvironmentFile`, `daemon-reload`,
`enable --now`. Skidbladnir's systemd renderer already routes secrets to a 0600
`EnvironmentFile=` and never inline in the world-readable unit
(`Skidbladnir/src/systemd.rs:511,904`); that policy is inherited, not rewritten.
* `install --dry-run` prints the unit and the env-file *keys* (never values) and
writes nothing.
* `install` is idempotent: same spec ⇒ no-op; changed spec ⇒ rewrite and reload.
* `uninstall` stops, disables and removes the unit. It **must not** delete the
cursor or the config. `uninstall --purge` does, and says what it removed.
Secrets (`DORIS_USER`, `DORIS_PASSWORD`, `MINIO_ACCESS_KEY`, `MINIO_SECRET_KEY`)
come from the environment only, never a flag: flags are in `/proc/<pid>/cmdline`,
readable by any user on the host, and land in shell history and process listings.
No shell anywhere in this path — the installer is Rust, per the zero-shell
charter.
`SIGTERM` finishes the in-flight window, commits the cursor, exits 0. Readiness
is every configured sink's `probe()` passing.
**Status:** f7 is design only. This document specifies the service; the code
landing alongside it is the sink, not the installer.
## f8 — what has to be measured before this is worth deploying
The claim is that incremental delivery beats Doris's own refresh. That is
arithmetic, not a measurement. In this order:
0. **Partition the Iceberg table and try Doris's partitioned MV refresh.** This
is new to the plan and it comes first, because if it works the rest may be
unnecessary for this workload. Doris 4.0.8 supports partitioned refresh over
Iceberg (since 3.1) for a **single** partition column. `bench.doris2g` has no
partition column today. Partitioning it is also what the bench doc already
wants for scan pruning, so the work pays twice.
1. Time a `COMPLETE` MV refresh on the 110 M-row Iceberg table, scale to 2 B.
*(Provenance warning: "110 M-row table" appears only in the earlier draft of
this spec, not in the red bench doc. Confirm it exists before planning around
it.)*
2. Time `read_changelog` over a real append window on the same table.
3. Time a Stream Load of that window into the native table.
(0) or (1) is the number to beat. (2) + (3) is what this costs. **If they are
within an order of magnitude, a scheduled watermark `INSERT ... SELECT` — or a
partitioned MV — wins on operational surface and this daemon should not run.**
That outcome is acceptable and should be reported rather than engineered around.
The sink is still worth having as a seam; running it in production is a separate
decision that these measurements make.
The bencher `knut.doris_stream_load` measures the encode half of (2) + (3)
against a fake transport, so it is tracked continuously; the network half needs
the cluster.
## f10 — status 2026-09-26: built, packaged, and run live
What was design above is now code on branch `iceberg-doris-cdc` (skade side:
`iceberg-s3-cdc`, skade 0.5.4 / skade-aegir 0.1.3 / iceberg-catalog-rest-arrow59
0.9.7, unpublished at the time of writing).
* **The driver** — `delivery::IcebergDelivery`: one Iceberg snapshot per window,
`deliver_window` per window, cursor after ack. The table is found three ways
(`IcebergLocator`): a skade warehouse; an **S3 root with no catalog**
(skade `connect::S3Table`: `version-hint.text`, else the newest
`*.metadata.json`); a **REST catalog**, incl. MinIO AIStor's `/_iceberg`
signed with SigV4 service `s3tables` (skade `IcebergRest.sigv4`). The f6
refusal runs off the snapshot operations in the metadata already loaded; an
overwrite in history is the whole current snapshot on the INITIAL load only.
* **f7 is no longer design only** — `bifrost doris run|install|uninstall|unit|
start|stop|status|k8s` (feature `service`). `install` is skidbladnir's
Lifecycle (systemd); `k8s` renders the same ServiceSpec through skidbladnir's
KubernetesRenderer (StatefulSet + cursor volume + 0600 Secret). A test holds
the unit and the StatefulSet to the same arguments and keeps secrets off both.
Build Thing row `bifrost-doris`; image `containers/bifrost/Containerfile`.
* **Payload** — `LoadFormat::Arrow` (one Arrow IPC stream, `format: arrow`)
beside JSON; a window's chunks load concurrently through gatling io
(`max_parallel_loads`, default 4), acked in chunk order; bodies are `Bytes`.
* **Two facts only a live Doris gave:**
- `Expect: 100-continue` through reqwest/hyper IS accepted by Doris 4.1.3/4.1.4
(it was "genuinely unverified" in Testing below).
- An arrow body without a `columns` header fails when it carries fewer
columns than the table — a key-only DELETE: `insert_range_from … out of
bound (data.size() = 0)` (4.1.4). Every load now names its columns.
**Measured.** Rows land, count checked with a scan-forcing `count(*) where id >= 0`.
| rig | source | rows | format · loads in flight | rows/s |
|---|---|---|---|---|
| red (t14s over WireGuard) | AIStor REST, `bench.smoke10m`, 1 commit | 10 M | arrow · 1 | 157 600 |
| red | same | 10 M | arrow · 8 | 202 600 |
| red | same | 10 M | json · 1 | 78 400 |
| loki rig (draupnir, Doris 4.1.4, BE on NVMe `medium:ssd`) | MinIO, no catalog, 10 commits | 20 M | arrow · 1 / 4 / 8 | 810 000 / 809 000 / 824 000 |
| loki rig | same | 20 M | json · 4 | 734 000 |
Reading: on red the tunnel is the ceiling (json's larger bodies halve the rate);
without it, arrow vs json is ~10 %, and loads-in-flight does nothing on the rig
because a 2 M-row commit is ONE 64 MiB chunk — the time is one Stream Load per
window in the BE (~1.15 s per 2 M rows on one BE). The next lever is smaller
chunks or more BEs, not more client concurrency. Read (skade changelog off
MinIO) is 1.2 s of 24 s.
**Live tests** (`#[ignore]`, arms in `.nornir/testmatrix-arms.json`):
`tests/live_iceberg_doris.rs` — green on red: AIStor REST + SigV4, a new table,
3 commits, replay refused by label, `MATCH_ANY` on the inverted index, and the
same table off S3 with no catalog. `tests/container_rig.rs` — green on loki:
inserts, equality deletes and updates (delete + append) off MinIO into a
unique-key merge-on-write table whose partitions report medium SSD.
## f9 — public surface + feature flags
| item | module | feature | default |
|---|---|---|---|
| `GraphSink`, `FalkorDbSink` | `sink::graph` | — | on (unchanged) |
| `TableSink`, `TableWindow`, `LoadAck`, `AckToken`, `SnapshotWindow`, `deliver_window`, `refuse_full_reload`, `stream_load_label`, `is_valid_doris_label` | `sink::table` | — | on, **no new dependency** |
| `DorisSink`, `DorisConfig`, `StreamLoadTransport`, `StreamLoadRequest/Response` | `sink::doris` | `doris` | **off** |
| `ReqwestTransport` (the real HTTP client) | `sink::doris` | `doris` | off |
`doris = ["dep:reqwest"]`, and `reqwest` is already an optional dependency for
feature `airflow` — enabling `doris` adds no crate to the workspace tree that is
not already reachable. The default build pays nothing.
The transport is a trait, not a concrete client, for one reason: every
correctness property in this document (label derivation, ordering, replay
rejection, full-reload refusal, schema refusal, delete-mode refusal) becomes
testable without a Doris cluster. Only throughput needs the real thing.
## Testing
Per workspace convention every change lands with tests **and** benchers.
* label derivation is stable across runs and collision-free across snapshot
pairs, including the `from = None` and multi-chunk cases
* the cursor is committed **only** after a successful apply — asserted by
observing call order, because this is the data-loss bug
* replaying an identical window is a no-op: duplicate labels are rejected and the
loaded row count is unchanged
* `needs_full_reload` refuses and reports `ok = false`, rather than reloading
* a schema mismatch is caught at `probe()` and refuses to start, not mid-stream
* a `DELETE` row arriving in `append` mode fails the window loudly
* an FE redirect to an unreachable BE is reported as a reachability failure
naming the BE and the timeouts — the measured behaviour is a hang, and the
timeout is what makes it reportable
* `Expect: 100-continue` is present on every PUT, asserted with a one-row window
* every generated label matches Doris's `^[\-_A-Za-z0-9:\p{L}]{1,128}$`
* a `Label Already Exists` verdict inside an HTTP 200 body is read as a duplicate,
not as a successful load
* bencher `knut.doris_stream_load`: rows/s through changelog → encode → load
Duplicate-label rejection, the label regex, the mandatory `Expect` header and
the FE→BE hang have since been **measured** against `red` (Doris 4.0.8) and the
findings are folded in above; the fake transport now encodes the measured
protocol rather than the documented one.
Still **not** covered without a live cluster, stated rather than faked: label
expiry behaviour, `merge_type: DELETE` semantics on a real unique-key table, and
whether `Expect: 100-continue` survives the real HTTP stack to the wire
(`reqwest`/`hyper` may handle that header specially — genuinely unverified).
Those are integration work against `red`.
**The bencher does not currently run.** `xtask/examples/nornir-bench.rs` does not
build on `main`: `nornir 0.5.4` pulls `iceberg-datafusion-arrow58 0.9.7`, which
fails against the resolved `datafusion`
(`unresolved import datafusion::sql::TableReference`). This reproduces with and
without the `doris` feature, so it is pre-existing and out of scope here. The
`knut.doris_stream_load` arm is registered, and the path it measures is covered
by `tests/doris_sink.rs::the_bencher_path_delivers_100k_rows_and_replays_for_free`
so the delivery path is not left unverified while the harness is red.
## See also
* [`cdc-design.md`](cdc-design.md) — the change-feed seam this consumes.
* [`bifrost-design.md`](bifrost-design.md) — the listener, sources and state.
* skade's `cdc-stream-design.md` — the delta primitives, which stay in skade.
* [`doris-sink-guide.md`](doris-sink-guide.md) ·
[`doris-sink-for-idiots.md`](doris-sink-for-idiots.md)