crates/knut-bifrost/tests/grpc_catalog_delivery.rs
//! **Delivery through a skade-katalog gRPC front** (`--grpc-uri`,
//! `skade::connect::SkadeGrpc`): the Doris delivery loop reading its table over
//! the native catalog wire, with no Doris and no network beyond loopback.
//!
//! * `grpc_delivery_equals_the_embedded_catalog_commit_by_commit` — one skade
//! warehouse, its catalog served in-process by skade-katalog's gRPC front; two
//! deliveries walk the same live history, one through the front and one
//! straight through the embedded catalog (both as `IcebergLocator::Catalog`,
//! i.e. `dyn CatalogConnector`). Every window — cursor pair, sequence number,
//! keys, the very rows — must be the same.
//! * `spark_112_fixture_through_grpc_matches_expected` (`#[ignore]`d, needs
//! `SKADE_SPARK112_FIXTURE`): a table written by Spark 4.1 + Iceberg 1.12 (v3,
//! MERGE INTO → deletion vectors) registered in a skade-katalog and delivered
//! over gRPC, window by window, against the fixture's `expected.json` — the
//! same fixture and layout as skade's `tests/deletion_vectors.rs`.
//!
//! ```text
//! cargo test -p knut-bifrost --features skade-grpc --test grpc_catalog_delivery
//! SKADE_SPARK112_FIXTURE=/path/to/spark112 \
//! cargo test -p knut-bifrost --features skade-grpc --test grpc_catalog_delivery -- --ignored
//! ```
#![cfg(feature = "skade-grpc")]
use std::collections::{BTreeMap, BTreeSet};
use std::sync::{Arc, Mutex};
use arrow_array::{Array, Int64Array, RecordBatch, StringArray};
use arrow_schema::{DataType, Field, Schema};
use async_trait::async_trait;
use knut_bifrost::delivery::{DeliveryOptions, IcebergDelivery, IcebergLocator};
use knut_bifrost::sink::table::{AckToken, LoadAck, TableSink, TableWindow};
use knut_bifrost::state::{SyncState, TableState};
use skade::connect::{CatalogConnector, SkadeGrpc};
use skade::iceberg::{Catalog, NamespaceIdent, TableIdent};
use skade::skade_katalog::RedbCatalog;
/// What a sink saw of one window: the cursor pair, the version, the keys and
/// the identities of the rows (as text, whatever the key's type).
#[derive(Clone, Debug, PartialEq)]
struct Seen {
from: Option<i64>,
to: i64,
sequence: Option<i64>,
keys: Vec<String>,
upserts: Vec<String>,
deletes: Vec<String>,
upsert_rows: Vec<RecordBatch>,
}
/// A sink that records every window, by the identity column `key`.
#[derive(Clone)]
struct Recording {
key: String,
windows: Arc<Mutex<Vec<Seen>>>,
}
impl Recording {
fn new(key: &str) -> Self {
Self { key: key.to_string(), windows: Arc::default() }
}
fn seen(&self) -> Vec<Seen> {
self.windows.lock().unwrap().clone()
}
}
fn ids(batches: &[RecordBatch], key: &str) -> Vec<String> {
let mut out = Vec::new();
for b in batches {
let col = b.column_by_name(key).unwrap_or_else(|| panic!("no {key} in {:?}", b.schema()));
let text = arrow::compute::cast(col, &DataType::Utf8).unwrap();
let text = text.as_any().downcast_ref::<StringArray>().unwrap();
out.extend((0..text.len()).map(|i| text.value(i).to_string()));
}
out.sort();
out
}
#[async_trait]
impl TableSink for Recording {
async fn probe(&mut self, _: &arrow_schema::Schema) -> knut_bifrost::Result<()> {
Ok(())
}
async fn apply(&mut self, w: &TableWindow<'_>, token: AckToken) -> knut_bifrost::Result<LoadAck> {
self.windows.lock().unwrap().push(Seen {
from: w.snapshots.from,
to: w.snapshots.to,
sequence: w.sequence,
keys: w.key_columns.to_vec(),
upserts: ids(w.upserts, &self.key),
deletes: ids(w.deletes, &self.key),
upsert_rows: w.upserts.to_vec(),
});
Ok(token.ack(w.snapshots.to, vec![]).with_counts(w.upsert_rows(), w.delete_rows(), 0))
}
}
/// The embedded catalog itself, as a connector: the door with no wire, which
/// the gRPC door must match.
#[derive(Debug)]
struct Embedded(Arc<RedbCatalog>);
#[async_trait]
impl CatalogConnector for Embedded {
fn kind(&self) -> &'static str {
"test-embedded"
}
fn describe(&self) -> String {
"the embedded catalog, shared in-process".into()
}
fn read_only(&self) -> bool {
true
}
async fn open(&self) -> skade::Result<Arc<dyn Catalog>> {
let cat: Arc<dyn Catalog> = self.0.clone();
Ok(cat)
}
}
fn people(ids: &[i64]) -> RecordBatch {
let schema = Arc::new(Schema::new(vec![
Field::new("id", DataType::Int64, false),
Field::new("name", DataType::Utf8, true),
]));
let names: Vec<String> = ids.iter().map(|i| format!("p{i}")).collect();
RecordBatch::try_new(schema, vec![Arc::new(Int64Array::from(ids.to_vec())), Arc::new(StringArray::from(names))])
.unwrap()
}
fn keys(ids: &[i64]) -> RecordBatch {
let schema = Arc::new(Schema::new(vec![Field::new("id", DataType::Int64, false)]));
RecordBatch::try_new(schema, vec![Arc::new(Int64Array::from(ids.to_vec()))]).unwrap()
}
fn delivery(loc: IcebergLocator, dir: &std::path::Path, name: &str, sink: Recording, key: &str) -> IcebergDelivery<Recording> {
let state = SyncState::open(dir.join(format!("{name}.cursor.json"))).unwrap();
let opts = DeliveryOptions { key_columns: vec![key.to_string()], ..DeliveryOptions::default() };
IcebergDelivery::new(loc, "test:db.people", sink, state, opts)
}
#[tokio::test]
async fn grpc_delivery_equals_the_embedded_catalog_commit_by_commit() {
let dir = tempfile::tempdir().unwrap();
let wh = skade::Warehouse::open(dir.path().join("lake")).await.unwrap();
let mut t = wh.create_table_with_identity("db.people", &people(&[]).schema(), &["id"]).await.unwrap();
t.append(&[people(&[1, 2, 3])]).await.unwrap();
let bound = skade_katalog_grpc_spawn(&wh).await;
let uri = format!("http://{bound}");
let (sa, sb) = (Recording::new("id"), Recording::new("id"));
let mut a = delivery(IcebergLocator::catalog(Embedded(wh.catalog()), "db.people"), dir.path(), "a", sa.clone(), "id");
let mut b = delivery(IcebergLocator::catalog(SkadeGrpc::local_files(uri.clone()), "db.people"), dir.path(), "b", sb.clone(), "id");
assert_eq!(b.cursor_key(), format!("grpc:{uri}#db.people -> test:db.people"));
// Each commit kind a CDC reader meets, delivered through both doors, one
// round per commit: appends, a v3 deletion vector (MERGE INTO), the update
// that re-inserts, an equality delete, a compaction (`replace`), an append.
let first = t.live_data_files().await.unwrap().remove(0);
let commits: Vec<&str> = vec!["initial", "append", "dv", "reinsert", "eq-delete", "replace", "append"];
for (i, what) in commits.iter().enumerate() {
match *what {
"initial" => {}
"append" => t.append(&[people(&[10 + i as i64, 20 + i as i64])]).await.unwrap(),
"dv" => t.delete_positions(&first, &[1]).await.unwrap(), // id 2
"reinsert" => t.append(&[people(&[2])]).await.unwrap(),
"eq-delete" => t.delete_equality(&keys(&[3]), &["id"]).await.unwrap(),
"replace" => {
let live = t.live_data_files().await.unwrap();
let rows = skade::read_all(t.inner()).await.unwrap();
t.rewrite_files(&live, vec![rows], None).await.unwrap();
}
other => unreachable!("{other}"),
}
let (oa, ob) = (a.round().await.unwrap(), b.round().await.unwrap());
assert_eq!((oa.windows, oa.rows_loaded, oa.rows_deleted), (ob.windows, ob.rows_loaded, ob.rows_deleted), "{what}");
assert_eq!(ob.windows, 1, "{what}: one window per commit");
assert_eq!(a.cursor(), b.cursor(), "{what}");
assert_eq!(b.cursor(), t.current_snapshot_id(), "{what}: gRPC delivery caught up");
}
let (wa, wb) = (sa.seen(), sb.seen());
assert_eq!(wb.len(), commits.len());
assert_eq!(wb, wa, "every window through gRPC equals the embedded catalog's");
// And what they carried, spelled out.
assert_eq!((wb[0].from, wb[0].upserts.clone()), (None, vec!["1".into(), "2".into(), "3".into()]));
assert_eq!(wb[2].deletes, ["2"], "the deletion vector is a keyed delete");
assert_eq!(wb[3].upserts, ["2"], "the re-insert");
assert_eq!(wb[4].deletes, ["3"], "the equality delete");
assert!(wb[5].upserts.is_empty() && wb[5].deletes.is_empty(), "a compaction changes no row");
assert!(wb.windows(2).all(|p| p[0].sequence < p[1].sequence), "sequence numbers rise: {wb:?}");
// Keyed by the identity column — except the compaction, which reads nothing.
let keyed: Vec<&Vec<String>> = wb.iter().map(|w| &w.keys).collect();
assert!(wb.iter().enumerate().all(|(i, w)| if i == 5 { w.keys.is_empty() } else { w.keys == ["id"] }), "{keyed:?}");
}
/// The gRPC front over the warehouse's own catalog (the front shares it), on
/// an ephemeral loopback port.
async fn skade_katalog_grpc_spawn(wh: &skade::Warehouse) -> std::net::SocketAddr {
let cat: RedbCatalog = (*wh.catalog()).clone();
skade::skade_katalog_grpc::spawn(cat, "127.0.0.1:0".parse().unwrap()).await.unwrap()
}
/// **Acceptance against a real Spark 4.1 + Iceberg 1.12 table**, read through
/// a skade-katalog gRPC front. The fixture layout is skade's
/// (`skade/skade/tests/deletion_vectors.rs`): `metadata_location` (one line,
/// the current `*.metadata.json`) and `expected.json` (`key`, and per snapshot
/// window `from`/`to`/`changes` as DELETE/INSERT of a key). The fixture is only
/// read: its metadata is registered in a fresh catalog in a temp dir.
///
/// The delivery starts with its cursor on the first window's `to` and must then
/// deliver every later window exactly as `expected.json` says; a second,
/// cursorless delivery must load the net state of all the windows at once.
#[tokio::test]
#[ignore = "needs SKADE_SPARK112_FIXTURE (a Spark 4.1 + Iceberg 1.12 table)"]
async fn spark_112_fixture_through_grpc_matches_expected() {
let root = std::path::PathBuf::from(std::env::var("SKADE_SPARK112_FIXTURE").expect("SKADE_SPARK112_FIXTURE"));
let location = std::fs::read_to_string(root.join("metadata_location")).unwrap().trim().to_string();
let expected: serde_json::Value =
serde_json::from_str(&std::fs::read_to_string(root.join("expected.json")).unwrap()).unwrap();
let key = expected["key"].as_str().unwrap().to_string();
let windows = expected["snapshots"].as_array().unwrap();
let dir = tempfile::tempdir().unwrap();
let wh = skade::Warehouse::open(dir.path().join("katalog")).await.unwrap();
let ns = NamespaceIdent::new("fixture".into());
wh.catalog().create_namespace(&ns, Default::default()).await.unwrap();
wh.catalog().register_table(&TableIdent::new(ns, "t".into()), location.clone()).await.unwrap();
let uri = format!("http://{}", skade_katalog_grpc_spawn(&wh).await);
let loc = || IcebergLocator::catalog(SkadeGrpc::local_files(uri.clone()), "fixture.t");
let table = loc().load().await.unwrap();
assert_eq!(table.metadata_location(), Some(location.as_str()), "the gRPC load is the registered metadata");
assert_eq!(table.metadata().format_version(), skade::iceberg::spec::FormatVersion::V3);
// Cursor on the first window's end, as if the initial load had run there.
let sink = Recording::new(&key);
let mut d = delivery(loc(), dir.path(), "fixture", sink.clone(), &key);
let first_to = windows[0]["to"].as_i64().unwrap();
let mut state = SyncState::open(dir.path().join("fixture.cursor.json")).unwrap();
state.advance(d.cursor_key(), TableState { snapshot_id: Some(first_to), ..TableState::default() }).await.unwrap();
drop(d);
d = delivery(loc(), dir.path(), "fixture", sink.clone(), &key);
assert_eq!(d.cursor(), Some(first_to));
let o = d.round().await.unwrap();
assert_eq!(o.windows, windows.len() - 1);
assert_eq!(d.cursor(), table.metadata().current_snapshot_id());
let seen = sink.seen();
let changes = |w: &serde_json::Value, ty: &str| -> Vec<String> {
let mut v: Vec<String> = w["changes"]
.as_array()
.unwrap()
.iter()
.filter(|c| c["type"] == ty)
.map(|c| c["key"].as_str().unwrap().to_string())
.collect();
v.sort();
v
};
for (w, s) in windows[1..].iter().zip(&seen) {
let (from, to) = (w["from"].as_i64(), w["to"].as_i64().unwrap());
assert_eq!((s.from, s.to), (from, to), "window order");
assert_eq!(s.deletes, changes(w, "DELETE"), "deletes of ({from:?}, {to}]");
assert_eq!(s.upserts, changes(w, "INSERT"), "inserts of ({from:?}, {to}]");
let seq = table.metadata().snapshot_by_id(to).unwrap().sequence_number();
assert_eq!(s.sequence, Some(seq), "the window is versioned by its snapshot's sequence number");
}
// A cursorless delivery: the initial load is the net state of every window.
let mut live: BTreeMap<String, ()> = BTreeMap::new();
for w in windows {
for c in w["changes"].as_array().unwrap() {
let k = c["key"].as_str().unwrap().to_string();
match c["type"].as_str().unwrap() {
"DELETE" => {
live.remove(&k);
}
_ => {
live.insert(k, ());
}
}
}
}
let fresh = Recording::new(&key);
let mut d = delivery(loc(), dir.path(), "fresh", fresh.clone(), &key);
assert_eq!(d.round().await.unwrap().windows, 1);
let initial = fresh.seen().remove(0);
let mut want: Vec<String> = live.into_keys().collect();
want.sort();
assert_eq!(initial.upserts, want, "the initial load is the table's live rows");
assert!(initial.deletes.is_empty());
let distinct: BTreeSet<&String> = initial.upserts.iter().collect();
assert_eq!(distinct.len(), initial.upserts.len(), "no key twice in the initial load");
assert!(initial.upsert_rows.iter().all(|b| b.num_rows() > 0 || initial.upserts.is_empty()));
}