knut-bifrost 0.3.0

#![cfg(test)]

use std::sync::{Arc, Mutex};

use arrow_array::{Int64Array, RecordBatch, StringArray};
use arrow_schema::{DataType, Field, Schema};
use async_trait::async_trait;
use tempfile::tempdir;

use crate::{
    error::Result,
    listener::Listener,
    mapping::{MappingSpec, SyncMode},
    mutation::{GraphMutation, SinkReport, VertexDelete, VertexSet},
    registry::TransformRegistry,
    sink::GraphSink,
    source::{TableSource, TableSpec},
    state::{SyncState, TableState},
    transform::{
        EdgeTransform, EndpointSpec, TableContext, Transform, VertexTransform,
        deletes_from_change_rows, is_changelog_batch, partition_change_rows,
    },
};

fn users_batch() -> RecordBatch {
    RecordBatch::try_new(
        Arc::new(Schema::new(vec![
            Field::new("user_id", DataType::Int64, false),
            Field::new("name", DataType::Utf8, false),
            Field::new("city", DataType::Utf8, false),
            Field::new("company_id", DataType::Int64, false),
        ])),
        vec![
            Arc::new(Int64Array::from(vec![1, 2])),
            Arc::new(StringArray::from(vec!["alice", "bob"])),
            Arc::new(StringArray::from(vec!["oslo", "bergen"])),
            Arc::new(Int64Array::from(vec![10, 10])),
        ],
    )
    .unwrap()
}

const MAPPING_YAML: &str = r#"
version: 1
tables:
  - table: lake.social.users
    mode: incremental
    pipeline:
      - vertex:
          label: Person
          id_column: user_id
          properties: [name, city]
          rename: { city: location }
      - fk_edges:
          type: WORKS_AT
          src: { label: Person,  column: user_id }
          dst: { label: Company, column: company_id, id_column: company_id }
          properties: [city]
          rename: { city: based_in }
"#;

// ── transforms ──────────────────────────────────────────────────────────────

#[test]
fn vertex_transform_projects_and_renames() {
    let t = VertexTransform {
        label: "Person".into(),
        id_column: "user_id".into(),
        properties: vec!["name".into(), "city".into()],
        rename: [("city".to_string(), "location".to_string())].into(),
    };
    let ctx = TableContext { table: "t".into() };
    let m = t.apply(&ctx, &[users_batch()]).unwrap();
    assert_eq!(m.vertices.len(), 1);
    assert!(m.edges.is_empty());
    let set = &m.vertices[0];
    assert_eq!(set.label, "Person");
    assert_eq!(set.id_column, "user_id");
    let names: Vec<_> = set
        .batch
        .schema()
        .fields()
        .iter()
        .map(|f| f.name().clone())
        .collect();
    assert_eq!(names, vec!["user_id", "name", "location"]);
    assert_eq!(set.batch.num_rows(), 2);
    crate::testmatrix::emit(
        crate::testmatrix::COMPONENT,
        "vertex_transform_projects_and_renames",
        set.label == "Person"
            && set.batch.num_rows() == 2
            && names == ["user_id", "name", "location"],
        &format!(
            "label={}, cols={:?}, rows={}",
            set.label,
            names,
            set.batch.num_rows()
        ),
    );
}

#[test]
fn vertex_transform_missing_column_errors() {
    let t = VertexTransform {
        label: "Person".into(),
        id_column: "nope".into(),
        properties: vec![],
        rename: Default::default(),
    };
    let ctx = TableContext { table: "t".into() };
    let res = t.apply(&ctx, &[users_batch()]);
    assert!(res.is_err());
    crate::testmatrix::emit(
        crate::testmatrix::COMPONENT,
        "vertex_transform_missing_column_errors",
        res.is_err(),
        "missing id column → Err",
    );
}

#[test]
fn edge_transform_maps_endpoints_to_src_dst() {
    let t = EdgeTransform {
        edge_type: "WORKS_AT".into(),
        src: EndpointSpec {
            label: "Person".into(),
            column: "user_id".into(),
            id_column: None,
        },
        dst: EndpointSpec {
            label: "Company".into(),
            column: "company_id".into(),
            id_column: None,
        },
        properties: vec![],
        rename: Default::default(),
        key: Vec::new(),
        replace: None,
    };
    let ctx = TableContext { table: "t".into() };
    let m = t.apply(&ctx, &[users_batch()]).unwrap();
    assert_eq!(m.edges.len(), 1);
    let set = &m.edges[0];
    assert_eq!(set.src_id_column, "user_id");
    assert_eq!(set.dst_id_column, "company_id");
    let names: Vec<_> = set
        .batch
        .schema()
        .fields()
        .iter()
        .map(|f| f.name().clone())
        .collect();
    assert_eq!(names, vec!["src", "dst"]);
    crate::testmatrix::emit(
        crate::testmatrix::COMPONENT,
        "edge_transform_maps_endpoints_to_src_dst",
        set.src_id_column == "user_id"
            && set.dst_id_column == "company_id"
            && names == ["src", "dst"],
        &format!(
            "src_id={}, dst_id={}, cols={:?}",
            set.src_id_column, set.dst_id_column, names
        ),
    );
}

#[test]
fn edge_transform_projects_properties_alongside_endpoints() {
    // `properties` columns ride along the `src`/`dst` endpoints, and `rename`
    // remaps a property column to its relationship-property name. The sink
    // folds everything but `src`/`dst` into `r.props` (`SET e += r.props`).
    let t = EdgeTransform {
        edge_type: "WORKS_AT".into(),
        src: EndpointSpec {
            label: "Person".into(),
            column: "user_id".into(),
            id_column: None,
        },
        dst: EndpointSpec {
            label: "Company".into(),
            column: "company_id".into(),
            id_column: None,
        },
        properties: vec!["name".into(), "city".into()],
        rename: [("city".to_string(), "location".to_string())].into(),
        key: Vec::new(),
        replace: None,
    };
    let ctx = TableContext { table: "t".into() };
    let m = t.apply(&ctx, &[users_batch()]).unwrap();
    assert_eq!(m.edges.len(), 1);
    let set = &m.edges[0];
    let names: Vec<_> = set
        .batch
        .schema()
        .fields()
        .iter()
        .map(|f| f.name().clone())
        .collect();
    // Endpoints first, then the (renamed) property columns.
    assert_eq!(names, vec!["src", "dst", "name", "location"]);
    assert_eq!(set.batch.num_rows(), 2);

    // The whole row renders the way the FalkorDB sink expects: endpoints out,
    // the property columns folded into `props`.
    let rows = crate::sink::rows_as_cypher_maps(&set.batch).unwrap();
    assert_eq!(
        rows[0],
        "{src: 1, dst: 10, props: {`name`: 'alice', `location`: 'oslo'}}"
    );
    crate::testmatrix::emit(
        crate::testmatrix::COMPONENT,
        "edge_transform_projects_properties_alongside_endpoints",
        names == ["src", "dst", "name", "location"] && set.batch.num_rows() == 2,
        &format!(
            "cols={:?}, rows={}, cypher={}",
            names,
            set.batch.num_rows(),
            rows[0]
        ),
    );
}

#[test]
fn edge_transform_self_key_emits_both_src_and_dst() {
    // FAIL-ON-BUG: a self-key edge (src.column == dst.column) used to collapse
    // to a single output field ('dst' won the rename map), so RecordBatch lost
    // the 'src' endpoint (or `try_new` failed on a duplicate field). PASS-ON-FIX:
    // both 'src' and 'dst' are emitted from the one shared input column.
    let t = EdgeTransform {
        edge_type: "SAME_AS".into(),
        src: EndpointSpec {
            label: "Person".into(),
            column: "user_id".into(),
            id_column: None,
        },
        dst: EndpointSpec {
            label: "Person".into(),
            column: "user_id".into(),
            id_column: None,
        },
        properties: vec![],
        rename: Default::default(),
        key: Vec::new(),
        replace: None,
    };
    let ctx = TableContext { table: "t".into() };
    let m = t.apply(&ctx, &[users_batch()]).unwrap();
    let set = &m.edges[0];
    let names: Vec<_> = set
        .batch
        .schema()
        .fields()
        .iter()
        .map(|f| f.name().clone())
        .collect();
    let pass = names == ["src", "dst"] && set.batch.num_rows() == 2;
    assert!(
        pass,
        "self-key edge must keep both endpoints, got {names:?}"
    );
    // Both endpoints carry the same value (the shared column).
    let rows = crate::sink::rows_as_cypher_maps(&set.batch).unwrap();
    assert_eq!(rows[0], "{src: 1, dst: 1, props: {}}");
    crate::testmatrix::emit(
        crate::testmatrix::COMPONENT,
        "edge_transform_self_key_emits_both_src_and_dst",
        pass,
        &format!("cols={names:?}, cypher={}", rows[0]),
    );
}

#[test]
fn edge_transform_property_colliding_with_endpoint_errors_clearly() {
    // A property that renames onto 'src'/'dst' (or two props onto one name)
    // must produce a clear DuplicateOutputColumn error, not an opaque Arrow
    // duplicate-field failure.
    let t = EdgeTransform {
        edge_type: "WORKS_AT".into(),
        src: EndpointSpec {
            label: "Person".into(),
            column: "user_id".into(),
            id_column: None,
        },
        dst: EndpointSpec {
            label: "Company".into(),
            column: "company_id".into(),
            id_column: None,
        },
        properties: vec!["name".into()],
        rename: [("name".to_string(), "src".to_string())].into(),
        key: Vec::new(),
        replace: None,
    };
    let ctx = TableContext { table: "t".into() };
    let res = t.apply(&ctx, &[users_batch()]);
    let is_dup = matches!(
        res,
        Err(crate::error::BifrostError::DuplicateOutputColumn(ref n)) if n == "src"
    );
    assert!(is_dup, "expected DuplicateOutputColumn(src), got {res:?}");
    crate::testmatrix::emit(
        crate::testmatrix::COMPONENT,
        "edge_transform_property_colliding_with_endpoint_errors_clearly",
        is_dup,
        "property→'src' collision → DuplicateOutputColumn",
    );
}

#[test]
fn vertex_transform_id_also_in_properties_dedups() {
    // FAIL-ON-BUG: listing the id_column again in `properties` put it in `keep`
    // twice, so project() built two identically-named fields and RecordBatch
    // try_new failed. PASS-ON-FIX: the duplicate source column is taken once.
    let t = VertexTransform {
        label: "Person".into(),
        id_column: "user_id".into(),
        properties: vec!["user_id".into(), "name".into()],
        rename: Default::default(),
    };
    let ctx = TableContext { table: "t".into() };
    let m = t.apply(&ctx, &[users_batch()]).unwrap();
    let set = &m.vertices[0];
    let names: Vec<_> = set
        .batch
        .schema()
        .fields()
        .iter()
        .map(|f| f.name().clone())
        .collect();
    let pass = names == ["user_id", "name"];
    assert!(
        pass,
        "id_column listed in properties must dedup, got {names:?}"
    );
    crate::testmatrix::emit(
        crate::testmatrix::COMPONENT,
        "vertex_transform_id_also_in_properties_dedups",
        pass,
        &format!("cols={names:?}"),
    );
}

// ── mapping ─────────────────────────────────────────────────────────────────

#[test]
fn mapping_yaml_parses_and_compiles() {
    let spec = MappingSpec::from_yaml(MAPPING_YAML).unwrap();
    assert_eq!(spec.version, 1);
    assert_eq!(spec.tables.len(), 1);
    assert_eq!(spec.tables[0].mode, SyncMode::Incremental);
    let transforms = spec.tables[0].compile(&TransformRegistry::new()).unwrap();
    assert_eq!(transforms.len(), 2);
    crate::testmatrix::emit(
        crate::testmatrix::COMPONENT,
        "mapping_yaml_parses_and_compiles",
        spec.version == 1 && spec.tables.len() == 1 && transforms.len() == 2,
        &format!(
            "v{}, {} table(s), {} transform(s)",
            spec.version,
            spec.tables.len(),
            transforms.len()
        ),
    );
}

#[test]
fn mapping_edge_properties_flow_into_edge_set() {
    // End-to-end through the declarative mapping: the `fk_edges` step declares
    // `properties: [city]` with `rename: { city: based_in }`, so the compiled
    // edge transform must emit an EdgeSet whose batch carries `src`, `dst`,
    // and the renamed `based_in` property column.
    let spec = MappingSpec::from_yaml(MAPPING_YAML).unwrap();
    let transforms = spec.tables[0].compile(&TransformRegistry::new()).unwrap();
    let ctx = TableContext {
        table: "lake.social.users".into(),
    };

    let edge_set = transforms
        .iter()
        .filter_map(|t| {
            t.apply(&ctx, &[users_batch()])
                .unwrap()
                .edges
                .into_iter()
                .next()
        })
        .next()
        .expect("the fk_edges step yields an EdgeSet");

    assert_eq!(edge_set.edge_type, "WORKS_AT");
    let names: Vec<_> = edge_set
        .batch
        .schema()
        .fields()
        .iter()
        .map(|f| f.name().clone())
        .collect();
    assert_eq!(names, vec!["src", "dst", "based_in"]);

    let rows = crate::sink::rows_as_cypher_maps(&edge_set.batch).unwrap();
    assert_eq!(rows[0], "{src: 1, dst: 10, props: {`based_in`: 'oslo'}}");
    crate::testmatrix::emit(
        crate::testmatrix::COMPONENT,
        "mapping_edge_properties_flow_into_edge_set",
        edge_set.edge_type == "WORKS_AT" && names == ["src", "dst", "based_in"],
        &format!(
            "type={}, cols={:?}, cypher={}",
            edge_set.edge_type, names, rows[0]
        ),
    );
}

#[test]
fn mapping_custom_step_requires_registration() {
    let yaml = r#"
version: 1
tables:
  - table: t
    pipeline:
      - custom: missing
"#;
    let spec = MappingSpec::from_yaml(yaml).unwrap();
    assert!(spec.tables[0].compile(&TransformRegistry::new()).is_err());

    struct Nop;
    impl Transform for Nop {
        fn apply(&self, _: &TableContext, _: &[RecordBatch]) -> Result<GraphMutation> {
            Ok(GraphMutation::default())
        }
    }
    let mut registry = TransformRegistry::new();
    registry.register("missing", Arc::new(Nop));
    let after = spec.tables[0].compile(&registry).is_ok();
    assert!(after);
    crate::testmatrix::emit(
        crate::testmatrix::COMPONENT,
        "mapping_custom_step_requires_registration",
        after,
        "unregistered custom step errs; registered compiles",
    );
}

// ── state ───────────────────────────────────────────────────────────────────

#[tokio::test]
async fn sync_state_round_trips_through_file() {
    let dir = tempdir().unwrap();
    let path = dir.path().join("state.json");
    {
        let mut state = SyncState::open(&path).unwrap();
        assert_eq!(state.get("t"), TableState::default());
        state
            .advance(
                "t",
                TableState {
                    snapshot_id: Some(42),
                    ..Default::default()
                },
            )
            .await
            .unwrap();
    }
    let reopened = SyncState::open(&path).unwrap();
    assert_eq!(reopened.get("t").snapshot_id, Some(42));
    assert!(reopened.get("t").synced_at.is_some());
    crate::testmatrix::emit(
        crate::testmatrix::COMPONENT,
        "sync_state_round_trips_through_file",
        reopened.get("t").snapshot_id == Some(42) && reopened.get("t").synced_at.is_some(),
        &format!("reopened snapshot_id={:?}", reopened.get("t").snapshot_id),
    );
}

// ── listener end-to-end (mock source + sink, no containers) ─────────────────

struct MockSource {
    snapshot: i64,
}

#[async_trait]
impl TableSource for MockSource {
    async fn fetch_new(
        &self,
        _spec: &TableSpec,
        state: &TableState,
    ) -> Result<Option<(Vec<RecordBatch>, TableState)>> {
        if state.snapshot_id == Some(self.snapshot) {
            return Ok(None);
        }
        Ok(Some((
            vec![users_batch()],
            TableState {
                snapshot_id: Some(self.snapshot),
                ..state.clone()
            },
        )))
    }
}

#[derive(Clone, Default)]
struct RecordingSink {
    mutations: Arc<Mutex<Vec<GraphMutation>>>,
}

#[async_trait]
impl GraphSink for RecordingSink {
    async fn apply(&mut self, mutation: &GraphMutation) -> Result<SinkReport> {
        self.mutations.lock().unwrap().push(mutation.clone());
        Ok(SinkReport {
            vertices_written: mutation.vertices.iter().map(|v| v.batch.num_rows()).sum(),
            edges_written: mutation.edges.iter().map(|e| e.batch.num_rows()).sum(),
            deleted: mutation.deletes.iter().map(|d| d.batch.num_rows()).sum(),
        })
    }
}

#[tokio::test]
async fn listener_syncs_then_goes_idle() {
    let dir = tempdir().unwrap();
    let spec = MappingSpec::from_yaml(MAPPING_YAML).unwrap();
    let sink = RecordingSink::default();
    let mutations = sink.mutations.clone();
    let mut listener = Listener::new(
        spec,
        TransformRegistry::new(),
        MockSource { snapshot: 7 },
        sink,
        SyncState::open(dir.path().join("state.json")).unwrap(),
    );

    // First round: table is new → 2 rows become 2 nodes + 2 edges.
    let round = listener.sync_once().await.unwrap();
    assert_eq!(round.failed, vec![]);
    assert_eq!(round.synced.len(), 1);
    let (table, rows, report) = &round.synced[0];
    assert_eq!(table, "lake.social.users");
    assert_eq!(*rows, 2);
    assert_eq!(report.vertices_written, 2);
    assert_eq!(report.edges_written, 2);
    assert_eq!(mutations.lock().unwrap().len(), 1);

    // Second round: snapshot unchanged → nothing happens.
    let round = listener.sync_once().await.unwrap();
    assert!(round.synced.is_empty());
    assert!(round.failed.is_empty());
    assert_eq!(mutations.lock().unwrap().len(), 1);
    crate::testmatrix::emit(
        crate::testmatrix::COMPONENT,
        "listener_syncs_then_goes_idle",
        round.synced.is_empty() && mutations.lock().unwrap().len() == 1,
        &format!(
            "1 sync round (2v/2e) then idle; {} mutation(s)",
            mutations.lock().unwrap().len()
        ),
    );
}

#[tokio::test]
async fn listener_metrics_accumulate_across_rounds() {
    // The cumulative `Metrics` surface (plan.md §7) is folded in `sync_once` and
    // exposed via `Listener::metrics()`. After one syncing round it carries the
    // round's rows/vertices/edges + one asset; an idle round only bumps
    // rounds_total (not a failure → no rounds_failed).
    let dir = tempdir().unwrap();
    let spec = MappingSpec::from_yaml(MAPPING_YAML).unwrap();
    let mut listener = Listener::new(
        spec,
        TransformRegistry::new(),
        MockSource { snapshot: 7 },
        RecordingSink::default(),
        SyncState::open(dir.path().join("state.json")).unwrap(),
    );

    listener.sync_once().await.unwrap();
    let m = listener.metrics().clone();
    assert_eq!(m.rounds_total, 1);
    assert_eq!(m.rounds_failed, 0);
    assert_eq!(m.tables_synced, 1);
    assert_eq!(m.rows_ingested, 2);
    assert_eq!(m.vertices_written, 2);
    assert_eq!(m.edges_written, 2);
    assert_eq!(m.assets_emitted, 1);

    // Idle round: counters hold except rounds_total; still no failure.
    listener.sync_once().await.unwrap();
    let m = listener.metrics();
    assert_eq!(m.rounds_total, 2);
    assert_eq!(m.rounds_failed, 0);
    assert_eq!(m.tables_synced, 1);
    assert_eq!(m.rows_ingested, 2);

    // Prometheus render reflects the live totals.
    let text = crate::render_prometheus(listener.metrics());
    assert!(text.contains("bifrost_rounds_total 2"));
    assert!(text.contains("bifrost_rows_ingested 2"));
    crate::testmatrix::emit(
        crate::testmatrix::COMPONENT,
        "listener_metrics_accumulate_across_rounds",
        text.contains("bifrost_rounds_total 2") && text.contains("bifrost_rows_ingested 2"),
        "prometheus reflects rounds_total=2, rows_ingested=2",
    );
}

#[tokio::test]
async fn listener_metrics_count_a_fully_failed_round() {
    // A source that always errors → every round is a no-progress failure, which
    // both backs off (tested in metrics::tests) and bumps `rounds_failed`.
    struct FailSource;
    #[async_trait]
    impl TableSource for FailSource {
        async fn fetch_new(
            &self,
            _spec: &TableSpec,
            _state: &TableState,
        ) -> Result<Option<(Vec<RecordBatch>, TableState)>> {
            Err(crate::error::BifrostError::Other("source down".into()))
        }
    }

    let dir = tempdir().unwrap();
    let spec = MappingSpec::from_yaml(MAPPING_YAML).unwrap();
    let mut listener = Listener::new(
        spec,
        TransformRegistry::new(),
        FailSource,
        RecordingSink::default(),
        SyncState::open(dir.path().join("state.json")).unwrap(),
    );

    let round = listener.sync_once().await.unwrap();
    assert!(round.is_failure(), "all tables failed → no-progress round");
    let m = listener.metrics();
    assert_eq!(m.rounds_total, 1);
    assert_eq!(m.rounds_failed, 1);
    assert_eq!(m.tables_synced, 0);
    crate::testmatrix::emit(
        crate::testmatrix::COMPONENT,
        "listener_metrics_count_a_fully_failed_round",
        round.is_failure() && m.rounds_failed == 1 && m.tables_synced == 0,
        &format!(
            "rounds_failed={}, tables_synced={}",
            m.rounds_failed, m.tables_synced
        ),
    );
}

// ── FileSource (watched directory of chunk files) ───────────────────────────

mod file_source {
    use super::*;
    use crate::source::FileSource;

    /// Write `batch` as a parquet chunk into `root/<table>/<name>`.
    fn drop_parquet(root: &std::path::Path, table: &str, name: &str, batch: &RecordBatch) {
        let dir = root.join(table);
        std::fs::create_dir_all(&dir).unwrap();
        graphar::io::write_chunk(
            dir.join(name),
            std::slice::from_ref(batch),
            &graphar::FileType::Parquet,
        )
        .unwrap();
    }

    fn spec(table: &str, mode: SyncMode) -> TableSpec {
        TableSpec::new(table, mode)
    }

    #[tokio::test]
    async fn missing_directory_yields_nothing() {
        let dir = tempdir().unwrap();
        let src = FileSource::new(dir.path());
        let got = src
            .fetch_new(
                &spec("lake.absent", SyncMode::Incremental),
                &TableState::default(),
            )
            .await
            .unwrap();
        assert!(got.is_none());
        crate::testmatrix::emit(
            crate::testmatrix::COMPONENT,
            "file_source::missing_directory_yields_nothing",
            got.is_none(),
            "absent table dir → None",
        );
    }

    #[tokio::test]
    async fn reads_files_then_records_them_done() {
        let dir = tempdir().unwrap();
        drop_parquet(dir.path(), "lake.users", "chunk0.parquet", &users_batch());
        let src = FileSource::new(dir.path());

        let (batches, state) = src
            .fetch_new(
                &spec("lake.users", SyncMode::Incremental),
                &TableState::default(),
            )
            .await
            .unwrap()
            .expect("first drop produces rows");
        assert_eq!(batches.iter().map(|b| b.num_rows()).sum::<usize>(), 2);
        assert_eq!(state.done_files, vec!["chunk0.parquet".to_string()]);

        // Idle round: same files, incremental mode → nothing new.
        let idle = src
            .fetch_new(&spec("lake.users", SyncMode::Incremental), &state)
            .await
            .unwrap();
        assert!(idle.is_none(), "no new files → no round");
        crate::testmatrix::emit(
            crate::testmatrix::COMPONENT,
            "file_source::reads_files_then_records_them_done",
            state.done_files == vec!["chunk0.parquet".to_string()] && idle.is_none(),
            &format!("done_files={:?}, idle round skipped", state.done_files),
        );
    }

    #[tokio::test]
    async fn incremental_ships_only_new_files() {
        let dir = tempdir().unwrap();
        drop_parquet(dir.path(), "lake.users", "chunk0.parquet", &users_batch());
        let src = FileSource::new(dir.path());

        let (_, state) = src
            .fetch_new(
                &spec("lake.users", SyncMode::Incremental),
                &TableState::default(),
            )
            .await
            .unwrap()
            .unwrap();

        // A second file drops.
        drop_parquet(dir.path(), "lake.users", "chunk1.parquet", &users_batch());
        let (batches, state2) = src
            .fetch_new(&spec("lake.users", SyncMode::Incremental), &state)
            .await
            .unwrap()
            .expect("new file produces a round");
        // Only the new file is read (2 rows), not both.
        assert_eq!(batches.iter().map(|b| b.num_rows()).sum::<usize>(), 2);
        assert_eq!(
            state2.done_files,
            vec!["chunk0.parquet".to_string(), "chunk1.parquet".to_string()]
        );
        crate::testmatrix::emit(
            crate::testmatrix::COMPONENT,
            "file_source::incremental_ships_only_new_files",
            batches.iter().map(|b| b.num_rows()).sum::<usize>() == 2
                && state2.done_files.len() == 2,
            &format!(
                "read {} new row(s), {} done files",
                batches.iter().map(|b| b.num_rows()).sum::<usize>(),
                state2.done_files.len()
            ),
        );
    }

    #[tokio::test]
    async fn full_mode_rereads_every_file() {
        let dir = tempdir().unwrap();
        drop_parquet(dir.path(), "lake.users", "chunk0.parquet", &users_batch());
        drop_parquet(dir.path(), "lake.users", "chunk1.parquet", &users_batch());
        let src = FileSource::new(dir.path());

        // Even with both files already in the ledger, Full re-reads them.
        let prior = TableState {
            done_files: vec!["chunk0.parquet".into(), "chunk1.parquet".into()],
            ..Default::default()
        };
        let (batches, _) = src
            .fetch_new(&spec("lake.users", SyncMode::Full), &prior)
            .await
            .unwrap()
            .expect("full mode always reads");
        let total = batches.iter().map(|b| b.num_rows()).sum::<usize>();
        assert_eq!(total, 4);
        crate::testmatrix::emit(
            crate::testmatrix::COMPONENT,
            "file_source::full_mode_rereads_every_file",
            total == 4,
            &format!("full mode re-read {total} rows from 2 files"),
        );
    }

    #[tokio::test]
    async fn non_chunk_files_are_ignored() {
        let dir = tempdir().unwrap();
        drop_parquet(dir.path(), "lake.users", "chunk0.parquet", &users_batch());
        // A stray non-data file in the same directory must not break the scan.
        std::fs::write(dir.path().join("lake.users").join("_SUCCESS"), b"").unwrap();
        let src = FileSource::new(dir.path());

        let (_, state) = src
            .fetch_new(
                &spec("lake.users", SyncMode::Incremental),
                &TableState::default(),
            )
            .await
            .unwrap()
            .unwrap();
        assert_eq!(state.done_files, vec!["chunk0.parquet".to_string()]);
        crate::testmatrix::emit(
            crate::testmatrix::COMPONENT,
            "file_source::non_chunk_files_are_ignored",
            state.done_files == vec!["chunk0.parquet".to_string()],
            "stray _SUCCESS ignored; only chunk0.parquet read",
        );
    }
}

// ── Airflow AIP-82 asset-event model ────────────────────────────────────────

mod airflow {
    use crate::airflow::{AssetEmitter, AssetEvent, asset_event_body, emit_round, round_to_events};
    use crate::error::Result;
    use crate::listener::RoundReport;
    use crate::mutation::SinkReport;
    use async_trait::async_trait;
    use std::sync::{Arc, Mutex};

    fn report() -> RoundReport {
        RoundReport {
            synced: vec![
                (
                    "lake.users".into(),
                    3,
                    SinkReport {
                        vertices_written: 3,
                        edges_written: 2,
                        deleted: 0,
                    },
                ),
                (
                    "lake.companies".into(),
                    1,
                    SinkReport {
                        vertices_written: 1,
                        edges_written: 0,
                        deleted: 0,
                    },
                ),
            ],
            failed: vec![("lake.broken".into(), "boom".into())],
        }
    }

    #[test]
    fn round_to_events_one_per_synced_table() {
        let events = round_to_events("graph://social/", &report());
        assert_eq!(events.len(), 2, "failed table emits nothing");
        assert_eq!(events[0].uri, "graph://social/lake.users");
        assert_eq!(events[0].extra["rows"], 3);
        assert_eq!(events[0].extra["vertices_written"], 3);
        assert_eq!(events[0].extra["edges_written"], 2);
        crate::testmatrix::emit(
            crate::testmatrix::COMPONENT,
            "airflow::round_to_events_one_per_synced_table",
            events.len() == 2 && events[0].uri == "graph://social/lake.users",
            &format!("{} event(s); failed table emits none", events.len()),
        );
    }

    #[test]
    fn empty_round_emits_no_events() {
        assert!(round_to_events("p", &RoundReport::default()).is_empty());
    }

    #[test]
    fn asset_event_body_matches_airflow_shape() {
        let ev = AssetEvent {
            uri: "graph://social/lake.users".into(),
            extra: serde_json::json!({ "rows": 3 }),
        };
        let body = asset_event_body(&ev);
        assert_eq!(body["asset_uri"], "graph://social/lake.users");
        assert_eq!(body["extra"]["rows"], 3);
        crate::testmatrix::emit(
            crate::testmatrix::COMPONENT,
            "airflow::asset_event_body_matches_airflow_shape",
            body["asset_uri"] == "graph://social/lake.users" && body["extra"]["rows"] == 3,
            "AIP-82 body carries asset_uri + extra.rows",
        );
    }

    #[derive(Default)]
    struct RecordingEmitter {
        seen: Arc<Mutex<Vec<AssetEvent>>>,
    }

    #[async_trait]
    impl AssetEmitter for RecordingEmitter {
        async fn emit(&self, events: &[AssetEvent]) -> Result<()> {
            self.seen.lock().unwrap().extend_from_slice(events);
            Ok(())
        }
    }

    #[tokio::test]
    async fn emit_round_delivers_each_event() {
        let emitter = RecordingEmitter::default();
        emit_round(&emitter, "graph://social/", &report())
            .await
            .unwrap();
        let seen = emitter.seen.lock().unwrap();
        assert_eq!(seen.len(), 2);
        assert_eq!(seen[1].uri, "graph://social/lake.companies");
        crate::testmatrix::emit(
            crate::testmatrix::COMPONENT,
            "airflow::emit_round_delivers_each_event",
            seen.len() == 2 && seen[1].uri == "graph://social/lake.companies",
            &format!("emitter saw {} event(s)", seen.len()),
        );
    }

    #[tokio::test]
    async fn emit_round_skips_empty() {
        let emitter = RecordingEmitter::default();
        emit_round(&emitter, "p", &RoundReport::default())
            .await
            .unwrap();
        assert!(emitter.seen.lock().unwrap().is_empty());
    }
}

// ── changelog (CDC) routing: partition by `_change_type` → upserts/deletes ───

mod cdc {
    use super::*;
    use crate::mapping::SyncMode;

    /// A users batch with a `_change_type` column appended (the shape the
    /// Iceberg changelog view returns).
    fn change_batch(change_types: &[&str]) -> RecordBatch {
        let n = change_types.len();
        let ids: Vec<i64> = (0..n as i64).collect();
        RecordBatch::try_new(
            Arc::new(Schema::new(vec![
                Field::new("user_id", DataType::Int64, false),
                Field::new("name", DataType::Utf8, false),
                Field::new("company_id", DataType::Int64, false),
                Field::new(SyncMode::CHANGE_TYPE_COLUMN, DataType::Utf8, false),
            ])),
            vec![
                Arc::new(Int64Array::from(ids)),
                Arc::new(StringArray::from(vec!["x"; n])),
                Arc::new(Int64Array::from(vec![10i64; n])),
                Arc::new(StringArray::from(change_types.to_vec())),
            ],
        )
        .unwrap()
    }

    fn col_names(batch: &RecordBatch) -> Vec<String> {
        batch
            .schema()
            .fields()
            .iter()
            .map(|f| f.name().clone())
            .collect()
    }

    #[test]
    fn detects_changelog_batches() {
        let cdc = is_changelog_batch(&change_batch(&["INSERT"]));
        let plain = is_changelog_batch(&users_batch());
        assert!(cdc);
        assert!(!plain);
        crate::testmatrix::emit(
            crate::testmatrix::COMPONENT,
            "cdc::detects_changelog_batches",
            cdc && !plain,
            "change batch detected as CDC; plain batch not",
        );
    }

    #[test]
    fn partition_splits_upserts_from_deletes_and_strips_sentinel() {
        let batch = change_batch(&["INSERT", "DELETE", "UPDATE_AFTER", "UPDATE_BEFORE"]);
        let split = partition_change_rows(&[batch]).unwrap();

        // INSERT + UPDATE_AFTER → upserts; DELETE → deletes; UPDATE_BEFORE is
        // dropped (its UPDATE_AFTER upsert overwrites the node in place, so a
        // delete would destroy the updated row).
        let up_rows: usize = split.upserts.iter().map(|b| b.num_rows()).sum();
        let del_rows: usize = split.deletes.iter().map(|b| b.num_rows()).sum();
        assert_eq!(up_rows, 2);
        assert_eq!(del_rows, 1);

        // `_change_type` is stripped from both sides.
        for b in split.upserts.iter().chain(split.deletes.iter()) {
            assert!(!col_names(b).contains(&SyncMode::CHANGE_TYPE_COLUMN.to_string()));
            assert!(col_names(b).contains(&"user_id".to_string()));
        }
        crate::testmatrix::emit(
            crate::testmatrix::COMPONENT,
            "cdc::partition_splits_upserts_from_deletes_and_strips_sentinel",
            up_rows == 2 && del_rows == 1,
            &format!("{up_rows} upsert rows, {del_rows} delete rows; sentinel stripped"),
        );
    }

    #[test]
    fn partition_passes_non_cdc_batches_through_as_upserts() {
        let split = partition_change_rows(&[users_batch()]).unwrap();
        assert_eq!(split.upserts.len(), 1);
        assert!(split.deletes.is_empty());
        // Untouched: still has all original columns.
        assert_eq!(col_names(&split.upserts[0]), col_names(&users_batch()));
    }

    #[test]
    fn partition_rejects_unknown_change_type() {
        let res = partition_change_rows(&[change_batch(&["WIGGLE"])]);
        assert!(res.is_err());
        crate::testmatrix::emit(
            crate::testmatrix::COMPONENT,
            "cdc::partition_rejects_unknown_change_type",
            res.is_err(),
            "unknown _change_type 'WIGGLE' → Err",
        );
    }

    #[test]
    fn deletes_from_change_rows_reinterprets_vertices_as_deletes() {
        let vt = VertexTransform {
            label: "Person".into(),
            id_column: "user_id".into(),
            properties: vec!["name".into()],
            rename: Default::default(),
        };
        let transforms: Vec<Arc<dyn Transform>> = vec![Arc::new(vt)];
        let ctx = TableContext {
            table: "lake.users".into(),
        };

        let split = partition_change_rows(&[change_batch(&["DELETE", "INSERT"])]).unwrap();
        let deletes = deletes_from_change_rows(&transforms, &ctx, &split.deletes).unwrap();

        assert_eq!(deletes.len(), 1);
        assert_eq!(deletes[0].label, "Person");
        assert_eq!(deletes[0].id_column, "user_id");
        assert_eq!(deletes[0].batch.num_rows(), 1); // only the DELETE row
        crate::testmatrix::emit(
            crate::testmatrix::COMPONENT,
            "cdc::deletes_from_change_rows_reinterprets_vertices_as_deletes",
            deletes.len() == 1 && deletes[0].label == "Person" && deletes[0].batch.num_rows() == 1,
            &format!(
                "{} delete set(s), {} row (only DELETE)",
                deletes.len(),
                deletes[0].batch.num_rows()
            ),
        );
    }

    #[test]
    fn graph_mutation_carries_and_merges_deletes() {
        // A delete-only mutation is not empty, and merge concatenates deletes.
        let del = VertexDelete {
            label: "Person".into(),
            id_column: "user_id".into(),
            batch: users_batch(),
        };
        let mut m = GraphMutation {
            edge_deletes: Vec::new(),
            deletes: vec![del.clone()],
            ..Default::default()
        };
        assert!(!m.is_empty());

        m.merge(GraphMutation {
            edge_deletes: Vec::new(),
            vertices: vec![VertexSet {
                label: "Person".into(),
                id_column: "user_id".into(),
                batch: users_batch(),
            }],
            deletes: vec![del],
            ..Default::default()
        });
        assert_eq!(m.deletes.len(), 2);
        assert_eq!(m.vertices.len(), 1);
        crate::testmatrix::emit(
            crate::testmatrix::COMPONENT,
            "cdc::graph_mutation_carries_and_merges_deletes",
            m.deletes.len() == 2 && m.vertices.len() == 1,
            &format!(
                "merged: {} delete set(s), {} vertex set(s)",
                m.deletes.len(),
                m.vertices.len()
            ),
        );
    }

    /// A changelog `_change_type` batch with a paired UPDATE_BEFORE/UPDATE_AFTER
    /// on the *same* primary key.
    fn update_pair_change_batch() -> RecordBatch {
        RecordBatch::try_new(
            Arc::new(Schema::new(vec![
                Field::new("user_id", DataType::Int64, false),
                Field::new("name", DataType::Utf8, false),
                Field::new(SyncMode::CHANGE_TYPE_COLUMN, DataType::Utf8, false),
            ])),
            vec![
                Arc::new(Int64Array::from(vec![1, 1])),
                Arc::new(StringArray::from(vec!["old", "new"])),
                Arc::new(StringArray::from(vec!["UPDATE_BEFORE", "UPDATE_AFTER"])),
            ],
        )
        .unwrap()
    }

    /// In-memory graph that reproduces the `FalkorDbSink::apply` phase ordering
    /// (vertices upserted via MERGE+SET, then deletes applied LAST via DETACH
    /// DELETE) without a database, so the end-to-end CDC routing semantics are
    /// testable as data.
    #[derive(Default)]
    struct InMemoryGraph {
        /// (label, id) → name property.
        nodes: std::collections::HashMap<(String, i64), String>,
    }

    impl InMemoryGraph {
        fn ids(batch: &RecordBatch, col: &str) -> Vec<i64> {
            batch
                .column(batch.schema().index_of(col).unwrap())
                .as_any()
                .downcast_ref::<Int64Array>()
                .unwrap()
                .values()
                .to_vec()
        }
        fn names(batch: &RecordBatch) -> Vec<String> {
            let c = batch
                .column(batch.schema().index_of("name").unwrap())
                .as_any()
                .downcast_ref::<StringArray>()
                .unwrap();
            (0..batch.num_rows())
                .map(|r| c.value(r).to_string())
                .collect()
        }
    }

    #[async_trait]
    impl GraphSink for InMemoryGraph {
        async fn apply(&mut self, mutation: &GraphMutation) -> Result<SinkReport> {
            let mut report = SinkReport::default();
            // Phase 1: vertices (MERGE on id, overwrite props).
            for set in &mutation.vertices {
                let ids = Self::ids(&set.batch, &set.id_column);
                let names = Self::names(&set.batch);
                for (id, name) in ids.into_iter().zip(names) {
                    self.nodes.insert((set.label.clone(), id), name);
                    report.vertices_written += 1;
                }
            }
            // Phase 3: deletes LAST (DETACH DELETE) — the exact ordering that
            // destroyed updated nodes when UPDATE_BEFORE was routed here.
            for set in &mutation.deletes {
                for id in Self::ids(&set.batch, &set.id_column) {
                    self.nodes.remove(&(set.label.clone(), id));
                    report.deleted += 1;
                }
            }
            Ok(report)
        }
    }

    fn person_transforms() -> Vec<Arc<dyn Transform>> {
        vec![Arc::new(VertexTransform {
            label: "Person".into(),
            id_column: "user_id".into(),
            properties: vec!["name".into()],
            rename: Default::default(),
        })]
    }

    /// End-to-end: a CDC UPDATE (UPDATE_BEFORE + UPDATE_AFTER on one id) must
    /// leave the node present with its NEW values. The pre-image must not be
    /// routed to `deletes`, or deletes-last would DETACH DELETE the row the
    /// UPDATE_AFTER just upserted.
    #[tokio::test]
    async fn cdc_update_pair_keeps_node_with_new_values() {
        let transforms = person_transforms();
        let ctx = TableContext {
            table: "lake.users".into(),
        };

        let change = partition_change_rows(&[update_pair_change_batch()]).unwrap();
        // The whole point: UPDATE_BEFORE is dropped, so there are no deletes and
        // exactly one upsert row (the UPDATE_AFTER post-image).
        assert!(
            change.deletes.is_empty(),
            "UPDATE_BEFORE must not become a delete"
        );
        let up_rows: usize = change.upserts.iter().map(|b| b.num_rows()).sum();
        assert_eq!(up_rows, 1);

        let mut mutation = GraphMutation::default();
        for t in &transforms {
            mutation.merge(t.apply(&ctx, &change.upserts).unwrap());
        }
        mutation.deletes = deletes_from_change_rows(&transforms, &ctx, &change.deletes).unwrap();

        let mut sink = InMemoryGraph::default();
        let report = sink.apply(&mutation).await.unwrap();

        let survived = sink.nodes.get(&("Person".to_string(), 1)).cloned();
        assert_eq!(
            survived.as_deref(),
            Some("new"),
            "updated node must survive with its new value, not be deleted"
        );
        assert_eq!(report.deleted, 0, "an UPDATE must delete nothing");

        crate::testmatrix::emit(
            crate::testmatrix::COMPONENT,
            "cdc::cdc_update_pair_keeps_node_with_new_values",
            survived.as_deref() == Some("new") && report.deleted == 0,
            &format!(
                "after UPDATE: Person(1).name={survived:?}, deleted={}",
                report.deleted
            ),
        );
    }

    /// Control: a genuine DELETE in the same round still removes the node, so the
    /// UPDATE_BEFORE fix did not weaken real deletions.
    #[tokio::test]
    async fn cdc_delete_still_removes_node() {
        let transforms = person_transforms();
        let ctx = TableContext {
            table: "lake.users".into(),
        };

        // Seed the node, then process a DELETE for it.
        let seed = RecordBatch::try_new(
            Arc::new(Schema::new(vec![
                Field::new("user_id", DataType::Int64, false),
                Field::new("name", DataType::Utf8, false),
                Field::new(SyncMode::CHANGE_TYPE_COLUMN, DataType::Utf8, false),
            ])),
            vec![
                Arc::new(Int64Array::from(vec![1, 1])),
                Arc::new(StringArray::from(vec!["x", "x"])),
                Arc::new(StringArray::from(vec!["INSERT", "DELETE"])),
            ],
        )
        .unwrap();

        let change = partition_change_rows(&[seed]).unwrap();
        assert_eq!(
            change.deletes.iter().map(|b| b.num_rows()).sum::<usize>(),
            1
        );

        let mut mutation = GraphMutation::default();
        for t in &transforms {
            mutation.merge(t.apply(&ctx, &change.upserts).unwrap());
        }
        mutation.deletes = deletes_from_change_rows(&transforms, &ctx, &change.deletes).unwrap();

        let mut sink = InMemoryGraph::default();
        let report = sink.apply(&mutation).await.unwrap();
        assert!(
            sink.nodes.get(&("Person".to_string(), 1)).is_none(),
            "INSERT+DELETE in one round resolves to gone"
        );
        assert_eq!(report.deleted, 1);
    }
}

// ── the CDC verdict must be readable out of state_json ──────────────────────

/// A source that reports a **fallback** verdict: the 4.2 native engine was
/// expected but the connector could not serve `CHANGES`. Exactly the state of
/// the world for Iceberg on Spark 4.2 today.
struct FellBackSource;

#[async_trait]
impl TableSource for FellBackSource {
    fn cdc_capability(&self, _table: &str) -> Option<crate::cdc::CdcCapability> {
        Some(crate::cdc::CdcCapability {
            chosen: crate::cdc::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 — connector implements no Changelog".into(),
        })
    }

    async fn fetch_new(
        &self,
        _spec: &TableSpec,
        state: &TableState,
    ) -> Result<Option<(Vec<RecordBatch>, TableState)>> {
        if state.snapshot_id == Some(9) {
            return Ok(None);
        }
        Ok(Some((
            vec![users_batch()],
            TableState {
                snapshot_id: Some(9),
                ..state.clone()
            },
        )))
    }
}

/// **RED-when-broken: a fallback must never be silent.** After a CDC round the
/// persisted `state_json` must name the backend that actually ran, say the
/// native path was expected, say it was NOT usable, and carry the reason. If
/// the listener ever stops stamping the verdict, the round still succeeds and
/// this test is the only thing that notices.
#[tokio::test]
async fn cdc_round_records_the_backend_and_its_fallback_in_state_json() {
    let dir = tempdir().unwrap();
    let path = dir.path().join("state.json");
    let yaml = MAPPING_YAML.replace("mode: incremental", "mode: cdc");
    assert!(
        yaml.contains("mode: cdc"),
        "the fixture must exercise CDC mode"
    );
    let spec = MappingSpec::from_yaml(&yaml).unwrap();

    let mut listener = Listener::new(
        spec,
        TransformRegistry::new(),
        FellBackSource,
        RecordingSink::default(),
        SyncState::open(&path).unwrap(),
    );
    let round = listener.sync_once().await.unwrap();
    assert_eq!(round.failed, vec![], "the round itself succeeds");
    assert_eq!(round.synced.len(), 1);

    // Read the LEDGER BACK OFF DISK — the observability contract is the file,
    // not an in-memory field.
    let raw = std::fs::read_to_string(&path).unwrap();
    let json: serde_json::Value = serde_json::from_str(&raw).unwrap();
    let cdc = &json["lake.social.users"]["cdc"];
    assert!(
        !cdc.is_null(),
        "state_json must carry the CDC verdict:\n{raw}"
    );
    assert_eq!(cdc["chosen"], "iceberg_changelog_view");
    assert_eq!(cdc["server_version"], "4.2.0");
    assert_eq!(cdc["native_expected"], true);
    assert_eq!(cdc["native_usable"], false, "the degrade is recorded");
    assert_eq!(cdc["fell_back"], true, "and it is NOT silent");
    assert!(
        cdc["reason"].as_str().unwrap().contains("Changelog"),
        "the engine's own reason survives: {cdc}",
    );

    // The typed round trip a UI reads.
    let back = SyncState::open(&path).unwrap();
    let cap = back
        .get("lake.social.users")
        .cdc
        .expect("verdict persisted");
    assert!(cap.fell_back);
    assert_eq!(cap.chosen, crate::cdc::CdcKind::IcebergChangelogView);
    crate::testmatrix::emit(
        crate::testmatrix::COMPONENT,
        "cdc::cdc_round_records_the_backend_and_its_fallback_in_state_json",
        cap.fell_back,
        &format!(
            "state_json records backend={} native_expected=true native_usable=false fell_back=true",
            cap.chosen.as_str()
        ),
    );
}