.nornir/doris-sink-guide.md

Doris sink guide — delivering a CDC window into Doris

Class: guide · Function: doris-sink. The what and the why are in doris-sink-design.md; the no-prior-knowledge version is in doris-sink-for-idiots.md.

Everything here is behind the doris cargo feature, which is off by default:

cargo build -p knut-bifrost --features doris,skade
cargo test  -p knut-bifrost --features doris

The 30-second version

use knut_bifrost::sink::doris::{DorisConfig, DorisSink, LoadMode};
use knut_bifrost::sink::table::{SnapshotWindow, TableWindow, deliver_window};

// Credentials come from DORIS_USER / DORIS_PASSWORD. Never from a flag.
let cfg = DorisConfig::from_env("http://172.16.12.13:8030", "fake", "personer2g")?
    .with_mode(LoadMode::Append);
let mut sink = DorisSink::new(cfg);

// Refuses here, before any row moves, if the Arrow schema and the Doris table
// disagree.
sink.probe(&schema).await?;

let window = TableWindow {
    table: "bench.doris2g",
    snapshots: SnapshotWindow { from: Some(1042), to: 1043 },
    upserts: &upserts,
    deletes: &deletes,
};

// The ONLY way to advance the cursor. The load acks first; the cursor moves
// second. There is no call that does it the other way round.
let ack = deliver_window(&mut sink, &mut state, &window).await?;
println!("{} rows under label {}", ack.rows, ack.labels[0]);

upserts and deletes come from bifröst's existing changelog split:

use knut_bifrost::partition_change_rows;

let change = partition_change_rows(&batches)?;   // strips `_change_type`
// change.upserts  <- INSERT / UPDATE_AFTER / folded UPDATE
// change.deletes  <- DELETE   (UPDATE_BEFORE is dropped: its AFTER overwrites)

Before the first window: refuse a full reload

Do not call read_changelog first. On a window that needs a full reload it will happily re-read the entire to snapshot — for bench.doris2g that is 2 000 000 000 rows — and hand you a batch. The check has to come earlier:

let (_, plan) = table.read_delta(from, to).await?;   // metadata walk only
if plan.needs_full_reload {
    // Refuse. Do not load, do not advance the cursor.
    return Err(BifrostError::Other(format!(
        "{table_name}: snapshots {from:?}..{to} need a full reload — refusing"
    )));
}
let batch = table.read_changelog(from, to).await?;

read_delta returns early with no rows when the flag is set, so this pre-check costs a snapshot-lineage walk and nothing else.

DorisSink::apply also refuses a window that arrives already marked, so a caller that skips the pre-check still fails loudly — it just fails after paying for the read, which is why you should not rely on that path.

Modes

mode	Doris table model	what happens to `DELETE` rows
`LoadMode::Append`	duplicate key	asserted absent — the window fails if any appear
`LoadMode::Upsert`	unique key + `__DORIS_DELETE_SIGN__`	applied as deletes

Use Append for an append-only source. It is not merely an optimisation: it turns an unexpected delete into a loud failure instead of a silent divergence between the lake and the engine.

Upsert requires a Unique Key table on the Doris side. Aggregate-key and duplicate-key tables cannot express a row deletion through Stream Load at all.

Labels, and why replays are safe

Every Stream Load carries a label derived from the snapshot pair:

skade-fake:personer2g-1042-1043-0
      ^db  ^table     ^from ^to  ^chunk

The : is not cosmetic. Doris validates the label against ^[\-_A-Za-z0-9:\p{L}]{1,128}$ and a . is rejected outright with [ANALYSIS_ERROR] ... Label format error. Anything else in a database or table name is folded to _, plus a short hash when the fold actually changed something — otherwise personer.2g and personer_2g would share a label, and a shared label is a silently skipped load.

Doris rejects a load whose label it has already seen (Label Already Exists), so re-sending the same window is a no-op. Do not add your own dedup on top.

Three things to know:

The cursor moves only after the load acks. If the process dies in between, the next run recomputes the same window, derives the same labels, and Doris refuses them. Nothing is duplicated and nothing is lost.
Labels expire. A Stream Load label is kept for streaming_label_keep_max_second, default 12 hours (not the 3 days that label_keep_max_second suggests). A replay after that is not deduplicated. This matters only if the loader has been down for longer than the retention and died mid-window; a normal crash replays within seconds.
A cancelled load frees its label immediately. If you cancel a load in Doris, expect the next attempt to succeed rather than being rejected.

Big windows are chunked

A single Stream Load is capped at streaming_load_max_mb (BE config, default 10 GB), and Doris recommends staying well under it — a dropped connection restarts the whole request. The sink splits a window into chunks under a configured byte budget and gives each its own label (…-1042-1043-0, …-1042-1043-1, …).

A LoadAck is issued only when every chunk acked. A window that fails half-way leaves the cursor where it was; the replay re-sends all chunks, the already-loaded ones are rejected by label, and the missing ones go through.

Network: check BE reachability before anything else

Stream Load answers on the FE at :8030 and returns a 307 naming a BE at :8040. The loader must reach both.

If it can reach the FE but not the BE, the load hangs. Measured on Doris 4.0.8: the FE keeps answering /api/health with 200 while the redirected request sits there until it times out. Nothing looks broken. This is why the sink puts a hard connect (10 s) and request (300 s) timeout on every hop — without them a single routing mistake stalls the stream forever with no error. Tune them with DorisConfig::with_timeouts if a chunk legitimately takes longer than five minutes to ingest.

When it trips you get:

doris: the FE redirected this load to http://be-3.internal:8040/... and the
request did not complete — this is a NETWORK REACHABILITY failure. Measured
behaviour: an unreachable BE does not refuse the connection, it HANGS until the
timeout (10s connect / 300s request), while the FE keeps answering /api/health
with 200.

Also worth knowing: the address in the redirect is the BE's configured address. On a cluster with split internal/external addressing that address may not be routable from where you are running. Doris permits PUTing directly to a known-healthy BE, which is the escape hatch.

Where to run it, in order of preference:

On the Doris BE network. The redirect hop stays local.
Its own small host in the dataplattform zone. One extra hop, cleaner lifecycle; better once several tables stream and load is spiky.

Either way it needs: MinIO/S3, the skade catalog, FE :8030, and every BE :8040.

`Expect: 100-continue` — do not remove it

The sink sets Expect: 100-continue on every load. Doris requires it: without it you get {"status":"FAILED","msg":"There is no 100-continue header"}.

The reason this deserves a heading is how it fails. 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 — which looks like corruption long before it looks like a missing header. There is a test that loads a one-row window specifically to hold this in place.

Credentials

From the environment, always:

DORIS_USER
DORIS_PASSWORD
MINIO_ACCESS_KEY
MINIO_SECRET_KEY

DorisConfig::from_env reads them and fails if they are missing. There is no flag that takes a password, and there will not be one: flags are visible in /proc/<pid>/cmdline to every user on the host, and end up in shell history and process listings.

When it refuses to run

refusal	cause	what to do
`probe()` schema mismatch	Arrow schema ≠ Doris table columns	fix one side; do not start a partial load
`needs_full_reload`	an `INSERT OVERWRITE` or position-delete landed in the window	one-shot explicit reload, then resume
`DELETE` row in `Append` mode	the source is not append-only after all	switch the target to a unique-key table and `Upsert`, or fix the writer
unknown `_change_type`	the source emitted a tag neither engine defines	do not ignore it — a dropped delete is the worst failure a feed can have
FE reachable, BE not	network, not credentials — and it hangs until the timeout	fix routing, or `DorisConfig::with_be_url` to target a BE directly
`There is no 100-continue header`	the `Expect` header was stripped somewhere	check any proxy in the path; the sink always sets it
`Label format error`	a label reached Doris with a character outside `^[\-_A-Za-z0-9:\p{L}]{1,128}$`	should be impossible — the builder enforces it and a test asserts it. File it.

All of these are functional_status(..., ok = false) as well as errors, so they show as RED rows in the test matrix rather than only in a log nobody reads.

Verify a stream really is incremental

Watch the structured log. Every round emits bifrost::iceberg delta_read (or delta_read_cdc) with full_reload, snapshots, added_files and rows. If full_reload = true keeps appearing, the source writer is not append-only and the premise of the deployment is wrong — fix that before tuning anything else.

Running the tests

cargo test -p knut-bifrost --lib sink::table                    # label, ordering, refusals (default build)
cargo test -p knut-bifrost --features doris --test doris_sink   # the sink against the fake transport

Neither needs a Doris cluster.

cargo run -p xtask --release --example nornir-bench --features doris would run the knut.doris_stream_load arm, but that target does not build on main for an unrelated reason: nornir 0.5.4 pulls iceberg-datafusion-arrow58 0.9.7, which fails against the resolved datafusion. It fails identically without the doris feature. Until that is fixed, the path the bencher measures is covered by the_bencher_path_delivers_100k_rows_and_replays_for_free in the test file.

What the tests cannot cover — label expiry, merge_type: DELETE on a real unique-key table, and whether Expect: 100-continue survives the real HTTP stack to the wire — is integration work against red, listed as such in the design doc.