nordisk/knut

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, 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:

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:

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:

#[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.

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:

/// 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

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 installs 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 §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 §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:

1. 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.
2. 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.)
3. Time read_changelog over a real append window on the same table.
4. 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.

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

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 — the change-feed seam this consumes.
• 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-for-idiots.md