crates/knut-bifrost/tests/live_iceberg_doris.rs
//! **Live: an Iceberg table on MinIO/AIStor into Apache Doris native tables.**
//!
//! `#[ignore]`d and env-driven, because it needs a real object store and a real
//! Doris. Two rigs run it:
//!
//! * **red** (t14s over WireGuard): MinIO AIStor's Iceberg REST catalog
//! (`/_iceberg`, SigV4 `s3tables`) and red's Doris 4.1 on the hosts;
//! * **the container rig** (loki): community MinIO + Doris 4.1 FE/BE through
//! draupnir — `container_rig.rs` sets the same env and calls in here.
//!
//! ```text
//! LIVE_S3_ENDPOINT=https://172.16.12.9:9000 LIVE_S3_CA=~/.red-ca/red-mock-root.crt
//! LIVE_REST_URI=https://172.16.12.9:9000/_iceberg LIVE_REST_WAREHOUSE=red LIVE_SIGV4=s3tables
//! LIVE_DORIS_FE=http://172.16.12.8:8030 [LIVE_DORIS_PROPS=',"replication_num"="1"']
//! MINIO_ACCESS_KEY=… MINIO_SECRET_KEY=… DORIS_USER=… DORIS_PASSWORD=…
//! cargo test -p knut-bifrost --features service --test live_iceberg_doris -- --ignored --nocapture
//! ```
//!
//! Nothing is dropped or deleted afterwards: every run writes a fresh
//! `knut_cdc.people_r<unix>` (MinIO data is never deleted, by standing rule).
#![cfg(feature = "service")]
mod common;
use std::path::PathBuf;
use std::time::Instant;
use common::{DorisSql, people, people_ddl, 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::{CatalogConnector, IcebergRest, S3Table, SigV4};
use skade::iceberg::{NamespaceIdent, TableCreation};
struct Live {
s3: skade::S3Config,
rest_uri: String,
rest_warehouse: String,
sigv4: Option<String>,
doris_fe: String,
doris_props: String,
}
fn env(k: &str) -> Option<String> {
std::env::var(k).ok().filter(|v| !v.is_empty())
}
impl Live {
/// `None` (and the test says why) when the rig is not configured.
fn from_env() -> Option<Self> {
let Some(endpoint) = env("LIVE_S3_ENDPOINT") else {
eprintln!("SKIP: LIVE_S3_ENDPOINT is not set — no live rig configured");
return None;
};
let rest_warehouse = env("LIVE_REST_WAREHOUSE").unwrap_or_else(|| "red".into());
Some(Self {
s3: skade::S3Config {
endpoint,
region: env("LIVE_S3_REGION").unwrap_or_else(|| "us-east-1".into()),
bucket: env("LIVE_S3_BUCKET").unwrap_or_else(|| rest_warehouse.clone()),
access_key_id: env("MINIO_ACCESS_KEY").expect("MINIO_ACCESS_KEY"),
secret_access_key: env("MINIO_SECRET_KEY").expect("MINIO_SECRET_KEY"),
path_style: true,
tls: env("LIVE_S3_CA").map(|ca| skade::tls::ClientTls { ca_pem: Some(PathBuf::from(ca)), ..Default::default() }),
},
rest_uri: env("LIVE_REST_URI").unwrap_or_default(),
rest_warehouse,
sigv4: env("LIVE_SIGV4"),
doris_fe: env("LIVE_DORIS_FE").expect("LIVE_DORIS_FE"),
doris_props: env("LIVE_DORIS_PROPS").unwrap_or_default(),
})
}
fn rest(&self, read_only: bool) -> IcebergRest {
IcebergRest {
uri: self.rest_uri.clone(),
warehouse: Some(self.rest_warehouse.clone()),
prefix_or_ref: None,
token: None,
tls: self.s3.tls.clone(),
s3: self.s3.clone(),
read_only,
sigv4: self.sigv4.clone().map(|service| SigV4 { service }),
..Default::default()
}
}
fn sink(&self, db: &str, table: &str, format: LoadFormat) -> DorisSink<ReqwestTransport> {
// 2026-10-02: Unique Key + sequence column (the DDL below) + 2PC, keyed on
// the first column — the shape the delivery now loads by default.
let mut cfg = DorisConfig::from_env(&self.doris_fe, db, table)
.unwrap()
.with_mode(LoadMode::Upsert)
.with_key_columns(vec!["id".to_string()]); // common::people_schema's key
if let Some(n) = env("LIVE_PARALLEL_LOADS").and_then(|n| n.parse().ok()) {
cfg = cfg.with_max_parallel_loads(n);
}
cfg.format = format;
let t = ReqwestTransport::new(&cfg).unwrap();
DorisSink::new(cfg, t)
}
}
/// **The whole road, twice.** A new AIStor table gets two commits; bifröst
/// delivers it through the REST catalog into one Doris table and straight off
/// S3 (no catalog) into another; a third commit arrives and is delivered; a
/// replay from an old cursor is rejected by label and changes nothing; and the
/// inverted index answers on the delivered rows.
#[tokio::test]
#[ignore = "live: needs MinIO/AIStor + Doris (see module docs)"]
async fn aistor_table_into_doris_through_the_catalog_and_without_it() {
let Some(live) = Live::from_env() else { return };
let tag = run_tag();
let ns = "knut_cdc";
let tname = format!("people_{tag}");
let n: i64 = 50_000;
// ── the source: a new table in the lake, two commits ──
let cat = live.rest(false).open().await.expect("open the REST catalog (SigV4)");
let nsid = NamespaceIdent::new(ns.into());
if !cat.namespace_exists(&nsid).await.unwrap() {
cat.create_namespace(&nsid, Default::default()).await.unwrap();
}
let schema = skade::arrow_to_iceberg(&common::people_schema()).unwrap();
let t = cat
.create_table(&nsid, TableCreation::builder().name(tname.clone()).schema(schema).build())
.await
.expect("create the source table");
let t = skade::append(&*cat, &t, &[people(0..n, "p")]).await.unwrap();
let t = skade::append(&*cat, &t, &[people(n..2 * n, "p")]).await.unwrap();
let location = t.metadata().location().to_string();
eprintln!("source {ns}.{tname} at {location}");
// ── the destinations: native Doris tables with an inverted index ──
let sql = DorisSql::from_env(&live.doris_fe);
sql.exec(&format!("CREATE DATABASE IF NOT EXISTS {ns}")).await;
let via_rest = format!("people_{tag}_rest");
let via_s3 = format!("people_{tag}_s3");
for d in [&via_rest, &via_s3] {
sql.exec(&people_ddl(ns, d, &live.doris_props)).await;
}
let dir = tempfile::tempdir().unwrap();
let cursor = dir.path().join("cursor.json");
// ── through the catalog ──
let rest_loc = IcebergLocator::catalog(live.rest(true), format!("{ns}.{tname}"));
let mut d = IcebergDelivery::new(
rest_loc.clone(),
&format!("doris:{ns}.{via_rest}"),
live.sink(ns, &via_rest, LoadFormat::Arrow),
SyncState::open(&cursor).unwrap(),
DeliveryOptions::default(),
);
let o = d.round().await.expect("deliver through the catalog");
eprintln!("rest round: {o:?}");
assert_eq!(o.windows, 2, "one window per commit");
assert_eq!(o.rows_loaded as i64, 2 * n);
assert_eq!(sql.count(&format!("{ns}.{via_rest}")).await, 2 * n);
let after_two = d.cursor().unwrap();
// A third commit, delivered on the next round.
let _t = skade::append(&*cat, &t, &[people(2 * n..3 * n, "p")]).await.unwrap();
let o = d.round().await.unwrap();
assert_eq!((o.windows, o.rows_loaded as i64), (1, n));
assert_eq!(sql.count(&format!("{ns}.{via_rest}")).await, 3 * n);
// A replay from the old cursor: the label is the same, Doris refuses it,
// nothing changes.
drop(d);
let key = format!("{} -> doris:{ns}.{via_rest}", rest_loc.source_id());
let mut st = SyncState::open(&cursor).unwrap();
let mut back = st.get(&key);
back.snapshot_id = Some(after_two);
st.advance(&key, back).await.unwrap();
st.flush().unwrap();
let mut d = IcebergDelivery::new(
rest_loc,
&format!("doris:{ns}.{via_rest}"),
live.sink(ns, &via_rest, LoadFormat::Arrow),
SyncState::open(&cursor).unwrap(),
DeliveryOptions::default(),
);
let o = d.round().await.unwrap();
eprintln!("replay round: {o:?}");
assert_eq!(o.rows_loaded, 0, "a replayed window loads nothing");
assert_eq!(sql.count(&format!("{ns}.{via_rest}")).await, 3 * n);
// The index the lake cannot give: a MATCH on the inverted index.
let hit = sql
.scalar(&format!("select count(*) from {ns}.{via_rest} where name MATCH_ANY 'p4242'"))
.await;
assert_eq!(hit, 1);
// ── without a catalog: the same table off S3 by its root ──
let s3_loc = IcebergLocator::S3(S3Table { s3: live.s3.clone(), location: location.clone() });
let mut d = IcebergDelivery::new(
s3_loc,
&format!("doris:{ns}.{via_s3}"),
live.sink(ns, &via_s3, LoadFormat::Json),
SyncState::open(dir.path().join("cursor-s3.json")).unwrap(),
DeliveryOptions::default(),
);
let o = d.round().await.expect("deliver off S3 with no catalog");
eprintln!("s3 round: {o:?}");
assert_eq!(o.windows, 3);
assert_eq!(sql.count(&format!("{ns}.{via_s3}")).await, 3 * n);
}
/// **Throughput on a real table**: the initial load of an existing lake table
/// (default `bench.smoke10m`, 10 M rows) into a fresh Doris table, once per
/// format. Numbers are printed; the assertion is only that every row landed.
/// `LIVE_BENCH_TABLE=ns.table`, `LIVE_BENCH_FORMATS=arrow,json`.
#[tokio::test]
#[ignore = "live bench: needs MinIO/AIStor + Doris (see module docs)"]
async fn bench_initial_load_of_a_lake_table() {
let Some(live) = Live::from_env() else { return };
let src = env("LIVE_BENCH_TABLE").unwrap_or_else(|| "bench.smoke10m".into());
let formats = env("LIVE_BENCH_FORMATS").unwrap_or_else(|| "arrow,json".into());
let loc = IcebergLocator::catalog(live.rest(true), src.clone());
let ice = loc.load().await.unwrap();
let expected = skade::scan_count(&ice).await.unwrap() as i64;
let arrow_schema = skade::arrow_schema_of(&ice).unwrap();
let sql = DorisSql::from_env(&live.doris_fe);
let db = "knut_cdc";
sql.exec(&format!("CREATE DATABASE IF NOT EXISTS {db}")).await;
let tag = run_tag();
for fmt in formats.split(',') {
let format = match fmt.trim() {
"json" => LoadFormat::Json,
"arrow" => LoadFormat::Arrow,
other => panic!("format {other}"),
};
let dest = format!("{}_{tag}_{}", src.replace('.', "_"), fmt.trim());
sql.exec(&ddl_for(db, &dest, &arrow_schema, &live.doris_props)).await;
let dir = tempfile::tempdir().unwrap();
let mut d = IcebergDelivery::new(
loc.clone(),
&format!("doris:{db}.{dest}"),
live.sink(db, &dest, format),
SyncState::open(dir.path().join("c.json")).unwrap(),
DeliveryOptions::default(),
);
let t0 = Instant::now();
let o = d.round().await.unwrap();
let wall = t0.elapsed();
let landed = sql.count(&format!("{db}.{dest}")).await;
eprintln!(
"BENCH {src} -> doris {db}.{dest} [{fmt}]: {} rows in {:.1}s = {:.0} rows/s (read {:.1}s, load {:.1}s, {} window(s))",
o.rows_loaded,
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!(landed, expected, "{fmt}: every row landed");
}
}
/// A duplicate-key-free unique table for an arbitrary source: first column is
/// the key. Only the types the bench tables use are mapped.
fn ddl_for(db: &str, table: &str, schema: &arrow_schema::Schema, props: &str) -> String {
use arrow_schema::DataType as T;
let cols: Vec<String> = schema
.fields()
.iter()
.enumerate()
.map(|(i, f)| {
let ty = match f.data_type() {
T::Int64 => "BIGINT",
T::Int32 => "INT",
T::Float64 => "DOUBLE",
T::Boolean => "BOOLEAN",
T::Utf8 | T::LargeUtf8 | T::Utf8View => "STRING",
T::Date32 => "DATE",
T::Timestamp(..) => "DATETIME(6)",
other => panic!("no Doris type mapped for {other}"),
};
let null = if i == 0 { " NOT NULL" } else { "" };
format!("`{}` {ty}{null}", f.name())
})
.collect();
let key = schema.field(0).name();
format!(
"CREATE TABLE {db}.{table} ({}) UNIQUE KEY(`{key}`) DISTRIBUTED BY HASH(`{key}`) BUCKETS 8 \
PROPERTIES (\"enable_unique_key_merge_on_write\" = \"true\", \
\"function_column.sequence_type\" = \"BIGINT\"{props})",
cols.join(", ")
)
}