crates/knut-bifrost/src/cdc/tests.rs
//! RED-when-broken tests for the CDC vocabulary, the renderable row shape and
//! the encoder. The backend-specific SQL is tested in `cdc::spark::tests`; the
//! live insert/update/delete round trip is `tests/cdc_roundtrip.rs`.
use std::sync::Arc;
use arrow_array::{Int64Array, RecordBatch, StringArray, StructArray};
use arrow_schema::{DataType, Field, Schema};
use super::*;
/// A wide changelog batch: `(id, name, country, _change_type)`.
fn wide(rows: &[(i64, &str, &str, &str)]) -> RecordBatch {
RecordBatch::try_new(
Arc::new(Schema::new(vec![
Field::new("id", DataType::Int64, true),
Field::new("name", DataType::Utf8, true),
Field::new("country", DataType::Utf8, true),
Field::new(meta::CHANGE_TYPE, DataType::Utf8, false),
])),
vec![
Arc::new(Int64Array::from(
rows.iter().map(|r| r.0).collect::<Vec<_>>(),
)),
Arc::new(StringArray::from(
rows.iter().map(|r| r.1).collect::<Vec<_>>(),
)),
Arc::new(StringArray::from(
rows.iter().map(|r| r.2).collect::<Vec<_>>(),
)),
Arc::new(StringArray::from(
rows.iter().map(|r| r.3).collect::<Vec<_>>(),
)),
],
)
.unwrap()
}
/// `(op, before.id, after.id)` for every row of a renderable batch.
fn triples(b: &RecordBatch) -> Vec<(String, Option<i64>, Option<i64>)> {
let op = b
.column_by_name(col::OP)
.unwrap()
.as_any()
.downcast_ref::<StringArray>()
.unwrap();
let side = |name: &str| -> Vec<Option<i64>> {
let s = b
.column_by_name(name)
.unwrap()
.as_any()
.downcast_ref::<StructArray>()
.unwrap();
let ids = s
.column_by_name("id")
.unwrap()
.as_any()
.downcast_ref::<Int64Array>()
.unwrap();
(0..s.len())
.map(|i| {
if s.is_null(i) {
None
} else {
Some(ids.value(i))
}
})
.collect()
};
let before = side(col::BEFORE);
let after = side(col::AFTER);
(0..b.num_rows())
.map(|i| (op.value(i).to_string(), before[i], after[i]))
.collect()
}
/// **The vocabulary must span BOTH engines.** Iceberg spells an update
/// pre-image `UPDATE_BEFORE`, Spark 4.2 spells it `update_preimage`; a parser
/// that knows only one silently mis-routes half the change feed.
#[test]
fn parses_both_engine_dialects() {
// Iceberg's create_changelog_view spellings.
assert_eq!(ChangeOp::parse("INSERT"), Some(ChangeOp::Insert));
assert_eq!(ChangeOp::parse("DELETE"), Some(ChangeOp::Delete));
assert_eq!(
ChangeOp::parse("UPDATE_BEFORE"),
Some(ChangeOp::UpdateBefore)
);
assert_eq!(ChangeOp::parse("UPDATE_AFTER"), Some(ChangeOp::UpdateAfter));
// Spark 4.2's Changelog contract spellings.
assert_eq!(ChangeOp::parse("insert"), Some(ChangeOp::Insert));
assert_eq!(ChangeOp::parse("delete"), Some(ChangeOp::Delete));
assert_eq!(
ChangeOp::parse("update_preimage"),
Some(ChangeOp::UpdateBefore)
);
assert_eq!(
ChangeOp::parse("update_postimage"),
Some(ChangeOp::UpdateAfter)
);
// And nothing else — an unknown op must NOT be silently swallowed.
assert_eq!(ChangeOp::parse("upsert"), None);
assert_eq!(ChangeOp::parse(""), None);
crate::testmatrix::emit(
crate::testmatrix::COMPONENT,
"cdc::parses_both_engine_dialects",
true,
"Iceberg UPDATE_BEFORE/AFTER and Spark 4.2 update_preimage/postimage both parse",
);
}
/// The renderable schema is exactly `(op, before, after, snapshot, ts)` and the
/// two image structs are field-identical, so a UI can zip them into a diff.
#[test]
fn renderable_schema_is_the_published_shape() {
let s = change_row_schema(wide(&[(1, "a", "SE", "INSERT")]).schema_ref());
let names: Vec<&str> = s.fields().iter().map(|f| f.name().as_str()).collect();
assert_eq!(
names,
vec![col::OP, col::BEFORE, col::AFTER, col::SNAPSHOT, col::TS]
);
assert!(!s.field(0).is_nullable(), "`op` is never null");
let (DataType::Struct(b), DataType::Struct(a)) = (
s.field(1).data_type().clone(),
s.field(2).data_type().clone(),
) else {
panic!("before/after must be structs");
};
assert_eq!(b, a, "before and after must be field-identical");
// Metadata columns are NOT part of the rendered images.
let inner: Vec<&str> = b.iter().map(|f| f.name().as_str()).collect();
assert_eq!(inner, vec!["id", "name", "country"]);
assert_eq!(
s.field(4).data_type(),
&DataType::Timestamp(arrow_schema::TimeUnit::Microsecond, Some("UTC".into()))
);
crate::testmatrix::emit(
crate::testmatrix::COMPONENT,
"cdc::renderable_schema_is_the_published_shape",
true,
"(op, before, after, snapshot, ts) with field-identical images, no metadata leakage",
);
}
/// **The failure that matters**: insert / update / delete must come out as
/// exactly those ops, in that order, with the right side populated. A feed that
/// drops the delete, or lands the pre-image in `after`, fails here.
#[test]
fn encodes_insert_update_delete_in_order_with_the_right_side_filled() {
let batch = wide(&[
(1, "a", "SE", "INSERT"),
(2, "old", "NO", "UPDATE_BEFORE"),
(2, "new", "NO", "UPDATE_AFTER"),
(3, "c", "DK", "DELETE"),
]);
let out = encode_change_rows(&batch, Some(77)).unwrap();
assert_eq!(out.num_rows(), 4, "one output row per change event");
assert_eq!(
triples(&out),
vec![
("insert".into(), None, Some(1)),
("update_before".into(), Some(2), None),
("update_after".into(), None, Some(2)),
("delete".into(), Some(3), None),
],
);
// The snapshot rides on every row even though the batch carried no
// per-row metadata (skade's cursor is out-of-band).
let snap = out
.column_by_name(col::SNAPSHOT)
.unwrap()
.as_any()
.downcast_ref::<Int64Array>()
.unwrap();
assert!((0..4).all(|i| snap.value(i) == 77));
crate::testmatrix::emit(
crate::testmatrix::COMPONENT,
"cdc::encodes_insert_update_delete_in_order_with_the_right_side_filled",
true,
"insert/update_before/update_after/delete encoded in order, before/after filled correctly",
);
}
/// An unknown or null `_change_type` is an ERROR, never a dropped row.
#[test]
fn refuses_to_drop_an_untypable_change_row() {
let bad = wide(&[(1, "a", "SE", "UPSERT")]);
let err = encode_change_rows(&bad, None).unwrap_err().to_string();
assert!(err.contains("UPSERT"), "{err}");
assert!(err.contains("refusing to drop"), "{err}");
let no_col = RecordBatch::try_new(
Arc::new(Schema::new(vec![Field::new("id", DataType::Int64, false)])),
vec![Arc::new(Int64Array::from(vec![1i64]))],
)
.unwrap();
assert!(
encode_change_rows(&no_col, None).is_err(),
"no _change_type → error"
);
crate::testmatrix::emit(
crate::testmatrix::COMPONENT,
"cdc::refuses_to_drop_an_untypable_change_row",
true,
"unknown/absent _change_type errors instead of silently dropping the row",
);
}
/// Spark 4.2's per-row metadata is honoured: `_commit_version` becomes
/// `snapshot`, `_commit_timestamp` becomes `ts`.
#[test]
fn spark_native_metadata_lands_in_snapshot_and_ts() {
use arrow_array::TimestampMicrosecondArray;
let batch = RecordBatch::try_new(
Arc::new(Schema::new(vec![
Field::new("id", DataType::Int64, true),
Field::new(meta::CHANGE_TYPE, DataType::Utf8, false),
Field::new(meta::COMMIT_VERSION, DataType::Int64, true),
Field::new(
meta::COMMIT_TIMESTAMP,
DataType::Timestamp(arrow_schema::TimeUnit::Microsecond, Some("UTC".into())),
true,
),
])),
vec![
Arc::new(Int64Array::from(vec![1i64, 2])),
// Spark's lower-case dialect.
Arc::new(StringArray::from(vec!["insert", "delete"])),
Arc::new(Int64Array::from(vec![11i64, 12])),
Arc::new(
TimestampMicrosecondArray::from(vec![1_000_000i64, 2_000_000]).with_timezone("UTC"),
),
],
)
.unwrap();
// `default_snapshot` must LOSE to the per-row metadata.
let out = encode_change_rows(&batch, Some(999)).unwrap();
let snap = out
.column_by_name(col::SNAPSHOT)
.unwrap()
.as_any()
.downcast_ref::<Int64Array>()
.unwrap();
assert_eq!(
(snap.value(0), snap.value(1)),
(11, 12),
"_commit_version wins"
);
let ts = out
.column_by_name(col::TS)
.unwrap()
.as_any()
.downcast_ref::<TimestampMicrosecondArray>()
.unwrap();
assert_eq!((ts.value(0), ts.value(1)), (1_000_000, 2_000_000));
assert_eq!(
triples(&out),
vec![
("insert".into(), None, Some(1)),
("delete".into(), Some(2), None),
],
);
crate::testmatrix::emit(
crate::testmatrix::COMPONENT,
"cdc::spark_native_metadata_lands_in_snapshot_and_ts",
true,
"_commit_version → snapshot and _commit_timestamp → ts, overriding the default cursor",
);
}
/// Folding pairs the pre/post-images of one identity into a single `update`
/// row carrying BOTH sides — the diff shape a UI renders — and leaves
/// everything else alone, in order.
#[test]
fn folds_update_pairs_into_one_row_with_both_images() {
let batch = wide(&[
(1, "a", "SE", "INSERT"),
(2, "old", "NO", "UPDATE_BEFORE"),
(2, "new", "NO", "UPDATE_AFTER"),
(3, "c", "DK", "DELETE"),
]);
let enc = encode_change_rows(&batch, Some(5)).unwrap();
let folded = fold_updates(&enc, &["id".to_string()]).unwrap();
assert_eq!(folded.num_rows(), 3, "the update pair became one row");
assert_eq!(
triples(&folded),
vec![
("insert".into(), None, Some(1)),
("update".into(), Some(2), Some(2)),
("delete".into(), Some(3), None),
],
);
// The folded row's two images are the OLD and NEW values, not duplicates.
let names = |name: &str| -> String {
let s = folded
.column_by_name(name)
.unwrap()
.as_any()
.downcast_ref::<StructArray>()
.unwrap();
s.column_by_name("name")
.unwrap()
.as_any()
.downcast_ref::<StringArray>()
.unwrap()
.value(1)
.to_string()
};
assert_eq!(names(col::BEFORE), "old");
assert_eq!(names(col::AFTER), "new");
crate::testmatrix::emit(
crate::testmatrix::COMPONENT,
"cdc::folds_update_pairs_into_one_row_with_both_images",
true,
"update_before+update_after → one `update` row with before=old, after=new",
);
}
/// No identity → no folding, and nothing is lost either way.
#[test]
fn folding_without_identity_is_a_no_op() {
let enc = encode_change_rows(&wide(&[(2, "o", "NO", "UPDATE_BEFORE")]), None).unwrap();
let same = fold_updates(&enc, &[]).unwrap();
assert_eq!(same.num_rows(), enc.num_rows());
assert_eq!(triples(&same), triples(&enc));
}
/// An unmatched pre-image survives folding — a half-pair must never vanish.
#[test]
fn folding_keeps_an_unmatched_preimage() {
let batch = wide(&[(2, "old", "NO", "UPDATE_BEFORE"), (9, "z", "FI", "INSERT")]);
let enc = encode_change_rows(&batch, Some(1)).unwrap();
let folded = fold_updates(&enc, &["id".to_string()]).unwrap();
assert_eq!(
triples(&folded),
vec![
("update_before".into(), Some(2), None),
("insert".into(), None, Some(9)),
],
);
}
/// The post-filter drops non-matching rows and — critically — **keeps** a row
/// whose filter column is null, because that is how a widened `DELETE` looks
/// and dropping it would silently lose a deletion.
#[test]
fn post_filter_prunes_but_never_loses_a_delete() {
use crate::mapping::RowFilter;
let batch = RecordBatch::try_new(
Arc::new(Schema::new(vec![
Field::new("id", DataType::Int64, true),
Field::new("country", DataType::Utf8, true),
Field::new(meta::CHANGE_TYPE, DataType::Utf8, false),
])),
vec![
Arc::new(Int64Array::from(vec![1i64, 2, 3])),
Arc::new(StringArray::from(vec![Some("SE"), Some("NO"), None])),
Arc::new(StringArray::from(vec!["INSERT", "INSERT", "DELETE"])),
],
)
.unwrap();
let kept = post_filter(&batch, &RowFilter::eq("country", "SE")).unwrap();
assert_eq!(kept.num_rows(), 2, "SE insert + the null-country DELETE");
let enc = encode_change_rows(&kept, Some(1)).unwrap();
assert_eq!(
triples(&enc),
vec![
("insert".into(), None, Some(1)),
("delete".into(), Some(3), None)
],
);
// A filter on a column the projection dropped is an ERROR, not a silent
// pass-through of everything.
assert!(post_filter(&batch, &RowFilter::eq("missing", "x")).is_err());
crate::testmatrix::emit(
crate::testmatrix::COMPONENT,
"cdc::post_filter_prunes_but_never_loses_a_delete",
true,
"post-filter drops non-matching inserts, keeps a null-column DELETE, errors on a missing column",
);
}
/// **The pushdown-equivalence law.** 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. Proven here at the encoder, and again end-to-end
/// against a real Iceberg table in `tests/cdc_roundtrip.rs`.
#[test]
fn projected_encode_equals_full_width_on_the_shared_columns() {
let batch = wide(&[
(1, "a", "SE", "INSERT"),
(2, "old", "NO", "UPDATE_BEFORE"),
(2, "new", "NO", "UPDATE_AFTER"),
(3, "c", "DK", "DELETE"),
]);
let full = encode_change_rows(&batch, Some(7)).unwrap();
// The "projected" batch: only `id` + the metadata column survive.
let projected_wide = batch
.project(&[
batch.schema().index_of("id").unwrap(),
batch.schema().index_of(meta::CHANGE_TYPE).unwrap(),
])
.unwrap();
let projected = encode_change_rows(&projected_wide, Some(7)).unwrap();
assert_eq!(projected.num_rows(), full.num_rows(), "same row count");
assert_eq!(
triples(&projected),
triples(&full),
"same ops, same ids, same order"
);
// Byte-identity on the overlapping column: `before.id` / `after.id`.
for side in [col::BEFORE, col::AFTER] {
let get = |b: &RecordBatch| {
b.column_by_name(side)
.unwrap()
.as_any()
.downcast_ref::<StructArray>()
.unwrap()
.column_by_name("id")
.unwrap()
.clone()
};
assert_eq!(
get(&projected).to_data(),
get(&full).to_data(),
"{side}.id must be byte-identical between the projected and full-width reads",
);
}
// And the projection really did narrow the rendered image.
let width = |b: &RecordBatch| match b.schema().field(1).data_type() {
DataType::Struct(f) => f.len(),
_ => unreachable!(),
};
assert_eq!((width(&projected), width(&full)), (1, 3));
crate::testmatrix::emit(
crate::testmatrix::COMPONENT,
"cdc::projected_encode_equals_full_width_on_the_shared_columns",
true,
"projected CDC rows == full-width CDC rows (ops, order, byte-identical id images)",
);
}
/// The capability's `state_json` projection names the backend, the version and
/// — when it degraded — that it degraded, and why. Silence is the bug.
#[test]
fn capability_json_says_which_path_ran_and_whether_it_fell_back() {
let cap = CdcCapability {
chosen: CdcKind::IcebergChangelogView,
server_version: Some("4.2.0".into()),
server_major_minor: Some((4, 2)),
native_expected: true,
native_usable: false,
fell_back: true,
reason: "CHANGES did not resolve for `lake.a.b`".into(),
};
let j = cap.to_json();
assert_eq!(j["cdc_backend"], "iceberg_changelog_view");
assert_eq!(j["server_version"], "4.2.0");
assert_eq!(j["server_major_minor"], "4.2");
assert_eq!(j["native_expected"], true);
assert_eq!(j["native_usable"], false);
assert_eq!(j["fell_back"], true);
assert!(j["reason"].as_str().unwrap().contains("did not resolve"));
// Round-trips, so a listener can persist and re-read it.
let back: CdcCapability = serde_json::from_str(&serde_json::to_string(&cap).unwrap()).unwrap();
assert_eq!(back, cap);
crate::testmatrix::emit(
crate::testmatrix::COMPONENT,
"cdc::capability_json_says_which_path_ran_and_whether_it_fell_back",
true,
"state_json carries backend + version + fell_back + reason, and round-trips",
);
}
/// Version parsing across the shapes a Connect server actually returns.
#[test]
fn parses_the_version_strings_spark_really_sends() {
assert_eq!(parse_major_minor("4.2.0"), Some((4, 2)));
assert_eq!(parse_major_minor("4.1.2"), Some((4, 1)));
assert_eq!(parse_major_minor("4.2.0-SNAPSHOT"), Some((4, 2)));
assert_eq!(parse_major_minor("3.5.7"), Some((3, 5)));
assert_eq!(parse_major_minor("garbage"), None);
assert_eq!(parse_major_minor(""), None);
}
/// The stream really is incremental: batches arrive one at a time and the
/// consumer sees each before the feed ends.
#[tokio::test]
async fn change_stream_delivers_batch_by_batch() {
let batch = encode_change_rows(&wide(&[(1, "a", "SE", "INSERT")]), Some(1)).unwrap();
let schema = batch.schema();
let (sink, mut stream) = ChangeStream::channel(schema.clone(), CdcCapability::skade(), 1);
let producer = tokio::spawn(async move {
for _ in 0..3 {
if !sink.send(Ok(batch.clone())).await {
break;
}
}
});
let mut seen = 0;
while let Some(b) = stream.next().await {
assert_eq!(b.unwrap().num_rows(), 1);
seen += 1;
}
producer.await.unwrap();
assert_eq!(seen, 3, "every batch arrived");
assert_eq!(stream.capability.chosen, CdcKind::SkadeDelta);
crate::testmatrix::emit(
crate::testmatrix::COMPONENT,
"cdc::change_stream_delivers_batch_by_batch",
seen == 3,
"ChangeStream delivered 3 batches incrementally over a 1-slot channel",
);
}
/// An empty window still carries the renderable schema, so a UI can bind
/// columns before the first change arrives.
#[tokio::test]
async fn empty_feed_still_has_the_renderable_schema() {
let s = change_row_schema(&Schema::empty());
let stream = ChangeStream::ready(s, CdcCapability::skade(), Vec::new());
let names: Vec<String> = stream
.schema
.fields()
.iter()
.map(|f| f.name().clone())
.collect();
assert_eq!(
names,
vec![col::OP, col::BEFORE, col::AFTER, col::SNAPSHOT, col::TS]
);
assert!(stream.collect().await.unwrap().is_empty());
}
/// **The two null policies must differ in exactly one way, and that way must
/// keep deletes.** `cdc::post_filter` is `mapping::apply_row_filter_with(…,
/// FilterNulls::Keep)` — the SAME predicate implementation the `Full` /
/// `Incremental` paths use, with the one policy change a changelog needs.
///
/// If `post_filter` ever reverts to the scan-planner default, the DELETE row
/// below (whose `country` is null, exactly as skade widens an equality-delete)
/// silently disappears and this goes red.
#[test]
fn post_filter_keeps_unjudgeable_rows_where_the_scan_planner_would_drop_them() {
use crate::mapping::{FilterNulls, RowFilter, apply_row_filter_with};
let batch = RecordBatch::try_new(
Arc::new(Schema::new(vec![
Field::new("id", DataType::Int64, true),
Field::new("country", DataType::Utf8, true),
Field::new(meta::CHANGE_TYPE, DataType::Utf8, false),
])),
vec![
Arc::new(Int64Array::from(vec![1i64, 2, 3, 4])),
Arc::new(StringArray::from(vec![
Some("SE"),
Some("NO"),
None, // a widened DELETE row: every non-identity column is null
Some("SE"),
])),
Arc::new(StringArray::from(vec![
"INSERT", "INSERT", "DELETE", "INSERT",
])),
],
)
.unwrap();
let f = RowFilter::eq("country", "SE");
let ids = |b: &RecordBatch| -> Vec<i64> {
let c = b
.column_by_name("id")
.unwrap()
.as_any()
.downcast_ref::<Int64Array>()
.unwrap();
(0..b.num_rows()).map(|i| c.value(i)).collect()
};
// Scan-planner semantics: NULL fails the predicate, the DELETE is GONE.
let dropped =
apply_row_filter_with(std::slice::from_ref(&batch), &f, FilterNulls::Drop).unwrap();
assert_eq!(
ids(&dropped[0]),
vec![1, 4],
"Drop loses the delete — by design"
);
// Changelog semantics: the DELETE survives.
let kept = post_filter(&batch, &f).unwrap();
assert_eq!(
ids(&kept),
vec![1, 3, 4],
"post_filter must keep the unjudgeable DELETE row (id=3)",
);
// …and it is still tagged DELETE, so the listener still removes the node.
let ct = kept
.column_by_name(meta::CHANGE_TYPE)
.unwrap()
.as_any()
.downcast_ref::<StringArray>()
.unwrap();
assert_eq!(ct.value(1), "DELETE");
// Order is preserved — a change feed that reorders is a change feed that lies.
assert!(ids(&kept).windows(2).all(|w| w[0] < w[1]));
// Everything the predicate CAN judge is judged identically by both policies.
assert_eq!(
ids(&dropped[0]),
ids(&kept)
.into_iter()
.filter(|i| *i != 3)
.collect::<Vec<_>>(),
);
crate::testmatrix::emit(
crate::testmatrix::COMPONENT,
"cdc::post_filter_keeps_unjudgeable_rows_where_the_scan_planner_would_drop_them",
true,
"FilterNulls::Drop loses the null-column DELETE; cdc::post_filter (Keep) does not",
);
}
/// The predicate is one implementation, so its type discipline is inherited:
/// a literal that is not comparable with the column is REFUSED, not coerced.
#[test]
fn an_uncomparable_literal_is_refused_not_coerced() {
use crate::mapping::RowFilter;
let batch = RecordBatch::try_new(
Arc::new(Schema::new(vec![
Field::new("id", DataType::Int64, true),
Field::new(meta::CHANGE_TYPE, DataType::Utf8, false),
])),
vec![
Arc::new(Int64Array::from(vec![1i64, 2])),
Arc::new(StringArray::from(vec!["INSERT", "INSERT"])),
],
)
.unwrap();
let err = post_filter(&batch, &RowFilter::eq("id", "not-a-number"))
.unwrap_err()
.to_string();
assert!(err.contains("refusing to coerce"), "{err}");
// And a column that is not there at all is an error, not "match nothing".
assert!(post_filter(&batch, &RowFilter::eq("missing", 1i64)).is_err());
}