knut-bifrost 0.3.0

//! 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());
}