crates/knut-bifrost/tests/cdc_roundtrip.rs
//! **The CDC feed, end to end, against a real Iceberg table.**
//!
//! Insert, update and delete rows through `skade` (real Iceberg v2 snapshots:
//! appends and equality-delete files), then read them back through the
//! [`CdcBackend`] seam and assert:
//!
//! 1. the feed reports **exactly** those ops, in snapshot order — a feed that
//! silently drops a delete fails here (RED-when-broken);
//! 2. the **projected** CDC read returns the **same change rows** as the
//! full-width one, byte-identical in the columns that overlap;
//! 3. the renderable rows carry the published `(op, before, after, snapshot,
//! ts)` shape a UI binds to, with `before`/`after` on the right side.
//!
//! No container, no Spark: an embedded skade warehouse in a tempdir.
#![cfg(feature = "skade")]
use std::sync::Arc;
use arrow_array::{Array, Int64Array, RecordBatch, StringArray, StructArray};
use knut_bifrost::cdc::{CdcBackend, CdcKind, ChangeRequest, SkadeCdc, col};
use knut_bifrost::mapping::{RowFilter, SyncMode};
use knut_bifrost::source::{IcebergSource, TableSource, TableSpec};
use knut_bifrost::state::TableState;
use skade::arrow_array::{Int64Array as SkInt64, RecordBatch as SkBatch, StringArray as SkString};
use skade::arrow_schema::{DataType, Field, Schema};
const TABLE: &str = "people";
fn people(rows: &[(i64, &str, &str)]) -> SkBatch {
SkBatch::try_new(
Arc::new(Schema::new(vec![
Field::new("id", DataType::Int64, false),
Field::new("name", DataType::Utf8, false),
Field::new("country", DataType::Utf8, false),
])),
vec![
Arc::new(SkInt64::from(rows.iter().map(|r| r.0).collect::<Vec<_>>())),
Arc::new(SkString::from(rows.iter().map(|r| r.1).collect::<Vec<_>>())),
Arc::new(SkString::from(rows.iter().map(|r| r.2).collect::<Vec<_>>())),
],
)
.unwrap()
}
/// Append `rows` as one new Iceberg snapshot; return its snapshot id.
async fn insert(dir: &std::path::Path, rows: &[(i64, &str, &str)]) -> i64 {
let wh = skade::Warehouse::open(dir).await.unwrap();
let batch = people(rows);
let mut t = wh.table_or_create(TABLE, &batch.schema()).await.unwrap();
t.append(&[batch]).await.unwrap();
t.current_snapshot_id().unwrap()
}
/// Remove `ids` via a real Iceberg **equality-delete** snapshot.
async fn delete(dir: &std::path::Path, ids: &[i64]) -> i64 {
let wh = skade::Warehouse::open(dir).await.unwrap();
let mut t = wh.table(TABLE).await.unwrap();
let key = SkBatch::try_new(
Arc::new(Schema::new(vec![Field::new("id", DataType::Int64, false)])),
vec![Arc::new(SkInt64::from(ids.to_vec()))],
)
.unwrap();
t.delete_equality(&key, &["id"]).await.unwrap();
t.current_snapshot_id().unwrap()
}
/// `(op, id, name)` of every change row in the renderable feed. `id`/`name` are
/// read from whichever image the op populates — which is itself part of what
/// this asserts.
fn events(batches: &[RecordBatch]) -> Vec<(String, i64, Option<String>)> {
let mut out = Vec::new();
for b in batches {
let op = b
.column_by_name(col::OP)
.expect("op column")
.as_any()
.downcast_ref::<StringArray>()
.expect("op is Utf8");
let side = |name: &str| -> &StructArray {
b.column_by_name(name)
.expect("image column")
.as_any()
.downcast_ref::<StructArray>()
.expect("image is a struct")
};
let before = side(col::BEFORE);
let after = side(col::AFTER);
for r in 0..b.num_rows() {
let img = if after.is_valid(r) { after } else { before };
assert!(
after.is_valid(r) || before.is_valid(r),
"every change row must populate at least one image",
);
let id = img
.column_by_name("id")
.expect("id in the image")
.as_any()
.downcast_ref::<Int64Array>()
.unwrap()
.value(r);
let name = img.column_by_name("name").map(|c| {
c.as_any()
.downcast_ref::<StringArray>()
.unwrap()
.value(r)
.to_string()
});
out.push((op.value(r).to_string(), id, name));
}
}
out
}
/// Sorted `(op, id)` — the window's content, order-insensitive. Used where the
/// engine's within-window ordering (inserts before deletes) is not a contract.
fn ops_ids(batches: &[RecordBatch]) -> Vec<(String, i64)> {
let mut v: Vec<(String, i64)> = events(batches)
.into_iter()
.map(|(o, i, _)| (o, i))
.collect();
v.sort();
v
}
/// **RED-when-broken.** Insert / update / delete against a real Iceberg table;
/// the CDC feed must report exactly those ops, window by window, in order. If a
/// delete is ever dropped — the failure that matters — the assertion on window 3
/// fails.
#[tokio::test]
async fn cdc_feed_reports_insert_update_delete_in_snapshot_order() {
let dir = tempfile::tempdir().unwrap();
let cdc = SkadeCdc::new(dir.path());
assert_eq!(cdc.kind(), CdcKind::SkadeDelta);
// ── snapshot A: INSERT three rows ────────────────────────────────────────
let a = insert(
dir.path(),
&[(1, "ada", "SE"), (2, "bo", "NO"), (3, "cai", "DK")],
)
.await;
let w1 = cdc
.changes(&ChangeRequest::new(TABLE, None).to(Some(a)))
.await
.unwrap()
.collect()
.await
.unwrap();
assert_eq!(
ops_ids(&w1),
vec![
("insert".into(), 1),
("insert".into(), 2),
("insert".into(), 3),
],
"window 1 is three inserts",
);
// ── snapshot B, C: UPDATE id=2 (Iceberg v2 = equality-delete + append) ───
let b = delete(dir.path(), &[2]).await;
let c = insert(dir.path(), &[(2, "bo-renamed", "NO")]).await;
let w2 = cdc
.changes(&ChangeRequest::new(TABLE, Some(a)).to(Some(c)))
.await
.unwrap()
.collect()
.await
.unwrap();
assert_eq!(
ops_ids(&w2),
vec![("delete".into(), 2), ("insert".into(), 2)],
"window 2 is the update, as Iceberg records it: delete then re-insert of id=2",
);
// The re-inserted image carries the NEW value — an update that reported the
// old name would be a stale feed.
let renamed = events(&w2)
.into_iter()
.find(|(op, id, _)| op == "insert" && *id == 2)
.expect("the re-insert of id=2");
assert_eq!(renamed.2.as_deref(), Some("bo-renamed"));
assert_ne!(b, c, "the update really was two snapshots");
// ── snapshot D: DELETE id=3 ─────────────────────────────────────────────
let d = delete(dir.path(), &[3]).await;
let w3 = cdc
.changes(&ChangeRequest::new(TABLE, Some(c)).to(Some(d)))
.await
.unwrap()
.collect()
.await
.unwrap();
assert_eq!(
ops_ids(&w3),
vec![("delete".into(), 3)],
"window 3 is exactly one delete — a feed that drops it fails HERE",
);
// …and it is on the `before` side, with `after` null. A delete whose payload
// landed in `after` would render as a resurrection.
let del = &w3[0];
let before = del
.column_by_name(col::BEFORE)
.unwrap()
.as_any()
.downcast_ref::<StructArray>()
.unwrap();
let after = del
.column_by_name(col::AFTER)
.unwrap()
.as_any()
.downcast_ref::<StructArray>()
.unwrap();
assert!(before.is_valid(0), "a delete carries a pre-image");
assert!(after.is_null(0), "a delete carries NO post-image");
// The snapshot column is populated on every row, so a UI can group by commit.
for w in [&w1, &w2, &w3] {
for batch in w.iter() {
let s = batch
.column_by_name(col::SNAPSHOT)
.unwrap()
.as_any()
.downcast_ref::<Int64Array>()
.unwrap();
assert!(
(0..batch.num_rows()).all(|i| s.is_valid(i)),
"snapshot is set"
);
}
}
// A caught-up cursor yields an empty feed, not a replay.
let idle = cdc
.changes(&ChangeRequest::new(TABLE, Some(d)).to(Some(d)))
.await
.unwrap()
.collect()
.await
.unwrap();
assert_eq!(events(&idle), vec![], "caught-up cursor → nothing");
}
/// **The pushdown-equivalence law, end to end.** A projected CDC read must
/// return the SAME change rows as the full-width one — same ops, same order,
/// byte-identical in the columns that overlap. If the projection ever dropped,
/// reordered or duplicated a change row, this fails.
#[tokio::test]
async fn projected_cdc_read_equals_full_width_cdc_read() {
let dir = tempfile::tempdir().unwrap();
let cdc = SkadeCdc::new(dir.path());
let a = insert(
dir.path(),
&[
(1, "ada", "SE"),
(2, "bo", "NO"),
(3, "cai", "DK"),
(4, "dag", "SE"),
],
)
.await;
delete(dir.path(), &[2]).await;
insert(dir.path(), &[(2, "bo2", "NO")]).await;
let d = delete(dir.path(), &[4]).await;
let window = |cols: Vec<String>| {
ChangeRequest::new(TABLE, Some(a))
.to(Some(d))
.with_columns(cols)
};
let full = cdc
.changes(&window(vec![]))
.await
.unwrap()
.collect()
.await
.unwrap();
let projected = cdc
.changes(&window(vec!["id".into()]))
.await
.unwrap()
.collect()
.await
.unwrap();
// Same events, same order.
let strip = |v: Vec<(String, i64, Option<String>)>| -> Vec<(String, i64)> {
v.into_iter().map(|(o, i, _)| (o, i)).collect()
};
assert_eq!(
strip(events(&projected)),
strip(events(&full)),
"the projection must not change WHICH changes are reported, nor their order",
);
assert!(!events(&full).is_empty(), "the window is non-empty");
// BYTE-IDENTICAL on the overlapping column.
let image_ids = |batches: &[RecordBatch], side: &str| -> Vec<arrow::array::ArrayData> {
batches
.iter()
.map(|b| {
b.column_by_name(side)
.unwrap()
.as_any()
.downcast_ref::<StructArray>()
.unwrap()
.column_by_name("id")
.unwrap()
.to_data()
})
.collect()
};
for side in [col::BEFORE, col::AFTER] {
assert_eq!(
image_ids(&projected, side),
image_ids(&full, side),
"{side}.id must be byte-identical between the projected and full-width CDC reads",
);
}
// And the projection really narrowed the rendered image: 1 column vs 3.
let width = |batches: &[RecordBatch]| -> usize {
match batches[0].schema().field(1).data_type() {
arrow_schema::DataType::Struct(f) => f.len(),
_ => unreachable!(),
}
};
assert_eq!(width(&full), 3, "full-width feed renders id+name+country");
assert_eq!(width(&projected), 1, "projected feed renders id only");
assert_ne!(
width(&full),
width(&projected),
"the pushdown did something"
);
}
/// A predicate on the CDC feed prunes the change rows — and still never loses a
/// delete, because a widened `DELETE` row's non-identity columns are null and
/// an unjudgeable row is kept.
#[tokio::test]
async fn filtered_cdc_read_prunes_rows_but_keeps_every_delete() {
let dir = tempfile::tempdir().unwrap();
let cdc = SkadeCdc::new(dir.path());
insert(
dir.path(),
&[
(1, "ada", "SE"),
(2, "bo", "NO"),
(3, "cai", "SE"),
(4, "dag", "DK"),
],
)
.await;
let b = delete(dir.path(), &[4]).await;
let unfiltered = cdc
.changes(&ChangeRequest::new(TABLE, None).to(Some(b)))
.await
.unwrap()
.collect()
.await
.unwrap();
let filtered = cdc
.changes(
&ChangeRequest::new(TABLE, None)
.to(Some(b))
.with_columns(vec!["id".into(), "name".into()])
.with_filter(Some(RowFilter::eq("country", "SE"))),
)
.await
.unwrap()
.collect()
.await
.unwrap();
assert_eq!(
ops_ids(&unfiltered),
vec![
("delete".into(), 4),
("insert".into(), 1),
("insert".into(), 2),
("insert".into(), 3),
("insert".into(), 4),
],
);
assert_eq!(
ops_ids(&filtered),
vec![
// the DELETE survives: its `country` is null, so it is not judged
("delete".into(), 4),
("insert".into(), 1),
("insert".into(), 3),
],
"the predicate dropped the NO/DK inserts and kept every delete",
);
}
/// The `SyncMode::Cdc` **source** path (what the listener actually drives) is
/// projected too — this is the gap that was open: `Full` had pushdown,
/// `Incremental`/`Cdc` did not. Same change rows, narrower batches.
#[tokio::test]
async fn source_cdc_mode_is_projected_and_agrees_with_the_full_width_read() {
let dir = tempfile::tempdir().unwrap();
let a = insert(dir.path(), &[(1, "ada", "SE"), (2, "bo", "NO")]).await;
let src = IcebergSource::new(dir.path());
// Seed the cursor at A, then make a change window with a real delete.
let seeded = TableState {
snapshot_id: Some(a),
..Default::default()
};
delete(dir.path(), &[2]).await;
let full_spec = TableSpec::new(TABLE, SyncMode::Cdc);
let proj_spec = TableSpec::new(TABLE, SyncMode::Cdc).with_columns(vec!["id".into()]);
let (full, s1) = src.fetch_new(&full_spec, &seeded).await.unwrap().unwrap();
let (proj, s2) = src.fetch_new(&proj_spec, &seeded).await.unwrap().unwrap();
assert_eq!(s1.snapshot_id, s2.snapshot_id, "same cursor advance");
let rows = |bs: &[RecordBatch]| bs.iter().map(|b| b.num_rows()).sum::<usize>();
assert_eq!(
rows(&full),
rows(&proj),
"the projection drops columns, not rows"
);
assert_eq!(rows(&full), 1, "the window is exactly the delete of id=2");
let width = |bs: &[RecordBatch]| bs.first().map(|b| b.num_columns()).unwrap_or(0);
// `_change_type` always survives (delete routing depends on it), so the
// projected batch is `id` + `_change_type` against `id,name,country,_change_type`.
assert_eq!(width(&full), 4, "full-width CDC batch");
assert_eq!(width(&proj), 2, "projected CDC batch: id + _change_type");
// The `_change_type` column is byte-identical — the routing decision is the
// same on both paths.
let ct = |bs: &[RecordBatch]| {
let b = &bs[0];
b.column(b.schema().index_of(SyncMode::CHANGE_TYPE_COLUMN).unwrap())
.to_data()
};
assert_eq!(ct(&full), ct(&proj), "_change_type must be byte-identical");
// And the ids agree.
let ids = |bs: &[RecordBatch]| {
let b = &bs[0];
b.column(b.schema().index_of("id").unwrap()).to_data()
};
assert_eq!(ids(&full), ids(&proj), "id must be byte-identical");
}