<!-- nornir:class: guide -->
# 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`](doris-sink-design.md); the no-prior-knowledge
> version is in [`doris-sink-for-idiots.md`](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
```rust
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:
```rust
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:
```rust
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:
1. **On the Doris BE network.** The redirect hop stays local.
2. **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.