crates/knut-bifrost/tests/doris_sink.rs
//! The Doris [`TableSink`] against a fake Stream Load transport.
//!
//! Run with: `cargo test -p knut-bifrost --features doris`.
//!
//! The fake encodes the **documented** Stream Load protocol: a PUT to the FE
//! answers 307 with a `Location` naming a BE; the BE answers a JSON result
//! object; a label already seen comes back `Label Already Exists`. That makes
//! every property below assertable without a Doris cluster — which is the whole
//! reason the transport is a trait. What it cannot prove is that Doris itself
//! behaves this way; that is integration work against a live cluster and is
//! listed as such in `.nornir/doris-sink-design.md`.
#![cfg(feature = "doris")]
use std::collections::{HashMap, HashSet};
use std::sync::{Arc, Mutex};
use arrow_array::{Int64Array, RecordBatch, StringArray};
use arrow_schema::{DataType, Field, Schema};
use async_trait::async_trait;
use knut_bifrost::sink::{
DorisConfig, DorisSink, LoadMode, SnapshotWindow, StreamLoadRequest, StreamLoadResponse,
StreamLoadTransport, TableSink, TableWindow, deliver_window,
};
use knut_bifrost::{Result, SyncState};
// ── the fake ────────────────────────────────────────────────────────────────
#[derive(Debug, Default)]
struct Recorded {
/// Every load request the sink issued, in order: (method, url, label, merge_type).
calls: Vec<(String, String, String, String)>,
/// The headers of every load that reached the BE, in order.
load_headers: Vec<Vec<(String, String)>>,
/// label → rows accepted (committed or not). Second use of a label is rejected.
accepted: HashMap<String, usize>,
/// Labels pre-committed under 2PC and not yet committed: their row ops.
pending: HashMap<String, Vec<RowOp>>,
/// Labels committed through `_stream_load_2pc`, in order.
commits: Vec<String>,
/// Labels aborted through `_stream_load_2pc`, in order.
aborts: Vec<String>,
/// Rows currently "in the table" (visible), by primary key, so a replay can
/// be shown to change nothing.
rows: HashSet<i64>,
/// Per key: the highest sequence value applied (Doris Unique Key sequence
/// column — a lower one never replaces a higher one).
seqs: HashMap<i64, i64>,
}
/// One row's effect: (id, delete?, sequence).
type RowOp = (i64, bool, Option<i64>);
impl Recorded {
fn apply(&mut self, ops: &[RowOp]) {
for &(id, delete, seq) in ops {
if let Some(seq) = seq {
if self.seqs.get(&id).is_some_and(|have| seq < *have) {
continue; // stale version: Doris keeps the newer row
}
self.seqs.insert(id, seq);
}
if delete {
self.rows.remove(&id);
} else {
self.rows.insert(id);
}
}
}
}
struct FakeDoris {
log: Arc<Mutex<Recorded>>,
/// Columns the destination table has.
columns: Vec<&'static str>,
/// BE host the FE redirects to.
be: String,
/// When true, a PUT to the BE fails at the transport layer (unreachable).
be_unreachable: bool,
/// A load whose label ends with this suffix fails (`Status: Fail`).
fail_label_suffix: Option<String>,
/// The next N commits fail (`status: Fail`), then commits work again.
fail_commits: Arc<Mutex<usize>>,
/// Doris 4.1.4.1 as measured: a PUT whose label already exists is answered
/// and the connection closed mid-upload, so the client sees a transport
/// error ("Broken pipe") instead of the `Label Already Exists` reply.
label_exists_breaks_pipe: bool,
/// The next N commits land in Doris but their reply is lost (transport error).
lose_commit_replies: Arc<Mutex<usize>>,
}
impl FakeDoris {
fn new(log: Arc<Mutex<Recorded>>) -> Self {
Self {
log,
columns: vec!["id", "name"],
be: "http://be-3.internal:8040".into(),
be_unreachable: false,
fail_label_suffix: None,
fail_commits: Arc::new(Mutex::new(0)),
label_exists_breaks_pipe: false,
lose_commit_replies: Arc::new(Mutex::new(0)),
}
}
}
/// Pull `(id, op, seq)` out of the body the sink sent, so a test can assert on
/// the *rows*, not just on the call count. `op`/`seq` are the sink's
/// `__bifrost_op` / `__bifrost_seq` load columns when present.
fn rows_in(body: &[u8]) -> Vec<(i64, Option<i64>, Option<i64>)> {
// `format: arrow` bodies are an Arrow IPC stream; decode them for real, so
// the arrow path is held to the same row assertions as the JSON one.
if let Ok(reader) = arrow::ipc::reader::StreamReader::try_new(std::io::Cursor::new(body), None) {
let mut out = Vec::new();
for b in reader {
let b = b.expect("a well-formed IPC batch");
let ids = b.column_by_name("id").expect("id column").as_any().downcast_ref::<Int64Array>().expect("id is Int64").clone();
let int = |name: &str, i: usize| {
b.column_by_name(name).map(|c| {
arrow::compute::cast(c, &DataType::Int64).unwrap().as_any().downcast_ref::<Int64Array>().unwrap().value(i)
})
};
for i in 0..b.num_rows() {
out.push((ids.value(i), int("__bifrost_op", i), int("__bifrost_seq", i)));
}
}
return out;
}
serde_json::from_slice::<serde_json::Value>(body)
.ok()
.and_then(|v| v.as_array().cloned())
.unwrap_or_default()
.iter()
.filter_map(|r| {
let id = r.get("id").and_then(serde_json::Value::as_i64)?;
let op = r.get("__bifrost_op").and_then(serde_json::Value::as_i64);
let seq = r.get("__bifrost_seq").and_then(serde_json::Value::as_i64);
Some((id, op, seq))
})
.collect()
}
fn ids_in(body: &[u8]) -> Vec<i64> {
rows_in(body).into_iter().map(|r| r.0).collect()
}
fn json(v: serde_json::Value) -> Result<StreamLoadResponse> {
Ok(StreamLoadResponse::ok(v.to_string()))
}
#[async_trait]
impl StreamLoadTransport for FakeDoris {
async fn send(&self, req: &StreamLoadRequest) -> Result<StreamLoadResponse> {
if req.method == "GET" && req.url.contains("/get_load_state?label=") {
// FE `GET /api/{db}/get_load_state?label=…` (measured reply shape).
let label = req.url.rsplit("label=").next().unwrap_or_default().replace("%3A", ":");
let log = self.log.lock().unwrap();
let state = if log.pending.contains_key(&label) {
"PRECOMMITTED"
} else if log.accepted.contains_key(&label) {
"VISIBLE"
} else {
"UNKNOWN"
};
return json(serde_json::json!({ "msg": "success", "code": 0, "data": state, "count": 0 }));
}
if req.method == "GET" {
let props: Vec<_> = self
.columns
.iter()
.map(|c| serde_json::json!({ "name": c, "type": "BIGINT" }))
.collect();
return json(serde_json::json!({ "status": 200, "data": { "properties": props } }));
}
let label = req.header("label").unwrap_or_default().to_string();
// Stream Load 2PC (Doris transaction docs): commit/abort by label; the
// FE answers it itself.
if req.url.ends_with("/_stream_load_2pc") {
let op = req.header("txn_operation").unwrap_or_default();
let mut log = self.log.lock().unwrap();
if op == "commit" {
let mut fail = self.fail_commits.lock().unwrap();
if *fail > 0 {
*fail -= 1;
return json(serde_json::json!({ "status": "Fail", "msg": "publish timeout" }));
}
return match log.pending.remove(&label) {
Some(ops) => {
log.apply(&ops);
log.commits.push(label.clone());
let mut lose = self.lose_commit_replies.lock().unwrap();
if *lose > 0 {
*lose -= 1;
return Err(knut_bifrost::BifrostError::Other("doris: commit reply lost (connection reset)".into()));
}
json(serde_json::json!({ "status": "Success", "msg": format!("label [{label}] commit successfully.") }))
}
// Measured on Doris 4.1.4.1: re-committing a VISIBLE label.
None => json(serde_json::json!({
"status": "ANALYSIS_ERROR",
"msg": format!("TStatus: errCode = 2, detailMessage = running transaction not found, label={label}"),
})),
};
}
if !log.pending.contains_key(&label) {
// Measured on Doris 4.1.4.1: aborting a label with no running txn.
return json(serde_json::json!({
"status": "ANALYSIS_ERROR",
"msg": format!("TStatus: errCode = 2, detailMessage = running transaction not found, label={label}"),
}));
}
log.pending.remove(&label);
log.accepted.remove(&label); // an aborted label may be reused
log.aborts.push(label.clone());
return json(serde_json::json!({ "status": "Success", "msg": format!("label [{label}] abort successfully.") }));
}
let merge = req.header("merge_type").unwrap_or_default().to_string();
self.log.lock().unwrap().calls.push((
req.method.to_string(),
req.url.clone(),
label.clone(),
merge.clone(),
));
// The FE hands the load to a BE (documented: 307 + Location).
if !req.url.starts_with(&self.be) {
return Ok(StreamLoadResponse::redirect(format!(
"{}/api/fake/personer2g/_stream_load",
self.be
)));
}
if self.be_unreachable {
return Err(knut_bifrost::BifrostError::Other(
"connection refused".into(),
));
}
let mut log = self.log.lock().unwrap();
log.load_headers.push(req.headers.clone());
if log.accepted.contains_key(&label) {
if self.label_exists_breaks_pipe {
return Err(knut_bifrost::BifrostError::Other(format!(
"doris: PUT {}: error sending request (Broken pipe)",
req.url
)));
}
// Doris: "Data with the same label cannot be imported more than once."
// Measured on 4.1.4.1: an open 2PC chunk reports PRECOMMITTED.
let existing = if log.pending.contains_key(&label) { "PRECOMMITTED" } else { "FINISHED" };
return json(serde_json::json!({
"Status": "Label Already Exists",
"ExistingJobStatus": existing,
"Message": format!("Label [{label}] has already been used."),
}));
}
if self.fail_label_suffix.as_deref().is_some_and(|sfx| label.ends_with(sfx)) {
return json(serde_json::json!({ "Status": "Fail", "Message": "[INTERNAL_ERROR] injected" }));
}
// MERGE deletes the rows matching its `delete` condition (here always
// the sink's `__bifrost_op=1`); DELETE deletes every row; APPEND none.
assert!(
merge != "MERGE" || req.header("delete") == Some("__bifrost_op=1"),
"MERGE must carry the delete condition"
);
let sequenced = req.header("function_column.sequence_col") == Some("__bifrost_seq");
let ops: Vec<RowOp> = rows_in(&req.body)
.into_iter()
.map(|(id, op, seq)| {
let delete = match merge.as_str() {
"DELETE" => true,
"MERGE" => op == Some(1),
_ => false,
};
(id, delete, if sequenced { Some(seq.expect("sequenced load carries __bifrost_seq")) } else { None })
})
.collect();
let n = ops.len();
log.accepted.insert(label.clone(), n);
if req.header("two_phase_commit") == Some("true") {
log.pending.insert(label, ops);
} else {
log.apply(&ops);
}
json(serde_json::json!({
"Status": "Success",
"NumberTotalRows": n,
"NumberLoadedRows": n,
"NumberFilteredRows": 0,
}))
}
}
// ── fixtures ────────────────────────────────────────────────────────────────
fn schema() -> Schema {
Schema::new(vec![
Field::new("id", DataType::Int64, false),
Field::new("name", DataType::Utf8, false),
])
}
fn batch(ids: Vec<i64>) -> RecordBatch {
let n = ids.len();
RecordBatch::try_new(
Arc::new(schema()),
vec![
Arc::new(Int64Array::from(ids)),
Arc::new(StringArray::from(
(0..n).map(|i| format!("kalle-{i}")).collect::<Vec<_>>(),
)),
],
)
.unwrap()
}
fn cfg() -> DorisConfig {
DorisConfig::new("http://fe:8030", "fake", "personer2g", "u", "p")
}
// ── the tests ───────────────────────────────────────────────────────────────
/// **Replaying an identical window is a no-op.** The cursor advanced the first
/// time, but a crash-and-retry (or a manual re-run from the old cursor) derives
/// the same labels, Doris rejects them, and the destination is unchanged.
#[tokio::test]
async fn replaying_an_identical_window_changes_nothing() {
let log = Arc::new(Mutex::new(Recorded::default()));
let dir = tempfile::tempdir().unwrap();
let mut state = SyncState::open(dir.path().join("state.json")).unwrap();
let mut sink = DorisSink::new(cfg(), FakeDoris::new(Arc::clone(&log)));
let ups = [batch(vec![1, 2, 3])];
let w = TableWindow {
table: "bench.doris2g",
snapshots: SnapshotWindow::new(Some(1042), 1043),
upserts: &ups,
deletes: &[],
needs_full_reload: false,
sequence: None,
key_columns: &[],
};
let first = deliver_window(&mut sink, &mut state, &w).await.unwrap();
assert_eq!(first.rows_loaded, 3);
assert_eq!(first.rows_rejected_duplicate, 0);
assert_eq!(first.labels, vec!["skade-fake:personer2g-1042-1043-0"]);
let after_first = log.lock().unwrap().rows.clone();
assert_eq!(after_first, HashSet::from([1, 2, 3]));
// Replay the SAME window, exactly as a crash-restart would.
let second = deliver_window(&mut sink, &mut state, &w).await.unwrap();
assert_eq!(
second.labels, first.labels,
"the label must be derived, not generated"
);
assert_eq!(second.rows_loaded, 0, "nothing new may be written");
assert_eq!(
second.rows_rejected_duplicate, 3,
"the replayed chunk's rows are reported as refused, never as loaded"
);
assert_eq!(
log.lock().unwrap().commits,
vec!["skade-fake:personer2g-1042-1043-0".to_string()],
"2PC: the chunk was committed exactly once; the replay found it FINISHED"
);
assert_eq!(
log.lock().unwrap().rows,
after_first,
"the destination must be byte-identical after a replay"
);
assert_eq!(state.get("bench.doris2g").snapshot_id, Some(1043));
}
/// **The schema mismatch is caught at `probe()`**, before any row moves.
#[tokio::test]
async fn a_schema_mismatch_refuses_to_start() {
let log = Arc::new(Mutex::new(Recorded::default()));
let mut fake = FakeDoris::new(Arc::clone(&log));
fake.columns = vec!["id"]; // the destination has no `name`
let mut sink = DorisSink::new(cfg(), fake);
let err = sink
.probe(&schema())
.await
.expect_err("a missing destination column must refuse the start");
let msg = err.to_string();
assert!(msg.contains("missing column(s)"), "{msg}");
assert!(msg.contains("name"), "{msg}");
assert!(
log.lock().unwrap().calls.is_empty(),
"probe must not issue a load"
);
// The matching case starts clean.
let mut ok_sink = DorisSink::new(
cfg(),
FakeDoris::new(Arc::new(Mutex::new(Recorded::default()))),
);
ok_sink
.probe(&schema())
.await
.expect("matching schema probes clean");
}
/// **Append mode refuses a delete row.** On an append-only source an unexpected
/// delete means the premise of the deployment is wrong.
#[tokio::test]
async fn append_mode_refuses_delete_rows() {
let log = Arc::new(Mutex::new(Recorded::default()));
let dir = tempfile::tempdir().unwrap();
let mut state = SyncState::open(dir.path().join("state.json")).unwrap();
let mut sink = DorisSink::new(
cfg().with_mode(LoadMode::Append),
FakeDoris::new(Arc::clone(&log)),
);
let ups = [batch(vec![1])];
let dels = [batch(vec![2])];
let w = TableWindow {
table: "t",
snapshots: SnapshotWindow::new(Some(1), 2),
upserts: &ups,
deletes: &dels,
needs_full_reload: false,
sequence: None,
key_columns: &[],
};
let err = deliver_window(&mut sink, &mut state, &w).await.unwrap_err();
assert!(err.to_string().contains("not append-only"), "{err}");
assert!(
log.lock().unwrap().calls.is_empty(),
"nothing may be loaded"
);
assert_eq!(state.get("t").snapshot_id, None, "cursor must not move");
}
/// **Upsert mode: one MERGE load per snapshot.** Whole-row deletes ride beside
/// the upserts, tagged `__bifrost_op = 1`, with `merge_type: MERGE` and the
/// `delete: __bifrost_op=1` condition (Doris batch-delete manual, "Load Merge
/// Methods"), every row versioned by the snapshot's sequence number through
/// `function_column.sequence_col`. Pre-committed, then committed (2PC).
#[tokio::test]
async fn upsert_mode_ships_one_sequenced_merge_load() {
let log = Arc::new(Mutex::new(Recorded::default()));
let dir = tempfile::tempdir().unwrap();
let mut state = SyncState::open(dir.path().join("state.json")).unwrap();
let mut sink = DorisSink::new(cfg().with_mode(LoadMode::Upsert), FakeDoris::new(Arc::clone(&log)));
let keys = ["id".to_string()];
// Pre-load rows 1..3 so the delete has something to remove.
let seed = [batch(vec![1, 2, 3])];
let w0 = TableWindow {
table: "t",
snapshots: SnapshotWindow::new(None, 1),
upserts: &seed,
deletes: &[],
needs_full_reload: false,
sequence: Some(1),
key_columns: &keys,
};
deliver_window(&mut sink, &mut state, &w0).await.unwrap();
let ups = [batch(vec![4])];
let dels = [batch(vec![2])];
let w = TableWindow {
table: "t",
snapshots: SnapshotWindow::new(Some(1), 2),
upserts: &ups,
deletes: &dels,
needs_full_reload: false,
sequence: Some(2),
key_columns: &keys,
};
let ack = deliver_window(&mut sink, &mut state, &w).await.unwrap();
assert_eq!((ack.rows_loaded, ack.rows_deleted), (1, 1));
assert_eq!(ack.labels, vec!["skade-fake:personer2g-1-2-0"], "one body, one label");
let log = log.lock().unwrap();
assert_eq!(log.rows, HashSet::from([1, 3, 4]));
assert_eq!(log.seqs.get(&4), Some(&2));
let h = log.load_headers.last().unwrap();
let get = |k: &str| h.iter().find(|(n, _)| n == k).map(|(_, v)| v.as_str());
assert_eq!(get("merge_type"), Some("MERGE"));
assert_eq!(get("delete"), Some("__bifrost_op=1"));
assert_eq!(get("function_column.sequence_col"), Some("__bifrost_seq"));
assert_eq!(get("two_phase_commit"), Some("true"));
assert_eq!(get("columns"), Some("id,name,__bifrost_op,__bifrost_seq"));
assert_eq!(log.commits.len(), 2, "each window's chunk committed once");
}
/// **A key deleted and upserted in the same snapshot is an update: only the
/// upsert is sent.** Inside one load there is no defined order between two rows
/// of one key with the same sequence value.
#[tokio::test]
async fn a_delete_and_upsert_of_one_key_ships_only_the_upsert() {
let log = Arc::new(Mutex::new(Recorded::default()));
let dir = tempfile::tempdir().unwrap();
let mut state = SyncState::open(dir.path().join("state.json")).unwrap();
let mut sink = DorisSink::new(cfg().with_mode(LoadMode::Upsert), FakeDoris::new(Arc::clone(&log)));
let keys = ["id".to_string()];
let seed = [batch(vec![1, 2])];
let w0 = TableWindow {
table: "t",
snapshots: SnapshotWindow::new(None, 1),
upserts: &seed,
deletes: &[],
needs_full_reload: false,
sequence: Some(1),
key_columns: &keys,
};
deliver_window(&mut sink, &mut state, &w0).await.unwrap();
// MERGE INTO … WHEN MATCHED UPDATE: a DV delete of id 2 + an insert of id 2.
let ups = [batch(vec![2])];
let dels = [batch(vec![2, 1])];
let w = TableWindow {
table: "t",
snapshots: SnapshotWindow::new(Some(1), 2),
upserts: &ups,
deletes: &dels,
needs_full_reload: false,
sequence: Some(2),
key_columns: &keys,
};
let ack = deliver_window(&mut sink, &mut state, &w).await.unwrap();
assert_eq!((ack.rows_loaded, ack.rows_deleted), (1, 1), "id 2 upserted, id 1 deleted, id 2's delete dropped");
let body_ids: Vec<i64> = log.lock().unwrap().rows.iter().copied().collect();
assert_eq!(body_ids, vec![2], "id 2 present (updated), id 1 gone");
// Without keys the pair cannot be told from two changes: refused loudly.
let w_nokey = TableWindow { key_columns: &[], snapshots: SnapshotWindow::new(Some(2), 3), sequence: Some(3), ..w };
assert!(deliver_window(&mut sink, &mut state, &w_nokey).await.is_err());
assert_eq!(state.get("t").snapshot_id, Some(2), "cursor held");
}
/// **Identity-only deletes** (equality deletes: non-key columns null) go as a
/// keys-only `merge_type: DELETE` load, so NULLs never reach NOT NULL value
/// columns; still sequenced, still 2PC.
#[tokio::test]
async fn identity_only_deletes_go_as_a_keys_only_delete_load() {
let log = Arc::new(Mutex::new(Recorded::default()));
let dir = tempfile::tempdir().unwrap();
let mut state = SyncState::open(dir.path().join("state.json")).unwrap();
let mut sink = DorisSink::new(cfg().with_mode(LoadMode::Upsert), FakeDoris::new(Arc::clone(&log)));
let keys = ["id".to_string()];
let seed = [batch(vec![1, 2, 3])];
let w0 = TableWindow {
table: "t",
snapshots: SnapshotWindow::new(None, 1),
upserts: &seed,
deletes: &[],
needs_full_reload: false,
sequence: Some(1),
key_columns: &keys,
};
deliver_window(&mut sink, &mut state, &w0).await.unwrap();
let nullable = Arc::new(Schema::new(vec![
Field::new("id", DataType::Int64, false),
Field::new("name", DataType::Utf8, true),
]));
let ident = RecordBatch::try_new(
nullable,
vec![Arc::new(Int64Array::from(vec![3])), Arc::new(StringArray::from(vec![None::<&str>]))],
)
.unwrap();
let dels = [ident];
let w = TableWindow {
table: "t",
snapshots: SnapshotWindow::new(Some(1), 2),
upserts: &[],
deletes: &dels,
needs_full_reload: false,
sequence: Some(2),
key_columns: &keys,
};
let ack = deliver_window(&mut sink, &mut state, &w).await.unwrap();
assert_eq!(ack.rows_deleted, 1);
let log = log.lock().unwrap();
assert_eq!(log.rows, HashSet::from([1, 2]));
let h = log.load_headers.last().unwrap();
let get = |k: &str| h.iter().find(|(n, _)| n == k).map(|(_, v)| v.as_str());
assert_eq!(get("merge_type"), Some("DELETE"));
assert_eq!(get("columns"), Some("id,__bifrost_seq"), "keys only, plus the version");
}
/// **2PC: a failed chunk aborts the chunks this attempt pre-committed**, nothing
/// of the window becomes visible, the cursor stays — and the replay, with the
/// same labels, lands it whole.
#[tokio::test]
async fn a_failed_chunk_aborts_the_window_and_the_replay_lands_it() {
let log = Arc::new(Mutex::new(Recorded::default()));
let dir = tempfile::tempdir().unwrap();
let mut state = SyncState::open(dir.path().join("state.json")).unwrap();
let mut fake = FakeDoris::new(Arc::clone(&log));
fake.fail_label_suffix = Some("-1-2-2".into()); // the third chunk
let mut sink = DorisSink::new(cfg().with_max_chunk_bytes(1).with_max_parallel_loads(1), fake);
let ups: Vec<RecordBatch> = (0..4).map(|i| batch(vec![i])).collect();
let w = TableWindow {
table: "t",
snapshots: SnapshotWindow::new(Some(1), 2),
upserts: &ups,
deletes: &[],
needs_full_reload: false,
sequence: None,
key_columns: &[],
};
assert!(deliver_window(&mut sink, &mut state, &w).await.is_err());
{
let l = log.lock().unwrap();
assert!(l.rows.is_empty(), "nothing of the window is visible");
assert_eq!(l.aborts.len(), 3, "the three pre-committed chunks were aborted: {:?}", l.aborts);
assert!(l.commits.is_empty());
}
assert_eq!(state.get("t").snapshot_id, None, "cursor held");
// The replay: same labels, the injected failure gone.
let mut sink = DorisSink::new(cfg().with_max_chunk_bytes(1), FakeDoris::new(Arc::clone(&log)));
let ack = deliver_window(&mut sink, &mut state, &w).await.unwrap();
assert_eq!(ack.rows_loaded, 4);
assert_eq!(log.lock().unwrap().rows, HashSet::from([0, 1, 2, 3]));
assert_eq!(state.get("t").snapshot_id, Some(2));
}
/// **2PC: a commit that fails leaves the window un-acked; the replay finds the
/// committed chunks FINISHED and the rest pending, and commits those** — no row
/// loaded twice, nothing lost.
#[tokio::test]
async fn a_failed_commit_is_finished_by_the_replay() {
let log = Arc::new(Mutex::new(Recorded::default()));
let dir = tempfile::tempdir().unwrap();
let mut state = SyncState::open(dir.path().join("state.json")).unwrap();
let fake = FakeDoris::new(Arc::clone(&log));
let fail = Arc::clone(&fake.fail_commits);
let mut sink = DorisSink::new(cfg().with_max_chunk_bytes(1).with_max_parallel_loads(1), fake);
let ups: Vec<RecordBatch> = (0..3).map(|i| batch(vec![i])).collect();
let w = TableWindow {
table: "t",
snapshots: SnapshotWindow::new(Some(5), 6),
upserts: &ups,
deletes: &[],
needs_full_reload: false,
sequence: None,
key_columns: &[],
};
*fail.lock().unwrap() = 1; // the very first commit fails
assert!(deliver_window(&mut sink, &mut state, &w).await.is_err());
assert_eq!(state.get("t").snapshot_id, None, "cursor held");
assert!(log.lock().unwrap().rows.is_empty(), "nothing committed yet");
// Replay: every chunk's label exists and is pending → committed by label.
let ack = deliver_window(&mut sink, &mut state, &w).await.unwrap();
assert_eq!(ack.rows_loaded, 0, "no chunk loaded twice");
let l = log.lock().unwrap();
assert_eq!(l.rows, HashSet::from([0, 1, 2]));
assert_eq!(l.commits.len(), 3);
assert_eq!(state.get("t").snapshot_id, Some(6));
}
/// **Doris 4.1.4.1 breaks the pipe on an existing label** (live drill,
/// 2026-10-02): bifrost was killed after pre-committing a window's chunks; on the
/// replay every PUT of an existing label is answered and the connection closed
/// mid-upload, so the client sees a transport error, not the reply. The sink
/// asks the FE (`get_load_state`) what the label holds: PRECOMMITTED → commit it,
/// VISIBLE → already done — and the window lands once, nothing lost.
#[tokio::test]
async fn a_replay_survives_dorises_broken_pipe_on_existing_labels() {
let log = Arc::new(Mutex::new(Recorded::default()));
let dir = tempfile::tempdir().unwrap();
let mut state = SyncState::open(dir.path().join("state.json")).unwrap();
let mut fake = FakeDoris::new(Arc::clone(&log));
fake.label_exists_breaks_pipe = true;
let fail = Arc::clone(&fake.fail_commits);
let mut sink = DorisSink::new(cfg().with_max_chunk_bytes(1).with_max_parallel_loads(1), fake);
let ups: Vec<RecordBatch> = (0..4).map(|i| batch(vec![i])).collect();
let w = TableWindow {
table: "t",
snapshots: SnapshotWindow::new(Some(5), 6),
upserts: &ups,
deletes: &[],
needs_full_reload: false,
sequence: None,
key_columns: &[],
};
// The crash: every chunk pre-committed, the commit never lands.
*fail.lock().unwrap() = 1;
assert!(deliver_window(&mut sink, &mut state, &w).await.is_err());
assert_eq!(state.get("t").snapshot_id, None, "cursor held");
assert!(log.lock().unwrap().rows.is_empty(), "nothing visible yet");
// Replay 1: every label PRECOMMITTED, every PUT a broken pipe → commit by label.
let ack = deliver_window(&mut sink, &mut state, &w).await.unwrap();
assert_eq!(ack.rows_loaded, 0, "no chunk loaded twice");
{
let l = log.lock().unwrap();
assert_eq!(l.rows, HashSet::from([0, 1, 2, 3]), "nothing lost");
assert_eq!(l.commits.len(), 4);
}
// Replay 2 of the same window: every label VISIBLE → already done, no commit.
let ack = deliver_window(&mut sink, &mut state, &w).await.unwrap();
assert_eq!(ack.rows_loaded, 0);
assert_eq!(log.lock().unwrap().commits.len(), 4, "nothing committed twice");
assert_eq!(state.get("t").snapshot_id, Some(6));
}
/// **A commit that lands but whose reply is lost is confirmed by the label's
/// state** (Doris 4.1.4.1 answers a re-commit with "running transaction not
/// found", never "already visible"): the window is acked in the same attempt.
#[tokio::test]
async fn a_lost_commit_reply_is_confirmed_by_the_label_state() {
let log = Arc::new(Mutex::new(Recorded::default()));
let dir = tempfile::tempdir().unwrap();
let mut state = SyncState::open(dir.path().join("state.json")).unwrap();
let fake = FakeDoris::new(Arc::clone(&log));
let lose = Arc::clone(&fake.lose_commit_replies);
let mut sink = DorisSink::new(cfg().with_max_chunk_bytes(1).with_max_parallel_loads(1), fake);
let ups: Vec<RecordBatch> = (0..2).map(|i| batch(vec![i])).collect();
let w = TableWindow {
table: "t",
snapshots: SnapshotWindow::new(Some(1), 2),
upserts: &ups,
deletes: &[],
needs_full_reload: false,
sequence: None,
key_columns: &[],
};
*lose.lock().unwrap() = 1;
let ack = deliver_window(&mut sink, &mut state, &w).await.unwrap();
assert_eq!(ack.rows_loaded, 2);
assert_eq!(log.lock().unwrap().rows, HashSet::from([0, 1]));
assert_eq!(state.get("t").snapshot_id, Some(2));
}
/// **The sequence column keeps a late or replayed window from overwriting newer
/// data**: a delete versioned 5 does not remove a row written at version 7.
#[tokio::test]
async fn an_older_sequence_never_overwrites_a_newer_row() {
let log = Arc::new(Mutex::new(Recorded::default()));
let dir = tempfile::tempdir().unwrap();
let mut state = SyncState::open(dir.path().join("state.json")).unwrap();
let mut sink = DorisSink::new(cfg().with_mode(LoadMode::Upsert), FakeDoris::new(Arc::clone(&log)));
let keys = ["id".to_string()];
let ups = [batch(vec![9])];
let w7 = TableWindow {
table: "t",
snapshots: SnapshotWindow::new(Some(6), 7),
upserts: &ups,
deletes: &[],
needs_full_reload: false,
sequence: Some(7),
key_columns: &keys,
};
deliver_window(&mut sink, &mut state, &w7).await.unwrap();
let dels = [batch(vec![9])];
let w5 = TableWindow {
table: "t2",
snapshots: SnapshotWindow::new(Some(4), 5),
upserts: &[],
deletes: &dels,
needs_full_reload: false,
sequence: Some(5),
key_columns: &keys,
};
deliver_window(&mut sink, &mut state, &w5).await.unwrap();
assert_eq!(log.lock().unwrap().rows, HashSet::from([9]), "the version-5 delete lost to the version-7 row");
}
/// **The FE→BE 307 is followed**, and the BE is the host that actually gets the
/// body.
#[tokio::test]
async fn the_fe_redirect_is_followed_to_the_backend() {
let log = Arc::new(Mutex::new(Recorded::default()));
let dir = tempfile::tempdir().unwrap();
let mut state = SyncState::open(dir.path().join("state.json")).unwrap();
let mut sink = DorisSink::new(cfg(), FakeDoris::new(Arc::clone(&log)));
let ups = [batch(vec![7])];
let w = TableWindow {
table: "t",
snapshots: SnapshotWindow::new(Some(1), 2),
upserts: &ups,
deletes: &[],
needs_full_reload: false,
sequence: None,
key_columns: &[],
};
deliver_window(&mut sink, &mut state, &w).await.unwrap();
let log = log.lock().unwrap();
let urls: Vec<&str> = log.calls.iter().map(|c| c.1.as_str()).collect();
assert_eq!(urls.len(), 2, "one FE hop, one BE hop: {urls:?}");
assert!(
urls[0].starts_with("http://fe:8030/api/fake/personer2g/_stream_load"),
"{urls:?}"
);
assert!(
urls[1].starts_with("http://be-3.internal:8040/"),
"{urls:?}"
);
}
#[tokio::test]
async fn an_unreachable_backend_is_reported_as_a_hang_not_an_auth_error() {
let log = Arc::new(Mutex::new(Recorded::default()));
let mut fake = FakeDoris::new(Arc::clone(&log));
fake.be_unreachable = true;
let dir = tempfile::tempdir().unwrap();
let mut state = SyncState::open(dir.path().join("state.json")).unwrap();
let mut sink = DorisSink::new(cfg(), fake);
let ups = [batch(vec![1])];
let w = TableWindow {
table: "t",
snapshots: SnapshotWindow::new(Some(1), 2),
upserts: &ups,
deletes: &[],
needs_full_reload: false,
sequence: None,
key_columns: &[],
};
let err = deliver_window(&mut sink, &mut state, &w).await.unwrap_err();
let msg = err.to_string();
assert!(msg.contains("NETWORK REACHABILITY"), "{msg}");
assert!(
msg.contains("HANGS"),
"the measured failure mode is a hang: {msg}"
);
assert!(msg.contains("be-3.internal:8040"), "{msg}");
assert!(
msg.contains("/api/health"),
"say why nothing looks wrong: {msg}"
);
assert_eq!(state.get("t").snapshot_id, None, "cursor must not move");
}
/// **A window larger than the body budget is chunked**, each chunk with its own
/// derived label — and every row still arrives exactly once.
#[tokio::test]
async fn a_large_window_is_chunked_with_distinct_labels() {
let log = Arc::new(Mutex::new(Recorded::default()));
let dir = tempfile::tempdir().unwrap();
let mut state = SyncState::open(dir.path().join("state.json")).unwrap();
// A budget small enough to force several chunks out of 400 rows.
let mut sink = DorisSink::new(
cfg().with_max_chunk_bytes(2_000),
FakeDoris::new(Arc::clone(&log)),
);
let ids: Vec<i64> = (0..400).collect();
let ups = [batch(ids.clone())];
let w = TableWindow {
table: "t",
snapshots: SnapshotWindow::new(Some(1), 2),
upserts: &ups,
deletes: &[],
needs_full_reload: false,
sequence: None,
key_columns: &[],
};
let ack = deliver_window(&mut sink, &mut state, &w).await.unwrap();
assert!(
ack.labels.len() > 1,
"expected several chunks, got {}",
ack.labels.len()
);
let unique: HashSet<&String> = ack.labels.iter().collect();
assert_eq!(
unique.len(),
ack.labels.len(),
"chunk labels must be distinct"
);
assert_eq!(ack.rows_loaded, 400, "every row lands exactly once");
assert_eq!(log.lock().unwrap().rows.len(), 400);
// And a replay of the chunked window is still a no-op.
let again = deliver_window(&mut sink, &mut state, &w).await.unwrap();
assert_eq!(again.labels, ack.labels);
assert_eq!(again.rows_loaded, 0);
assert_eq!(log.lock().unwrap().rows.len(), 400);
}
/// **A filtered row fails the load.** A silently dropped row is the worst
/// failure a change feed can have, so `NumberFilteredRows > 0` is an error even
/// though Doris calls the load a success.
#[test]
fn a_filtered_row_fails_the_load() {
let body = serde_json::json!({
"Status": "Success",
"NumberLoadedRows": 9,
"NumberFilteredRows": 1,
"ErrorURL": "http://be:8040/api/_load_error_log?file=x",
})
.to_string();
let err = knut_bifrost::sink::doris::parse_load_result("l", &body).unwrap_err();
assert!(err.to_string().contains("filtered 1 row"), "{err}");
let clean = serde_json::json!({
"Status": "Success", "NumberLoadedRows": 9, "NumberFilteredRows": 0,
})
.to_string();
assert_eq!(
knut_bifrost::sink::doris::parse_load_result("l", &clean).unwrap(),
(9, false)
);
}
/// **`Expect: 100-continue` is on every load.** Measured against the live
/// cluster: without it Doris answers
/// `{"status":"FAILED","msg":"There is no 100-continue header"}`. curl omits the
/// header for small bodies and sends it for large ones, so leaving it out makes
/// SMALL windows fail while large ones succeed — a size-dependent failure that
/// looks like corruption. Assert it on both the FE hop and the BE hop.
///
/// What this cannot prove: that the header survives the real HTTP stack all the
/// way to the wire. That is integration work against a live cluster.
#[tokio::test]
async fn every_load_carries_the_expect_100_continue_header() {
#[derive(Default)]
struct Seen {
puts: usize,
with_expect: usize,
labels_ok: usize,
}
struct Checker(Arc<Mutex<Seen>>, String);
#[async_trait]
impl StreamLoadTransport for Checker {
async fn send(&self, req: &StreamLoadRequest) -> Result<StreamLoadResponse> {
if req.method == "GET" {
return Ok(StreamLoadResponse::ok(
r#"{"status":200,"data":{"properties":[]}}"#,
));
}
let mut s = self.0.lock().unwrap();
s.puts += 1;
if req.header("Expect").map(str::to_ascii_lowercase).as_deref() == Some("100-continue")
{
s.with_expect += 1;
}
// Doris also validates the label on the way in.
if req.header("label").is_some_and(knut_bifrost::sink::is_valid_doris_label) {
s.labels_ok += 1;
}
drop(s);
if !req.url.starts_with(&self.1) {
return Ok(StreamLoadResponse::redirect(format!("{}/x", self.1)));
}
Ok(StreamLoadResponse::ok(
r#"{"Status":"Success","NumberLoadedRows":1,"NumberFilteredRows":0}"#,
))
}
}
let seen = Arc::new(Mutex::new(Seen::default()));
let dir = tempfile::tempdir().unwrap();
let mut state = SyncState::open(dir.path().join("state.json")).unwrap();
let mut sink = DorisSink::new(cfg(), Checker(Arc::clone(&seen), "http://be:8040".into()));
// A deliberately TINY window — the size at which curl would drop the header.
let ups = [batch(vec![1])];
let w = TableWindow {
table: "t",
snapshots: SnapshotWindow::new(Some(1), 2),
upserts: &ups,
deletes: &[],
needs_full_reload: false,
sequence: None,
key_columns: &[],
};
deliver_window(&mut sink, &mut state, &w).await.unwrap();
let s = seen.lock().unwrap();
assert!(
s.puts >= 2,
"expected an FE hop and a BE hop, got {}",
s.puts
);
assert_eq!(
s.with_expect, s.puts,
"every PUT must carry Expect: 100-continue ({}/{} did)",
s.with_expect, s.puts
);
assert_eq!(
s.labels_ok, s.puts,
"every label must satisfy Doris's regex"
);
}
/// **A rejected replay arrives as HTTP 200 with `Status: Label Already Exists`.**
/// Measured wording included. Treating HTTP 200 as success would mistake a
/// rejected replay for a successful load and report rows that never landed.
#[test]
fn label_already_exists_is_parsed_from_a_200_body() {
let body = serde_json::json!({
"Status": "Label Already Exists",
"Message": "[LABEL_ALREADY_EXISTS] ... Label [skade-sltest:t-1042-1043] has \
already been used, relate to txn [862], status [VISIBLE]",
"NumberLoadedRows": 0,
})
.to_string();
let (rows, duplicate) =
knut_bifrost::sink::doris::parse_load_result("skade-sltest:t-1042-1043", &body).unwrap();
assert_eq!(rows, 0, "a rejected replay loaded nothing");
assert!(
duplicate,
"and must be reported as a duplicate, not a success"
);
}
/// **The `knut.doris_stream_load` bencher's code path, as a test.**
///
/// The bencher itself lives in `xtask/examples/nornir-bench.rs`, and that target
/// does **not** build on `main` right now for a reason unrelated to this work:
/// `nornir 0.5.4` pulls `iceberg-datafusion-arrow58 0.9.7`, which fails to
/// compile against the resolved `datafusion`
/// (`unresolved import datafusion::sql::TableReference`). Verified to fail
/// identically with and without the `doris` feature, so it is pre-existing.
///
/// Rather than claim a bencher runs when it cannot, this exercises the exact
/// path the bench measures — 100k rows through encode → chunk → label → load,
/// then a replay — so the delivery path is covered by something green.
#[tokio::test]
async fn the_bencher_path_delivers_100k_rows_and_replays_for_free() {
use std::sync::atomic::{AtomicUsize, Ordering};
struct NullTransport {
bytes: AtomicUsize,
loads: AtomicUsize,
seen: Mutex<HashSet<String>>,
}
#[async_trait]
impl StreamLoadTransport for NullTransport {
async fn send(&self, req: &StreamLoadRequest) -> Result<StreamLoadResponse> {
if req.method == "GET" {
return Ok(StreamLoadResponse::ok(
r#"{"status":200,"data":{"properties":[]}}"#,
));
}
if req.url.ends_with("/_stream_load_2pc") {
return Ok(StreamLoadResponse::ok(r#"{"status":"Success","msg":"ok"}"#));
}
if req.url.starts_with("http://fe") {
return Ok(StreamLoadResponse::redirect(
"http://be:8040/api/b/t/_stream_load",
));
}
let label = req.header("label").unwrap_or_default().to_string();
if !self.seen.lock().unwrap().insert(label) {
return Ok(StreamLoadResponse::ok(
r#"{"Status":"Label Already Exists","ExistingJobStatus":"FINISHED","NumberLoadedRows":0}"#,
));
}
self.bytes.fetch_add(req.body.len(), Ordering::Relaxed);
self.loads.fetch_add(1, Ordering::Relaxed);
let n = ids_in(&req.body).len();
Ok(StreamLoadResponse::ok(format!(
r#"{{"Status":"Success","NumberLoadedRows":{n},"NumberFilteredRows":0}}"#
)))
}
}
const ROWS: usize = 100_000;
let dir = tempfile::tempdir().unwrap();
let mut state = SyncState::open(dir.path().join("state.json")).unwrap();
let mut sink = DorisSink::new(
cfg().with_max_chunk_bytes(8 * 1024 * 1024),
NullTransport {
bytes: AtomicUsize::new(0),
loads: AtomicUsize::new(0),
seen: Mutex::new(HashSet::new()),
},
);
let ups = [batch((0..ROWS as i64).collect())];
let w = TableWindow {
table: "bench.rows",
snapshots: SnapshotWindow::new(Some(1), 2),
upserts: &ups,
deletes: &[],
needs_full_reload: false,
sequence: None,
key_columns: &[],
};
let ack = deliver_window(&mut sink, &mut state, &w).await.unwrap();
assert_eq!(ack.rows_loaded, ROWS);
let replay = deliver_window(&mut sink, &mut state, &w).await.unwrap();
assert_eq!(
replay.labels, ack.labels,
"replay must derive identical labels — that is the whole safety argument"
);
assert_eq!(replay.rows_loaded, 0, "a replay must write nothing");
}
/// **`format: arrow` carries the same rows as JSON, and only JSON gets
/// `strip_outer_array`.** The body is decoded as a real Arrow IPC stream by the
/// fake, so a writer that produced a malformed stream, or dropped a batch,
/// fails the row assertions exactly as a JSON bug would.
#[tokio::test]
async fn an_arrow_window_lands_the_same_rows_as_json() {
let log = Arc::new(Mutex::new(Recorded::default()));
let dir = tempfile::tempdir().unwrap();
let mut state = SyncState::open(dir.path().join("state.json")).unwrap();
let mut c = cfg().with_mode(LoadMode::Upsert);
c.format = knut_bifrost::sink::LoadFormat::Arrow;
let seen = Arc::new(Mutex::new(Vec::<(Option<String>, Option<String>)>::new()));
struct Spy(FakeDoris, Arc<Mutex<Vec<(Option<String>, Option<String>)>>>);
#[async_trait]
impl StreamLoadTransport for Spy {
async fn send(&self, req: &StreamLoadRequest) -> Result<StreamLoadResponse> {
if req.method == "PUT" && req.header("format").is_some() {
assert!(
matches!(req.header("columns"), Some("id,name,__bifrost_op,__bifrost_seq")),
"every load names its body's columns: {:?}",
req.header("columns")
);
self.1.lock().unwrap().push((
req.header("format").map(str::to_owned),
req.header("strip_outer_array").map(str::to_owned),
));
}
self.0.send(req).await
}
}
let mut sink = DorisSink::new(c, Spy(FakeDoris::new(Arc::clone(&log)), Arc::clone(&seen)));
let ups = [batch(vec![1, 2]), batch(vec![3, 4, 5])];
let w = TableWindow {
table: "bench.arrow",
snapshots: SnapshotWindow::new(None, 7),
upserts: &ups,
deletes: &[],
needs_full_reload: false,
sequence: Some(7),
key_columns: &[],
};
let ack = deliver_window(&mut sink, &mut state, &w).await.unwrap();
assert_eq!(ack.rows_loaded, 5);
assert_eq!(log.lock().unwrap().rows, HashSet::from([1, 2, 3, 4, 5]));
let dels = [batch(vec![2, 4])];
let w = TableWindow {
table: "bench.arrow",
snapshots: SnapshotWindow::new(Some(7), 8),
upserts: &[],
deletes: &dels,
needs_full_reload: false,
sequence: Some(8),
key_columns: &[],
};
let ack = deliver_window(&mut sink, &mut state, &w).await.unwrap();
assert_eq!(ack.rows_deleted, 2);
assert_eq!(log.lock().unwrap().rows, HashSet::from([1, 3, 5]));
for (format, strip) in seen.lock().unwrap().iter() {
assert_eq!(format.as_deref(), Some("arrow"));
assert_eq!(strip, &None, "strip_outer_array is a JSON header, never sent with arrow");
}
}
/// **A window's chunks load concurrently, bounded, and ack in chunk order.**
/// A slow fake BE records how many loads are in flight at once: more than one
/// (the chunks overlap), never more than the cap, and the labels in the ack are
/// in chunk order whichever finished first.
#[tokio::test]
async fn chunks_load_concurrently_up_to_the_cap_and_ack_in_order() {
use std::sync::atomic::{AtomicUsize, Ordering};
struct Slow {
inner: FakeDoris,
now: AtomicUsize,
peak: Arc<AtomicUsize>,
}
#[async_trait]
impl StreamLoadTransport for Slow {
async fn send(&self, req: &StreamLoadRequest) -> Result<StreamLoadResponse> {
let on_be = req.method == "PUT" && req.url.starts_with(&self.inner.be);
if on_be {
let n = self.now.fetch_add(1, Ordering::SeqCst) + 1;
self.peak.fetch_max(n, Ordering::SeqCst);
// Later chunks finish FIRST, to prove the ack re-sequences.
let label = req.header("label").unwrap_or_default();
let k: u64 = label.rsplit('-').next().and_then(|s| s.parse().ok()).unwrap_or(0);
tokio::time::sleep(std::time::Duration::from_millis(80u64.saturating_sub(k * 10))).await;
}
let r = self.inner.send(req).await;
if on_be {
self.now.fetch_sub(1, Ordering::SeqCst);
}
r
}
}
let log = Arc::new(Mutex::new(Recorded::default()));
let peak = Arc::new(AtomicUsize::new(0));
let dir = tempfile::tempdir().unwrap();
let mut state = SyncState::open(dir.path().join("state.json")).unwrap();
let cfg = cfg().with_max_chunk_bytes(1).with_max_parallel_loads(3);
let t = Slow { inner: FakeDoris::new(Arc::clone(&log)), now: AtomicUsize::new(0), peak: Arc::clone(&peak) };
let mut sink = DorisSink::new(cfg, t);
// Budget 1 byte: every one-row batch is its own chunk → 8 chunks.
let ups: Vec<RecordBatch> = (0..8).map(|i| batch(vec![i])).collect();
let w = TableWindow {
table: "bench.par",
snapshots: SnapshotWindow::new(Some(1), 2),
upserts: &ups,
deletes: &[],
needs_full_reload: false,
sequence: None,
key_columns: &[],
};
let ack = deliver_window(&mut sink, &mut state, &w).await.unwrap();
assert_eq!(ack.rows_loaded, 8);
let p = peak.load(Ordering::SeqCst);
assert!(p > 1, "chunks overlapped (peak {p})");
assert!(p <= 3, "never more than the cap in flight (peak {p})");
let expected: Vec<String> = (0..8).map(|k| format!("skade-fake:personer2g-1-2-{k}")).collect();
assert_eq!(ack.labels, expected, "labels in chunk order, not completion order");
assert_eq!(log.lock().unwrap().rows, (0..8).collect::<HashSet<i64>>());
}