crates/knut-bifrost/tests/container_rig.rs
//! **The container rig on loki**: Iceberg on MinIO → bifröst → Doris 4.1 native
//! tables on NVMe, all through draupnir (see `common/rig.rs`).
//!
//! What red cannot give: equality **deletes and updates** (written here through
//! skade's own catalog, whose metadata files sit on MinIO like anyone else's),
//! and a bench with no WireGuard tunnel in the measured path.
//!
//! ```text
//! cargo test -p knut-bifrost --features service --test container_rig -- --ignored --nocapture --test-threads 1
//! RIG_BENCH_ROWS=20000000 … rig_bench # size of the bench table
//! … rig_down # stop + remove the containers (data stays)
//! ```
#![cfg(feature = "service")]
mod common;
use std::time::Instant;
use common::rig::{BUCKET, Rig};
use common::{DorisSql, people, people_ddl, people_schema, run_tag};
use knut_bifrost::delivery::{DeliveryOptions, IcebergDelivery, IcebergLocator};
use knut_bifrost::sink::{DorisConfig, DorisSink, LoadFormat, LoadMode, ReqwestTransport};
use knut_bifrost::state::SyncState;
use skade::connect::S3Table;
/// Doris table properties on the rig: one replica (one BE), on the SSD medium
/// the BE's NVMe storage root is marked as.
const PROPS: &str = r#", "replication_num" = "1", "storage_medium" = "SSD""#;
fn sink(db: &str, table: &str, format: LoadFormat, parallel: usize) -> DorisSink<ReqwestTransport> {
let mut cfg = DorisConfig::from_env(common::rig::FE_HTTP, db, table)
.unwrap()
.with_mode(LoadMode::Upsert)
.with_max_parallel_loads(parallel);
cfg.format = format;
let t = ReqwestTransport::new(&cfg).unwrap();
DorisSink::new(cfg, t)
}
/// A skade warehouse whose catalog is a local redb file and whose data and
/// metadata live on the rig's MinIO.
async fn lake(rig: &Rig) -> skade::Warehouse {
skade::Warehouse::open_with_object_store_config(
rig.dir.join("skade-catalog"),
format!("s3://{BUCKET}/wh"),
skade::ObjectStoreConfig::Aegir(rig.s3()),
)
.await
.expect("open the skade warehouse on MinIO")
}
/// **Inserts, updates and deletes reach Doris, read off S3 with no catalog.**
/// Two appends; an equality delete of 100 ids; an "update" of 100 more (their
/// delete, then their new rows). One window per commit, deletes applied by key,
/// the updated names are the new ones, the deleted ids are gone.
#[tokio::test]
#[ignore = "container rig: needs a podman socket; run on loki"]
async fn rig_cdc_inserts_updates_deletes() {
let rig = Rig::up().await;
rig.export_env();
let sql = DorisSql::from_env(common::rig::FE_HTTP);
let tag = run_tag();
let n: i64 = 100_000;
let wh = lake(&rig).await;
let mut t = wh
.create_table_with_identity(&format!("people_{tag}"), &people_schema(), &["id"])
.await
.expect("create the source table");
t.append(&[people(0..n, "p")]).await.unwrap();
t.append(&[people(n..2 * n, "p")]).await.unwrap();
// delete ids 0..100
t.delete_equality(&people(0..100, "p").project(&[0]).unwrap(), &["id"]).await.unwrap();
// update ids 100..200: their old rows out, new rows in (two commits)
t.delete_equality(&people(100..200, "p").project(&[0]).unwrap(), &["id"]).await.unwrap();
t.append(&[people(100..200, "u")]).await.unwrap();
let location = t.inner().metadata().location().to_string();
eprintln!("source at {location}");
let db = "knut_rig";
let dest = format!("people_{tag}");
sql.exec(&format!("CREATE DATABASE IF NOT EXISTS {db}")).await;
sql.exec(&people_ddl(db, &dest, PROPS)).await;
let dir = tempfile::tempdir().unwrap();
let mut d = IcebergDelivery::new(
IcebergLocator::S3(S3Table { s3: rig.s3(), location }),
&format!("doris:{db}.{dest}"),
sink(db, &dest, LoadFormat::Arrow, 4),
SyncState::open(dir.path().join("c.json")).unwrap(),
DeliveryOptions::default(),
);
let o = d.round().await.expect("deliver");
eprintln!("rig round: {o:?}");
assert_eq!(o.windows, 5, "one window per commit");
assert_eq!(o.rows_deleted, 200, "100 deletes + 100 update pre-images, applied by key");
let table = format!("{db}.{dest}");
assert_eq!(sql.count(&table).await, 2 * n - 100, "100 deleted, 100 updated in place");
assert_eq!(sql.scalar(&format!("select count(*) from {table} where id < 100")).await, 0);
assert_eq!(
sql.scalar(&format!("select count(*) from {table} where id >= 100 and id < 200 and name like 'u%'")).await,
100,
"the updated rows carry the new values"
);
assert_eq!(sql.scalar(&format!("select count(*) from {table} where name MATCH_ANY 'u150'")).await, 1);
// The medium really is SSD: the partition says so.
let parts = sql.exec(&format!("SHOW PARTITIONS FROM {table}")).await;
assert!(
parts.iter().flatten().any(|c| c.as_str() == Some("SSD")),
"the table's partitions are on the SSD medium: {parts:?}"
);
}
/// **Throughput without a tunnel**: a lake table of `RIG_BENCH_ROWS` rows
/// (default 20 M, commits of 2 M) delivered into fresh Doris tables per
/// format × parallel loads. Prints one BENCH line per arm; asserts every row landed.
#[tokio::test]
#[ignore = "container rig bench: run on loki"]
async fn rig_bench() {
let rig = Rig::up().await;
rig.export_env();
let sql = DorisSql::from_env(common::rig::FE_HTTP);
let rows: i64 = std::env::var("RIG_BENCH_ROWS").ok().and_then(|v| v.parse().ok()).unwrap_or(20_000_000);
let per_commit: i64 = 2_000_000;
let tag = run_tag();
let wh = lake(&rig).await;
let mut t = wh.create_table(&format!("bench_{tag}"), &people_schema()).await.unwrap();
let t0 = Instant::now();
let mut at = 0;
while at < rows {
let end = (at + per_commit).min(rows);
t.append(&[people(at..end, "p")]).await.unwrap();
at = end;
}
eprintln!("lake table: {rows} rows in {:.1}s", t0.elapsed().as_secs_f64());
let location = t.inner().metadata().location().to_string();
let db = "knut_rig";
sql.exec(&format!("CREATE DATABASE IF NOT EXISTS {db}")).await;
let arms = std::env::var("RIG_BENCH_ARMS").unwrap_or_else(|_| "arrow:1,arrow:4,arrow:8,json:4".into());
for arm in arms.split(',') {
let (fmt, par) = arm.split_once(':').unwrap();
let par: usize = par.parse().unwrap();
let format = if fmt == "json" { LoadFormat::Json } else { LoadFormat::Arrow };
let dest = format!("bench_{tag}_{fmt}_{par}");
sql.exec(&people_ddl(db, &dest, PROPS)).await;
let dir = tempfile::tempdir().unwrap();
let mut d = IcebergDelivery::new(
IcebergLocator::S3(S3Table { s3: rig.s3(), location: location.clone() }),
&format!("doris:{db}.{dest}"),
sink(db, &dest, format, par),
SyncState::open(dir.path().join("c.json")).unwrap(),
DeliveryOptions::default(),
);
let t0 = Instant::now();
let o = d.round().await.unwrap();
let wall = t0.elapsed();
eprintln!(
"BENCH rig {rows} rows [{fmt}, {par} in flight]: {:.1}s = {:.0} rows/s (read {:.1}s, load {:.1}s, {} windows)",
wall.as_secs_f64(),
o.rows_loaded as f64 / wall.as_secs_f64(),
o.read.as_secs_f64(),
o.load.as_secs_f64(),
o.windows
);
assert_eq!(sql.count(&format!("{db}.{dest}")).await, rows);
}
}
/// Stop and remove the rig's containers (the data directories stay).
#[tokio::test]
#[ignore = "container rig teardown"]
async fn rig_down() {
Rig::down().await;
}