knut-bifrost 0.1.0

//! Table sources. `SparkConnectSource` (feature `spark`) polls Iceberg
//! tables through a Spark Connect 4.x endpoint.

use arrow_array::RecordBatch;
use async_trait::async_trait;

use crate::{error::Result, mapping::SyncMode, state::TableState};

/// What the listener asks a source each round.
#[derive(Debug, Clone)]
pub struct TableSpec {
    pub table: String,
    pub mode: SyncMode,
}

/// A place rows come from.
#[async_trait]
pub trait TableSource: Send + Sync {
    /// Return rows new since `state` and the state to persist once they are
    /// sunk, or `None` when nothing changed.
    async fn fetch_new(
        &self,
        spec: &TableSpec,
        state: &TableState,
    ) -> Result<Option<(Vec<RecordBatch>, TableState)>>;
}

pub use file::FileSource;

mod file {
    use std::path::{Path, PathBuf};

    use graphar::FileType;

    use super::*;
    use crate::error::BifrostError;

    /// A [`TableSource`] over a watched directory tree of GraphAr/columnar
    /// file drops — the local half of the Airflow AIP-82 story.
    ///
    /// Each mapped table maps to a subdirectory `root/<table>/`; every file
    /// in it with a recognised extension (`parquet`, `orc`, `csv`, `json`,
    /// `arrow`) is a chunk. The set of already-ingested file names lives in
    /// the table's [`TableState::done_files`] ledger, so a new file drop
    /// triggers exactly one sync round and nothing is re-read.
    ///
    /// ```no_run
    /// use knut_bifrost::source::FileSource;
    /// let source = FileSource::new("/data/drops");
    /// ```
    ///
    /// [`TableState::done_files`]: crate::state::TableState::done_files
    pub struct FileSource {
        root: PathBuf,
    }

    impl FileSource {
        /// Watch `root`; table `lake.foo.bar` reads from `root/lake.foo.bar/`.
        pub fn new(root: impl Into<PathBuf>) -> Self {
            Self { root: root.into() }
        }

        fn table_dir(&self, table: &str) -> PathBuf {
            self.root.join(table)
        }

        /// All readable chunk files in `dir`, sorted by name for determinism.
        fn list_files(dir: &Path) -> Result<Vec<String>> {
            let read = match std::fs::read_dir(dir) {
                Ok(r) => r,
                // A table whose drop directory does not exist yet is simply
                // empty — not an error (it may appear on a later round).
                Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(vec![]),
                Err(e) => return Err(e.into()),
            };
            let mut names = Vec::new();
            for entry in read {
                let entry = entry?;
                if !entry.file_type()?.is_file() {
                    continue;
                }
                let name = entry.file_name().to_string_lossy().into_owned();
                if file_type_of(&name).is_some() {
                    names.push(name);
                }
            }
            names.sort();
            Ok(names)
        }
    }

    /// Recognise a chunk file by extension; `None` means "ignore this file".
    fn file_type_of(name: &str) -> Option<FileType> {
        match name.rsplit_once('.').map(|(_, ext)| ext) {
            Some("parquet") => Some(FileType::Parquet),
            Some("orc") => Some(FileType::Orc),
            Some("csv") => Some(FileType::Csv),
            Some("json") => Some(FileType::Json),
            Some("arrow") => Some(FileType::ArrowIpc),
            _ => None,
        }
    }

    #[async_trait]
    impl TableSource for FileSource {
        async fn fetch_new(
            &self,
            spec: &TableSpec,
            state: &TableState,
        ) -> Result<Option<(Vec<RecordBatch>, TableState)>> {
            let dir = self.table_dir(&spec.table);
            let present = Self::list_files(&dir)?;

            // Full mode re-reads every file each round (idempotent MERGE);
            // incremental ships only files absent from the ledger.
            let to_read: Vec<&String> = match spec.mode {
                SyncMode::Full => present.iter().collect(),
                // File drops have no changelog view, so CDC degrades to the
                // append-only file ledger (new files only) — same as
                // Incremental. (Per-row CDC would need a `_change_type` column
                // inside the dropped files, which the Iceberg source supplies.)
                SyncMode::Incremental | SyncMode::Cdc => present
                    .iter()
                    .filter(|f| !state.done_files.contains(f))
                    .collect(),
            };
            if to_read.is_empty() {
                tracing::debug!(
                    target: "bifrost::file",
                    table = %spec.table,
                    present = present.len(),
                    "no new files",
                );
                return Ok(None);
            }

            let mut batches = Vec::new();
            let mut rows = 0usize;
            for name in &to_read {
                let ft = file_type_of(name).expect("filtered to known types");
                let chunk = graphar::io::read_chunk(dir.join(name), &ft)
                    .map_err(|e| BifrostError::Source(format!("{name}: {e}")))?;
                let n: usize = chunk.iter().map(|b| b.num_rows()).sum();
                rows += n;
                // One event per detected-and-read file — the "file detected"
                // signal a test/bencher asserts on without a human.
                tracing::info!(
                    target: "bifrost::file",
                    event = "file_detected",
                    table = %spec.table,
                    file = %name,
                    rows = n,
                    "detected new file",
                );
                batches.extend(chunk);
            }
            tracing::info!(
                target: "bifrost::file",
                event = "files_read",
                table = %spec.table,
                files = to_read.len(),
                rows,
                "read new files",
            );

            let new_state = TableState {
                done_files: present, // everything now present is done
                ..state.clone()
            };
            Ok(Some((batches, new_state)))
        }
    }
}

#[cfg(feature = "spark")]
pub use spark::SparkConnectSource;

#[cfg(feature = "spark")]
mod spark {
    use super::*;
    use arrow_array::cast::AsArray;
    use arrow_array::types::Int64Type;
    use spark_connect::{SparkSession, SparkSessionBuilder};

    use crate::error::BifrostError;

    /// Make a table name safe to embed in a temp-view identifier: keep
    /// alphanumerics, map everything else (`.`, `-`, …) to `_`.
    fn sanitise_view(table: &str) -> String {
        table
            .chars()
            .map(|c| if c.is_ascii_alphanumeric() { c } else { '_' })
            .collect()
    }

    /// Polls Iceberg tables via Spark Connect (`sc://host:15002`).
    ///
    /// Snapshot detection queries the table's `.snapshots` metadata table;
    /// incremental mode reads only the delta between two snapshots.
    pub struct SparkConnectSource {
        session: SparkSession,
    }

    impl SparkConnectSource {
        pub async fn connect(url: &str) -> Result<Self> {
            let session = SparkSessionBuilder::new(url)
                .build()
                .await
                .map_err(|e| BifrostError::Source(e.to_string()))?;
            Ok(Self { session })
        }

        /// Run arbitrary SQL on the Spark Connect session (DDL, admin,
        /// test fixtures). Returns the result batches, bridged from
        /// spark-connect's arrow 56 to the workspace arrow 58.
        pub async fn sql(&self, query: &str) -> Result<Vec<RecordBatch>> {
            let plan = self
                .session
                .sql(query, vec![])
                .await
                .map_err(|e| BifrostError::Source(format!("{query}: {e}")))?;
            let batches56 = self
                .session
                .collect(plan)
                .await
                .map_err(|e| BifrostError::Source(format!("{query}: {e}")))?;
            bridge_spark_arrow56(&batches56)
        }

        /// Read Iceberg's changelog between two snapshots via the
        /// `system.create_changelog_view` procedure.
        ///
        /// The procedure materialises a temp view whose rows carry a
        /// `_change_type` column (`INSERT` / `UPDATE_AFTER` / `UPDATE_BEFORE` /
        /// `DELETE`); we then `SELECT *` from it. `net-changes` collapses an
        /// update into a single `INSERT`/`DELETE` pair and drops rows that an
        /// in-window insert+delete cancels out — exactly the upsert/delete
        /// split the graph wants.
        ///
        /// `prev` is the *start* snapshot (already applied, so the changelog is
        /// open-ended after it) and `latest` the end snapshot.
        async fn read_changelog(
            &self,
            table: &str,
            prev: i64,
            latest: i64,
        ) -> Result<Vec<RecordBatch>> {
            // A stable, table-scoped view name; recreated each call (the proc
            // replaces it), so concurrent tables don't collide.
            let view = format!("bifrost_changes_{}", sanitise_view(table));
            self.sql(&format!(
                "CALL system.create_changelog_view(\
                   table => '{table}', \
                   options => map('start-snapshot-id','{prev}','end-snapshot-id','{latest}'), \
                   net_changes => true, \
                   changelog_view => '{view}')"
            ))
            .await?;
            let batches = self.sql(&format!("SELECT * FROM {view}")).await?;
            // Best-effort cleanup; a leftover temp view is harmless and gets
            // replaced next round, so a drop failure must not fail the round.
            let _ = self.sql(&format!("DROP VIEW IF EXISTS {view}")).await;
            Ok(batches)
        }

        async fn latest_snapshot(&self, table: &str) -> Result<Option<i64>> {
            // Spark caches Iceberg table metadata per session; without a
            // refresh, commits from other writers stay invisible to a
            // long-lived polling session.
            self.sql(&format!("REFRESH TABLE {table}")).await?;
            let batches = self
                .sql(&format!(
                    "SELECT snapshot_id FROM {table}.snapshots \
                     ORDER BY committed_at DESC LIMIT 1"
                ))
                .await?;
            for batch in &batches {
                if batch.num_rows() > 0 {
                    let col = batch.column(0).as_primitive::<Int64Type>();
                    return Ok(Some(col.value(0)));
                }
            }
            Ok(None)
        }
    }

    #[async_trait]
    impl TableSource for SparkConnectSource {
        async fn fetch_new(
            &self,
            spec: &TableSpec,
            state: &TableState,
        ) -> Result<Option<(Vec<RecordBatch>, TableState)>> {
            let latest = self.latest_snapshot(&spec.table).await?;
            let Some(latest) = latest else {
                return Ok(None); // table has no snapshots yet
            };
            if state.snapshot_id == Some(latest) {
                return Ok(None); // nothing new
            }

            // CDC reads the changelog *view* (a separate SQL call that creates
            // a temp view), so it can't share the single-`SELECT` path below.
            // (First CDC round has no prior snapshot to diff against — it falls
            // through to the whole-table seed load tagged INSERT, below.)
            if let (SyncMode::Cdc, Some(prev)) = (spec.mode, state.snapshot_id) {
                let batches = self.read_changelog(&spec.table, prev, latest).await?;
                let new_state = TableState {
                    snapshot_id: Some(latest),
                    ..state.clone()
                };
                return Ok(Some((batches, new_state)));
            }

            let query = match (spec.mode, state.snapshot_id) {
                // First sync, full mode, or first CDC round: whole table at the
                // new snapshot. For CDC, tag every row INSERT so the listener's
                // partitioning treats the seed load as upserts.
                (SyncMode::Cdc, None) => format!(
                    "SELECT *, 'INSERT' AS {ct} FROM {t} VERSION AS OF {latest}",
                    ct = SyncMode::CHANGE_TYPE_COLUMN,
                    t = spec.table,
                ),
                (SyncMode::Full, _) | (SyncMode::Incremental, None) => {
                    format!("SELECT * FROM {} VERSION AS OF {latest}", spec.table)
                }
                // Appended rows only. EXCEPT ALL is portable SQL; CDC mode above
                // sees deletes/updates via Iceberg's changelog view instead.
                (SyncMode::Incremental, Some(prev)) => format!(
                    "SELECT * FROM {t} VERSION AS OF {latest} \
                     EXCEPT ALL \
                     SELECT * FROM {t} VERSION AS OF {prev}",
                    t = spec.table
                ),
                // Unreachable: CDC with a prior snapshot returned above.
                (SyncMode::Cdc, Some(_)) => unreachable!("CDC delta handled above"),
            };
            let batches = self.sql(&query).await?;

            let new_state = TableState {
                snapshot_id: Some(latest),
                ..state.clone()
            };
            Ok(Some((batches, new_state)))
        }
    }

    /// Bridge spark-connect's arrow-56 batches to the workspace arrow-58 via an
    /// Arrow-IPC round-trip.
    ///
    /// spark-connect 0.2.2 is still on arrow 56, so its `collect()` yields
    /// arrow-56 `RecordBatch`es that are a *different* Rust type from the
    /// workspace's arrow-58 `RecordBatch`. Rather than the old FFI transmute
    /// (the constellation is collapsing to a single arrow major and FFI is being
    /// retired), we serialize each batch to the Arrow IPC stream format with
    /// arrow 56 and re-read it with arrow 58. The IPC byte layout is stable
    /// across these majors, so every column, value and null survives. Empty
    /// input → empty output (no schema to carry).
    fn bridge_spark_arrow56(
        batches56: &[spark_arrow_56::array::RecordBatch],
    ) -> Result<Vec<RecordBatch>> {
        let err = |m: String| BifrostError::Source(m);
        if batches56.is_empty() {
            return Ok(Vec::new());
        }
        // Serialize with arrow 56.
        let schema56 = batches56[0].schema();
        let mut buf: Vec<u8> = Vec::new();
        {
            let mut writer = spark_arrow_56::ipc::writer::StreamWriter::try_new(&mut buf, &schema56)
                .map_err(|e| err(format!("ipc writer (arrow56): {e}")))?;
            for b in batches56 {
                writer
                    .write(b)
                    .map_err(|e| err(format!("ipc write (arrow56): {e}")))?;
            }
            writer
                .finish()
                .map_err(|e| err(format!("ipc finish (arrow56): {e}")))?;
        }
        // Re-read with workspace arrow 58.
        let reader = arrow_ipc::reader::StreamReader::try_new(std::io::Cursor::new(buf), None)
            .map_err(|e| err(format!("ipc reader (arrow58): {e}")))?;
        let mut out = Vec::with_capacity(batches56.len());
        for b in reader {
            out.push(b.map_err(|e| err(format!("ipc read (arrow58): {e}")))?);
        }
        Ok(out)
    }

    #[cfg(test)]
    mod tests {
        use super::*;
        use std::sync::Arc;

        #[test]
        fn spark_arrow56_bridge_preserves_values_and_nulls() {
            // Build an arrow-56 batch (Int64 + Utf8 with a null) — the shape
            // spark-connect 0.2.2 hands back — then bridge it to the workspace
            // arrow 58 via the IPC round-trip and assert every value, the null,
            // and the row/column counts survive the major-version crossing.
            use spark_arrow_56::array::{Int64Array, RecordBatch as B56, StringArray};
            use spark_arrow_56::datatypes::{DataType, Field, Schema};

            let batch56 = B56::try_new(
                Arc::new(Schema::new(vec![
                    Field::new("id", DataType::Int64, false),
                    Field::new("name", DataType::Utf8, true),
                ])),
                vec![
                    Arc::new(Int64Array::from(vec![10, 20, 30])),
                    Arc::new(StringArray::from(vec![Some("a"), None, Some("c")])),
                ],
            )
            .unwrap();

            let out = bridge_spark_arrow56(std::slice::from_ref(&batch56)).unwrap();
            assert_eq!(out.len(), 1, "one batch in, one batch out");
            let b58 = &out[0];
            assert_eq!(b58.num_rows(), 3);
            assert_eq!(b58.num_columns(), 2);

            use arrow_array::cast::AsArray;
            use arrow_array::types::Int64Type;
            use arrow_array::Array;
            let ids = b58.column(0).as_primitive::<Int64Type>();
            assert_eq!(ids.values(), &[10, 20, 30]);

            let names = b58.column(1).as_string::<i32>();
            assert_eq!(names.value(0), "a");
            assert!(names.is_null(1), "the null survived the arrow 56→58 bridge");
            assert_eq!(names.value(2), "c");
        }

        #[test]
        fn spark_arrow56_bridge_empty_is_empty() {
            // No input batches → no schema to carry → empty output, no panic.
            let out = bridge_spark_arrow56(&[]).unwrap();
            assert!(out.is_empty());
        }
    }
}

#[cfg(feature = "skade")]
pub use iceberg::{DeltaPlan, IcebergSource};

/// Iceberg source via `skade` (feature `skade`) — read an Iceberg table
/// directly from an embedded warehouse, no Spark.
#[cfg(feature = "skade")]
mod iceberg {
    use std::path::PathBuf;

    use super::*;
    use crate::error::BifrostError;

    /// A [`TableSource`] over an Iceberg table read through `skade`'s embedded
    /// catalog. Each poll compares the table's current snapshot id to the
    /// recorded one; on change it ships rows to the listener. The behaviour
    /// depends on the [`SyncMode`] the listener asks for:
    ///
    /// * [`SyncMode::Full`] — re-read the **whole** table at the new snapshot
    ///   (idempotent MERGE in the sink makes re-reads safe). Robust to any
    ///   commit kind (overwrite, delete, compaction) at the cost of re-scanning
    ///   every row each change.
    /// * [`SyncMode::Incremental`] / [`SyncMode::Cdc`] — read **only the files
    ///   added since the recorded snapshot** (a manifest-level delta over the
    ///   snapshot lineage; see [`Self::ingest_delta`]). This is the
    ///   streaming/delta path: a continuous poll injects just the changes. It is
    ///   **CDC-aware**: appended data files become inserts, and a `Delete`
    ///   snapshot's **equality-delete** files become `_change_type = DELETE`
    ///   rows that remove the matching graph nodes (pure Rust, no Spark — skade
    ///   writes the deletes and surfaces them, knut tags + routes them through
    ///   the listener's changelog split). A position-delete or overwrite/replace
    ///   commit falls back to a full re-read so the graph never goes stale.
    ///
    /// For append-only tables this replaces `SparkConnectSource` — no Spark
    /// cluster needed, no JVM, pure Rust.
    ///
    /// skade is on arrow 57 and knut on 56; the read batches cross the version
    /// boundary **zero-copy** through the Arrow C Data Interface (see
    /// [`bridge_57_to_56`]).
    pub struct IcebergSource {
        warehouse_dir: PathBuf,
    }

    /// The outcome of a manifest-level delta plan: which snapshots were new and
    /// which data files they added/deleted. Lets a caller see *what* the delta
    /// resolved to (for tests/telemetry) before the rows are read. Mirrors
    /// [`skade::DeltaPlan`] (the engine-side plan this wraps).
    #[derive(Debug, Clone, Default)]
    pub struct DeltaPlan {
        /// Snapshot ids strictly after the cursor, up to and including the
        /// target, oldest-first.
        pub snapshots: Vec<i64>,
        /// Absolute data-file paths added by those snapshots (rows to insert).
        pub added_files: Vec<String>,
        /// Equality-delete file paths added by those snapshots (rows to delete —
        /// CDC deletes). Non-empty ⇒ the delta carries `_change_type` rows.
        pub delete_files: Vec<String>,
        /// `true` when an overwrite/replace/position-delete commit in the window
        /// forced a full re-read instead of an additive delta.
        pub needs_full_reload: bool,
    }

    impl IcebergSource {
        /// Read tables from the embedded Iceberg warehouse rooted at `dir`
        /// (`<dir>/catalog.redb` + `warehouse/`). A `TableSpec.table` of
        /// `name` resolves to the Iceberg table `name` in the default namespace.
        pub fn new(dir: impl Into<PathBuf>) -> Self {
            Self {
                warehouse_dir: dir.into(),
            }
        }

        /// Open the warehouse and load the table handle.
        async fn open_table(&self, table: &str) -> Result<skade::Table> {
            let wh = skade::Warehouse::open(&self.warehouse_dir)
                .await
                .map_err(|e| BifrostError::Source(e.to_string()))?;
            wh.table(table)
                .await
                .map_err(|e| BifrostError::Source(format!("{table}: {e}")))
        }

        /// **The delta primitive.** Read the changes to `table` between the
        /// snapshots strictly after `from_snapshot` up to (and including)
        /// `to_snapshot` into Arrow-56 batches — only the files those snapshots
        /// added, never the whole table.
        ///
        /// One-shot, pure Rust, no Spark. The manifest-level snapshot diff is
        /// built by [`skade::Table::read_delta`] (iceberg-rust 0.9.1 has no
        /// first-class incremental scan): it walks the snapshot lineage and
        /// classifies each added file — appended **data** files are inserts,
        /// **equality-delete** files are CDC deletes.
        ///
        /// `from_snapshot = None` means "from the beginning" → every data file
        /// up to `to` (equivalent to a full read, but file-by-file).
        ///
        /// Returns the read batches **and** the [`DeltaPlan`] so a caller can
        /// assert on the plan (snapshots, added/delete files) without a human.
        /// When the window has identity deletes the batches carry a
        /// `_change_type` column (`INSERT`/`DELETE`) for the listener's changelog
        /// routing; a pure-append window returns untagged batches. A position-
        /// delete or overwrite/replace commit sets `needs_full_reload` and the
        /// batches are a **full** re-read of `to`.
        pub async fn ingest_delta(
            &self,
            table_name: &str,
            from_snapshot: Option<i64>,
            to_snapshot: i64,
        ) -> Result<(Vec<RecordBatch>, DeltaPlan)> {
            // Opens the warehouse once (redb is single-open); the delta runs
            // against this same handle so we never reopen it mid-call.
            let table = self.open_table(table_name).await?;
            Self::delta_from_table(&table, table_name, from_snapshot, to_snapshot).await
        }

        /// The delta over an already-open [`skade::Table`] — the shared core of
        /// [`Self::ingest_delta`] (which opens its own handle) and the
        /// [`TableSource::fetch_new`] Incremental path (which reuses the handle
        /// it already opened; redb is single-open, so reopening would deadlock).
        async fn delta_from_table(
            table: &skade::Table,
            table_name: &str,
            from_snapshot: Option<i64>,
            to_snapshot: i64,
        ) -> Result<(Vec<RecordBatch>, DeltaPlan)> {
            // The engine-side plan + insert batches (arrow 57), built by skade
            // from the manifest layer. This is the single source of truth for
            // the snapshot diff (knut no longer re-derives it).
            let (inserts57, sk_plan) = table
                .read_delta(from_snapshot, to_snapshot)
                .await
                .map_err(|e| BifrostError::Source(format!("{table_name}: {e}")))?;

            let mut plan = DeltaPlan {
                snapshots: sk_plan.snapshots.clone(),
                added_files: sk_plan.added_files.clone(),
                delete_files: sk_plan.delete_files.iter().map(|d| d.path.clone()).collect(),
                needs_full_reload: sk_plan.needs_full_reload,
            };

            if sk_plan.needs_full_reload {
                // An overwrite/replace/position-delete commit can't be expressed
                // as an additive delta; re-read the whole target snapshot (the
                // sink MERGEs, so re-reads are safe).
                tracing::info!(
                    target: "bifrost::iceberg",
                    event = "delta_full_reload",
                    table = %table_name,
                    from = ?from_snapshot,
                    to = to_snapshot,
                    "overwrite/position-delete in window — full re-read",
                );
                let batches57 = table
                    .read()
                    .await
                    .map_err(|e| BifrostError::Source(format!("{table_name}: {e}")))?;
                return Ok((bridge_57_to_56(&batches57)?, plan));
            }

            // The equality-delete identities (arrow 57) for any Delete snapshots
            // in the window — the rows to remove from the graph (CDC deletes).
            let del57 = table
                .read_equality_deletes(&sk_plan)
                .await
                .map_err(|e| BifrostError::Source(format!("{table_name}: {e}")))?;
            let has_deletes = del57.iter().any(|(_, b)| !b.is_empty());

            // Bridge inserts to arrow 56.
            let inserts = bridge_57_to_56(&inserts57)?;
            let insert_rows: usize = inserts.iter().map(|b| b.num_rows()).sum();

            // Pure-append delta (the common case): no `_change_type`, the rows
            // flow straight through the upsert pipeline exactly as before.
            if !has_deletes {
                tracing::info!(
                    target: "bifrost::iceberg",
                    event = "delta_read",
                    table = %table_name,
                    from = ?from_snapshot,
                    to = to_snapshot,
                    snapshots = plan.snapshots.len(),
                    files = plan.added_files.len(),
                    rows = insert_rows,
                    "read iceberg delta (appended data files only)",
                );
                return Ok((inserts, plan));
            }

            // CDC delta: the window has identity deletes, so tag every row with
            // `_change_type` and let the listener's changelog routing split
            // inserts (upserts) from deletes. INSERT rows get the full schema;
            // DELETE rows carry only the equality-delete columns, so we WIDEN
            // them to the table schema (absent columns null-filled) — that makes
            // a delete row shaped exactly like a Spark changelog DELETE row, so
            // the same vertex transform projects it (only the id is load-bearing;
            // the sink deletes by id, see `delete_ids_read_only_the_id_column`).
            let table_schema = bridged_table_schema(table).await?;
            let mut out = Vec::with_capacity(inserts.len() + del57.len());
            for b in &inserts {
                out.push(tag_change_type(b, "INSERT")?);
            }
            let mut delete_rows = 0usize;
            for (_eq_ids, batches) in &del57 {
                for b57 in batches {
                    let b = batch_57_to_56(b57)?;
                    delete_rows += b.num_rows();
                    let widened = widen_to_schema(&b, &table_schema)?;
                    out.push(tag_change_type(&widened, "DELETE")?);
                }
            }
            plan.delete_files = sk_plan.delete_files.iter().map(|d| d.path.clone()).collect();
            tracing::info!(
                target: "bifrost::iceberg",
                event = "delta_read_cdc",
                table = %table_name,
                from = ?from_snapshot,
                to = to_snapshot,
                snapshots = plan.snapshots.len(),
                added_files = plan.added_files.len(),
                delete_files = plan.delete_files.len(),
                insert_rows,
                delete_rows,
                "read iceberg CDC delta (inserts + identity deletes)",
            );
            Ok((out, plan))
        }

        /// The table's current snapshot id (the delta target), or `None` if it
        /// has no snapshots yet. The cursor a `watch` loop polls against.
        pub async fn latest_snapshot(&self, table_name: &str) -> Result<Option<i64>> {
            let table = self.open_table(table_name).await?;
            Ok(table
                .inner()
                .metadata()
                .current_snapshot()
                .map(|s| s.snapshot_id()))
        }
    }

    /// The table's current Arrow-**56** schema (bridged from skade's arrow-57):
    /// build a zero-row arrow-57 batch from the table schema and run it through
    /// the same C-Data-Interface bridge, then take its schema.
    async fn bridged_table_schema(table: &skade::Table) -> Result<arrow_schema::SchemaRef> {
        let s57 = table
            .arrow_schema()
            .map_err(|e| BifrostError::Source(format!("table schema: {e}")))?;
        let empty57 = skade::arrow_array::RecordBatch::new_empty(s57);
        let b56 = batch_57_to_56(&empty57)?;
        Ok(b56.schema())
    }

    /// Widen `batch` to `target` schema: keep the columns it has (by name), and
    /// fill every other `target` column with an all-null array of the right
    /// type. Used to shape an equality-delete identity batch (only the key
    /// columns) like a full table row so the vertex transform projects it.
    fn widen_to_schema(
        batch: &RecordBatch,
        target: &arrow_schema::SchemaRef,
    ) -> Result<RecordBatch> {
        let n = batch.num_rows();
        let mut fields: Vec<arrow_schema::Field> = Vec::with_capacity(target.fields().len());
        let mut columns: Vec<arrow_array::ArrayRef> = Vec::with_capacity(target.fields().len());
        for field in target.fields() {
            match batch.schema().index_of(field.name()) {
                Ok(i) => {
                    fields.push(field.as_ref().clone());
                    columns.push(batch.column(i).clone());
                }
                Err(_) => {
                    // Filled with nulls for a delete row, so the field must be
                    // nullable even if the table declares it NOT NULL (the value
                    // is never read — the sink deletes by id).
                    fields.push(arrow_schema::Field::new(
                        field.name(),
                        field.data_type().clone(),
                        true,
                    ));
                    columns.push(arrow_array::new_null_array(field.data_type(), n));
                }
            }
        }
        RecordBatch::try_new(std::sync::Arc::new(arrow_schema::Schema::new(fields)), columns)
            .map_err(|e| BifrostError::Source(format!("widen delete batch: {e}")))
    }

    /// Append a string `_change_type` column (all rows = `kind`) so the
    /// listener's changelog routing (`partition_change_rows`) splits the delta
    /// into upserts vs. deletes. This is the pure-Rust equivalent of the
    /// `_change_type` column Spark's `create_changelog_view` emits.
    fn tag_change_type(batch: &RecordBatch, kind: &str) -> Result<RecordBatch> {
        use arrow_array::StringArray;
        let n = batch.num_rows();
        let mut fields: Vec<arrow_schema::Field> =
            batch.schema().fields().iter().map(|f| f.as_ref().clone()).collect();
        let mut columns: Vec<arrow_array::ArrayRef> = batch.columns().to_vec();
        fields.push(arrow_schema::Field::new(
            crate::mapping::SyncMode::CHANGE_TYPE_COLUMN,
            arrow_schema::DataType::Utf8,
            false,
        ));
        columns.push(std::sync::Arc::new(StringArray::from(vec![kind; n])));
        RecordBatch::try_new(std::sync::Arc::new(arrow_schema::Schema::new(fields)), columns)
            .map_err(|e| BifrostError::Source(format!("tag _change_type: {e}")))
    }

    #[async_trait]
    impl TableSource for IcebergSource {
        async fn fetch_new(
            &self,
            spec: &TableSpec,
            state: &TableState,
        ) -> Result<Option<(Vec<RecordBatch>, TableState)>> {
            let table = self.open_table(&spec.table).await?;

            // The current Iceberg snapshot id; unchanged → nothing to do.
            let snapshot = table
                .inner()
                .metadata()
                .current_snapshot()
                .map(|s| s.snapshot_id());
            let Some(latest) = snapshot else {
                return Ok(None); // table has no snapshots yet
            };
            if Some(latest) == state.snapshot_id {
                tracing::debug!(
                    target: "bifrost::iceberg",
                    table = %spec.table,
                    snapshot = ?snapshot,
                    "no snapshot change",
                );
                return Ok(None);
            }

            let batches = match spec.mode {
                // Full mode: re-read the whole table at the new snapshot.
                SyncMode::Full => {
                    let batches57 = table
                        .read()
                        .await
                        .map_err(|e| BifrostError::Source(format!("{}: {e}", spec.table)))?;
                    let batches = bridge_57_to_56(&batches57)?;
                    let rows: usize = batches.iter().map(|b| b.num_rows()).sum();
                    tracing::info!(
                        target: "bifrost::iceberg",
                        event = "iceberg_read",
                        table = %spec.table,
                        snapshot = ?snapshot,
                        rows,
                        "read iceberg snapshot (full)",
                    );
                    batches
                }
                // Incremental / CDC: the streaming delta path — read only the
                // data files added since the recorded snapshot. First round
                // (no cursor) reads from the beginning (whole table).
                SyncMode::Incremental | SyncMode::Cdc => {
                    // Reuse the table handle already opened above — reopening the
                    // same redb warehouse mid-call would deadlock (single-open).
                    let (batches, _plan) =
                        Self::delta_from_table(&table, &spec.table, state.snapshot_id, latest)
                            .await?;
                    batches
                }
            };

            let new_state = TableState {
                snapshot_id: Some(latest),
                ..state.clone()
            };
            Ok(Some((batches, new_state)))
        }
    }

    /// Bridge skade's batches to the workspace's — now an **identity clone**.
    ///
    /// knut and skade both build on arrow 58, so `skade::arrow_array::RecordBatch`
    /// *is* the workspace `RecordBatch` (same crate, same major). The old
    /// C-Data-Interface FFI transmute bridge (arrow 57 → arrow 56), the mirror of
    /// `graphar-flight`'s `bridge_56_to_57`, is gone: a plain `clone()` (Arc bump,
    /// no buffer copy, no FFI) hands the batch straight through.
    fn bridge_57_to_56(batches57: &[skade::arrow_array::RecordBatch]) -> Result<Vec<RecordBatch>> {
        batches57.iter().map(batch_57_to_56).collect()
    }

    /// Identity clone of one skade (arrow-58) batch to the workspace's arrow-58
    /// `RecordBatch` — same type, no FFI.
    fn batch_57_to_56(batch57: &skade::arrow_array::RecordBatch) -> Result<RecordBatch> {
        Ok(batch57.clone())
    }

    #[cfg(test)]
    mod tests {
        use super::*;
        use std::sync::Arc;

        #[tokio::test]
        async fn reads_iceberg_table_and_tracks_snapshot() {
            use skade::arrow_array::{Int64Array, RecordBatch as B57, StringArray};
            use skade::arrow_schema::{DataType, Field, Schema};

            // Stage an Iceberg table via skade (arrow 57).
            let dir = tempfile::tempdir().unwrap();
            let batch = B57::try_new(
                Arc::new(Schema::new(vec![
                    Field::new("id", DataType::Int64, false),
                    Field::new("name", DataType::Utf8, false),
                ])),
                vec![
                    Arc::new(Int64Array::from(vec![1, 2, 3])),
                    Arc::new(StringArray::from(vec!["a", "b", "c"])),
                ],
            )
            .unwrap();
            {
                // Drop the setup warehouse before reading — redb is single-open
                // and IcebergSource opens the same dir itself.
                let wh = skade::Warehouse::open(dir.path()).await.unwrap();
                let mut table = wh.table_or_create("people", &batch.schema()).await.unwrap();
                table.append(&[batch]).await.unwrap();
            }

            // Read it back through IcebergSource (arrow 56).
            let src = IcebergSource::new(dir.path());
            let spec = TableSpec {
                table: "people".into(),
                mode: SyncMode::Full,
            };
            let (batches, st) = src
                .fetch_new(&spec, &TableState::default())
                .await
                .unwrap()
                .expect("a snapshot to read");
            let rows = batches.iter().map(|b| b.num_rows()).sum::<usize>();
            assert_eq!(rows, 3);
            assert!(st.snapshot_id.is_some(), "tracked the snapshot id");

            // Re-poll with the recorded snapshot → no change.
            let idle_none = src.fetch_new(&spec, &st).await.unwrap().is_none();
            assert!(idle_none, "unchanged snapshot → None");
            crate::testmatrix::emit(
                crate::testmatrix::COMPONENT,
                "source::reads_iceberg_table_and_tracks_snapshot",
                rows == 3 && st.snapshot_id.is_some() && idle_none,
                &format!("read {} rows, tracked snapshot {:?}, idle→None", rows, st.snapshot_id),
            );
        }

        use skade::arrow_array::{Int64Array, RecordBatch as B57, StringArray};
        use skade::arrow_schema::{DataType, Field, Schema};

        fn people57(ids: &[i64]) -> B57 {
            let names: Vec<String> = ids.iter().map(|i| format!("p{i}")).collect();
            B57::try_new(
                Arc::new(Schema::new(vec![
                    Field::new("id", DataType::Int64, false),
                    Field::new("name", DataType::Utf8, false),
                ])),
                vec![
                    Arc::new(Int64Array::from(ids.to_vec())),
                    Arc::new(StringArray::from(names)),
                ],
            )
            .unwrap()
        }

        /// Append `ids` as one new snapshot; return its snapshot id.
        async fn append_snapshot(dir: &std::path::Path, ids: &[i64]) -> i64 {
            let wh = skade::Warehouse::open(dir).await.unwrap();
            let batch = people57(ids);
            let mut table = wh.table_or_create("people", &batch.schema()).await.unwrap();
            table.append(&[batch]).await.unwrap();
            table
                .inner()
                .metadata()
                .current_snapshot()
                .unwrap()
                .snapshot_id()
        }

        /// LAW inject-assert: a real local Iceberg warehouse, two appends, and
        /// the delta reads ONLY the second append's rows — not the whole table.
        #[tokio::test]
        async fn delta_reads_only_appended_rows() {
            let dir = tempfile::tempdir().unwrap();

            // Snapshot A: 3 rows.
            let snap_a = append_snapshot(dir.path(), &[1, 2, 3]).await;
            // Snapshot B: 2 MORE rows (a separate fast_append commit).
            let snap_b = append_snapshot(dir.path(), &[4, 5]).await;
            assert_ne!(snap_a, snap_b, "two distinct snapshots");

            let src = IcebergSource::new(dir.path());

            // From the beginning to A → all 3 rows.
            let (batches, plan) = src.ingest_delta("people", None, snap_a).await.unwrap();
            assert_eq!(batches.iter().map(|b| b.num_rows()).sum::<usize>(), 3);
            assert!(!plan.needs_full_reload);
            assert_eq!(plan.snapshots, vec![snap_a]);
            assert_eq!(plan.added_files.len(), 1, "A added exactly one data file");

            // THE DELTA: from A to B → ONLY the 2 new rows, not 5.
            let (delta, plan) = src.ingest_delta("people", Some(snap_a), snap_b).await.unwrap();
            assert_eq!(
                delta.iter().map(|b| b.num_rows()).sum::<usize>(),
                2,
                "delta is just B's 2 appended rows (not a full 5-row reload)",
            );
            assert!(!plan.needs_full_reload);
            assert_eq!(plan.snapshots, vec![snap_b], "only snapshot B is new");
            assert_eq!(plan.added_files.len(), 1, "B added exactly one data file");
            // The delta's ids are exactly {4,5}.
            let col = delta[0]
                .column(0)
                .as_any()
                .downcast_ref::<arrow_array::Int64Array>()
                .unwrap();
            let mut got: Vec<i64> = (0..col.len()).map(|i| col.value(i)).collect();
            got.sort();
            assert_eq!(got, vec![4, 5]);

            // Cursor already at B → empty delta, no files.
            let (none, plan) = src.ingest_delta("people", Some(snap_b), snap_b).await.unwrap();
            assert!(none.is_empty(), "caught-up cursor → no rows");
            assert!(plan.snapshots.is_empty() && plan.added_files.is_empty());
            crate::testmatrix::emit(
                crate::testmatrix::COMPONENT,
                "source::delta_reads_only_appended_rows",
                got == vec![4, 5] && none.is_empty(),
                &format!("A→B delta = ids {got:?} (2 rows, not full 5); caught-up→empty"),
            );
        }

        /// The `TableSource::fetch_new` Incremental path advances the cursor and
        /// ships only the new snapshot's rows on the second poll.
        #[tokio::test]
        async fn incremental_fetch_new_advances_cursor_and_ships_delta() {
            let dir = tempfile::tempdir().unwrap();
            append_snapshot(dir.path(), &[1, 2, 3]).await;

            let src = IcebergSource::new(dir.path());
            let spec = TableSpec {
                table: "people".into(),
                mode: SyncMode::Incremental,
            };

            // First poll (no cursor) → whole table (3 rows), cursor set.
            let (b1, st1) = src
                .fetch_new(&spec, &TableState::default())
                .await
                .unwrap()
                .expect("first snapshot");
            assert_eq!(b1.iter().map(|b| b.num_rows()).sum::<usize>(), 3);
            assert!(st1.snapshot_id.is_some());

            // Idle: unchanged snapshot → None.
            assert!(src.fetch_new(&spec, &st1).await.unwrap().is_none());

            // Append B (2 rows), poll again → ONLY the 2 new rows.
            append_snapshot(dir.path(), &[4, 5]).await;
            let (b2, st2) = src
                .fetch_new(&spec, &st1)
                .await
                .unwrap()
                .expect("delta snapshot");
            let b2_rows = b2.iter().map(|b| b.num_rows()).sum::<usize>();
            assert_eq!(b2_rows, 2, "incremental ships only the delta");
            assert_ne!(st2.snapshot_id, st1.snapshot_id, "cursor advanced");
            crate::testmatrix::emit(
                crate::testmatrix::COMPONENT,
                "source::incremental_fetch_new_advances_cursor_and_ships_delta",
                b2_rows == 2 && st2.snapshot_id != st1.snapshot_id,
                &format!("second poll shipped {b2_rows} delta rows; cursor advanced"),
            );
        }

        /// A three-step append chain (A→B→C): the delta from A to C spans BOTH
        /// new snapshots and reads only their files (not A's). Proves the
        /// lineage walk collects multiple snapshots in one delta.
        #[tokio::test]
        async fn delta_spans_multiple_snapshots() {
            let dir = tempfile::tempdir().unwrap();
            let snap_a = append_snapshot(dir.path(), &[1, 2]).await;
            let snap_b = append_snapshot(dir.path(), &[3, 4]).await;
            let snap_c = append_snapshot(dir.path(), &[5, 6]).await;

            let src = IcebergSource::new(dir.path());
            let (delta, plan) = src.ingest_delta("people", Some(snap_a), snap_c).await.unwrap();
            assert_eq!(
                delta.iter().map(|b| b.num_rows()).sum::<usize>(),
                4,
                "B+C rows only (4), not A's 2 and not a full 6",
            );
            let delta_rows = delta.iter().map(|b| b.num_rows()).sum::<usize>();
            assert_eq!(plan.snapshots, vec![snap_b, snap_c], "oldest-first lineage");
            assert_eq!(plan.added_files.len(), 2, "B and C each added one file");
            crate::testmatrix::emit(
                crate::testmatrix::COMPONENT,
                "source::delta_spans_multiple_snapshots",
                delta_rows == 4 && plan.snapshots == vec![snap_b, snap_c] && plan.added_files.len() == 2,
                &format!("A→C delta spans 2 snapshots, {delta_rows} rows, 2 files"),
            );
        }

        /// Delete `ids` from `people` via a real Iceberg equality-delete
        /// snapshot (skade), returning its snapshot id.
        async fn delete_snapshot(dir: &std::path::Path, ids: &[i64]) -> i64 {
            let wh = skade::Warehouse::open(dir).await.unwrap();
            let mut table = wh.table("people").await.unwrap();
            let key = B57::try_new(
                Arc::new(skade::arrow_schema::Schema::new(vec![
                    skade::arrow_schema::Field::new(
                        "id",
                        skade::arrow_schema::DataType::Int64,
                        false,
                    ),
                ])),
                vec![Arc::new(Int64Array::from(ids.to_vec()))],
            )
            .unwrap();
            table.delete_equality(&key, &["id"]).await.unwrap();
            table.current_snapshot_id().unwrap()
        }

        /// Collect (id, change_type) from a CDC delta's batches.
        fn id_change_pairs(batches: &[RecordBatch]) -> Vec<(i64, String)> {
            use arrow_array::StringArray;
            let mut out = Vec::new();
            for b in batches {
                let id = b
                    .column(b.schema().index_of("id").unwrap())
                    .as_any()
                    .downcast_ref::<arrow_array::Int64Array>()
                    .unwrap();
                let ct = b
                    .column(b.schema().index_of(SyncMode::CHANGE_TYPE_COLUMN).unwrap())
                    .as_any()
                    .downcast_ref::<StringArray>()
                    .unwrap();
                for i in 0..b.num_rows() {
                    out.push((id.value(i), ct.value(i).to_string()));
                }
            }
            out
        }

        /// CDC inject-assert: snapshot A inserts {1,2,3}; snapshot B is a real
        /// Iceberg equality-delete of id=2. The Incremental delta over A→B must
        /// emit CDC `_change_type` rows — id=2 tagged DELETE — so the listener
        /// removes that node. (The pure-Rust delete path, no Spark.)
        #[tokio::test]
        async fn cdc_delta_emits_change_type_delete_for_equality_delete() {
            let dir = tempfile::tempdir().unwrap();
            let snap_a = append_snapshot(dir.path(), &[1, 2, 3]).await;

            let src = IcebergSource::new(dir.path());

            // Pure-append delta carries NO _change_type (back-compat).
            let (appended, plan) = src.ingest_delta("people", None, snap_a).await.unwrap();
            assert!(plan.delete_files.is_empty(), "no deletes yet");
            assert!(
                appended
                    .iter()
                    .all(|b| b.schema().index_of(SyncMode::CHANGE_TYPE_COLUMN).is_err()),
                "an append-only delta is untagged (flows straight to upserts)",
            );

            // Snapshot B: equality-delete id=2.
            let snap_b = delete_snapshot(dir.path(), &[2]).await;
            assert_ne!(snap_a, snap_b);

            // THE CDC DELTA A→B: a _change_type column, id=2 = DELETE.
            let (delta, plan) = src.ingest_delta("people", Some(snap_a), snap_b).await.unwrap();
            assert!(!plan.needs_full_reload, "equality-delete needs no reload");
            assert_eq!(plan.delete_files.len(), 1, "one equality-delete file");
            assert!(plan.added_files.is_empty(), "the delete added no data files");
            let pairs = id_change_pairs(&delta);
            assert_eq!(
                pairs,
                vec![(2i64, "DELETE".to_string())],
                "the delta is exactly: delete id=2",
            );

            // And it routes through the listener as a real vertex delete.
            let change = crate::transform::partition_change_rows(&delta).unwrap();
            assert!(change.upserts.iter().all(|b| b.num_rows() == 0));
            assert_eq!(
                change.deletes.iter().map(|b| b.num_rows()).sum::<usize>(),
                1,
                "one delete row routed to the delete pipeline",
            );
            let vt: Arc<dyn crate::transform::Transform> =
                Arc::new(crate::transform::VertexTransform {
                    label: "Person".into(),
                    id_column: "id".into(),
                    properties: vec![],
                    rename: Default::default(),
                });
            let ctx = crate::transform::TableContext { table: "people".into() };
            let dels = crate::transform::deletes_from_change_rows(
                std::slice::from_ref(&vt),
                &ctx,
                &change.deletes,
            )
            .unwrap();
            assert_eq!(dels.len(), 1, "one VertexDelete emitted");
            assert_eq!(dels[0].label, "Person");
            crate::testmatrix::emit(
                crate::testmatrix::COMPONENT,
                "source::cdc_delta_emits_change_type_delete_for_equality_delete",
                pairs == vec![(2i64, "DELETE".to_string())] && dels.len() == 1 && dels[0].label == "Person",
                &format!("equality-delete A→B tagged id=2 DELETE → 1 VertexDelete; pairs={pairs:?}"),
            );
        }
    }
}

// NOTE: row-level deletes now flow through the pure-Rust path as Iceberg
// *equality* deletes (skade's `Table::delete_equality` writes them, `read_delta`
// surfaces them, and `delta_from_table` tags them `_change_type = DELETE`). The
// `needs_full_reload` branch remains for *position* deletes (delete-by-offset,
// unmappable to a graph identity) and overwrite/replace commits, which an
// additive delta can't express. skade can't *produce* those via `fast_append`,
// so that branch is covered by the code path + the design doc, not a unit test.