knut-bifrost 0.3.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::{RowFilter, SyncMode},
    state::TableState,
};

/// What the listener asks a source each round.
///
/// `columns` + `filter` are the **pushdown** half of the ask: they say what the
/// round actually needs, so a source that can prune (the Iceberg one) reads only
/// that. A source that cannot prune ignores them and returns everything — the
/// batches are then a superset, which every downstream transform already
/// tolerates (they project what they name).
#[derive(Debug, Clone, Default)]
pub struct TableSpec {
    pub table: String,
    pub mode: SyncMode,
    /// The projection: columns the round needs. **Empty = every column**, which
    /// is the pre-pushdown behaviour. Derived from the mapping's pipeline by
    /// [`TableMapping::projected_columns`](crate::mapping::TableMapping::projected_columns).
    pub columns: Vec<String>,
    /// An optional row predicate to push into the scan planner (partition / data
    /// file / row-group pruning). `None` = no filter.
    pub filter: Option<RowFilter>,
}

impl TableSpec {
    /// The pre-pushdown ask: this table, this mode, every column, no filter.
    pub fn new(table: impl Into<String>, mode: SyncMode) -> Self {
        Self {
            table: table.into(),
            mode,
            columns: Vec::new(),
            filter: None,
        }
    }

    /// Narrow the read to `columns` (empty leaves it unprojected).
    pub fn with_columns(mut self, columns: Vec<String>) -> Self {
        self.columns = columns;
        self
    }

    /// Push `filter` into the scan planner.
    pub fn with_filter(mut self, filter: Option<RowFilter>) -> Self {
        self.filter = filter;
        self
    }

    /// `columns` as the `&[&str]` the skade read API takes.
    pub fn column_refs(&self) -> Vec<&str> {
        self.columns.iter().map(String::as_str).collect()
    }
}

/// 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)>>;

    /// **Which CDC backend this source resolved to for `table`, and why.**
    ///
    /// The listener stamps this onto the table's [`TableState`] after every
    /// changelog round, so the selection — and any *fallback* — is readable out
    /// of `state_json` rather than only out of the log. `None` means the source
    /// has no CDC story at all (the file source); it never means "unknown".
    ///
    /// Default: `None`, so an out-of-tree source keeps compiling.
    fn cdc_capability(&self, table: &str) -> Option<crate::cdc::CdcCapability> {
        let _ = table;
        None
    }
}

pub use arrow_batches::{ArrowBatches, ArrowSource, EmptyRead};

mod arrow_batches {
    use super::*;
    use crate::error::BifrostError;

    /// **A caller-supplied producer of Arrow rows for one table.**
    ///
    /// This is the seam that lets *anything that already speaks Arrow* feed bifröst:
    /// a warehouse client, a Flight stream, an ADBC read, a query result held in
    /// memory. Implement it, wrap it in an [`ArrowSource`], and the whole mapping →
    /// transform → graph-sink pipeline runs unchanged.
    ///
    /// `spec.columns` is the round's projection ask. Honour it if the producer can
    /// prune; ignoring it is fine — the batches are then a superset, which every
    /// downstream transform already tolerates (they project what they name).
    ///
    /// [`crate::rel2graph::ArrowTables`] — the in-memory holder the OFFLINE engine
    /// already reads from — implements this, so one bag of Arrow batches drives both
    /// the offline `build_mutation` and the live [`crate::Listener`].
    #[async_trait]
    pub trait ArrowBatches: Send + Sync {
        /// Every row of `spec.table`, as Arrow.
        async fn batches(&self, spec: &TableSpec) -> Result<Vec<RecordBatch>>;

        /// The tables this producer knows about, when it can say. Used ONLY to make
        /// an empty read's error useful ("you asked for `orders`; I hold `Orders`").
        /// `None` means "cannot enumerate", which is honest for a live catalog.
        fn known_tables(&self) -> Option<Vec<String>> {
            None
        }
    }

    /// The in-memory Arrow holder is a live-listener producer too — ONE type, both
    /// paths. `rel2graph::ArrowTables` was already the crate's answer to "some Arrow
    /// I already have"; this is what lets it drive a round loop as well as a one-shot
    /// `build_mutation`.
    #[async_trait]
    impl ArrowBatches for crate::rel2graph::ArrowTables {
        async fn batches(&self, spec: &TableSpec) -> Result<Vec<RecordBatch>> {
            Ok(crate::rel2graph::TableSource::batches(self, &spec.table).to_vec())
        }

        fn known_tables(&self) -> Option<Vec<String>> {
            Some(self.table_names().map(str::to_string).collect())
        }
    }

    /// **What an Arrow read of ZERO rows means.**
    ///
    /// It is a policy, not a detail, because the two readings are opposite and both
    /// look identical on the wire:
    ///
    /// * "the table really is empty" — a legitimate `Ok`, nothing to build;
    /// * "I asked for a full build of `orders` and the catalog handed me nothing" —
    ///   a mis-pointed warehouse, a wrong namespace, an unrefreshed snapshot. Shipping
    ///   that as a successful round writes an EMPTY GRAPH that looks exactly like a
    ///   built one, and nobody finds out until someone opens the viewer.
    ///
    /// The default is [`EmptyRead::Refuse`], because the second reading is the common
    /// one and the first is easy to declare.
    #[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
    pub enum EmptyRead {
        /// A zero-row read is a typed [`BifrostError::Source`] naming the table and
        /// the producer. **The default.**
        #[default]
        Refuse,
        /// A zero-row read is `Ok(None)` — the caller has said this table may
        /// legitimately be empty.
        Allow,
    }

    /// **A [`TableSource`] over Arrow batches a caller already has.**
    ///
    /// `SparkConnectSource` needs a Spark Connect endpoint and `IcebergSource` needs
    /// a local skade warehouse *directory*. Neither can be pointed at a warehouse
    /// reached some other way — a remote skade over its Iceberg-REST front, an
    /// Iceberg REST catalog, a Spark whose client lives in the calling process. All
    /// of those already produce Arrow, and this is the source that takes it.
    ///
    /// ```no_run
    /// # async fn demo() -> knut_bifrost::Result<()> {
    /// use knut_bifrost::source::ArrowSource;
    /// # let batches = Vec::new();
    /// // A one-shot build from batches already in hand.
    /// let source = ArrowSource::from_batches("my-warehouse", "lake.db.people", batches);
    /// # let _ = source;
    /// # Ok(()) }
    /// ```
    ///
    /// ## What it is NOT
    ///
    /// **It is a batch read, not change-data-capture.** CDC and delta-watching in
    /// bifröst (`SparkIcebergCdc`, `SparkNativeCdc`, `watch_iceberg_delta`,
    /// `IcebergSource`'s incremental path) all rest on a *snapshot lineage* — a
    /// before-id and an after-id the engine can diff. A bag of Arrow batches has no
    /// such lineage and no engine behind it to ask, so:
    ///
    /// | [`SyncMode`] | what this source does |
    /// |---|---|
    /// | `Full` | re-reads the producer every round (idempotent MERGE in the sink makes that safe) |
    /// | `Incremental` | ships once, then reports nothing — there is no delta to compute |
    /// | `Cdc` | **refuses**, typed, naming the reason and the sources that can |
    ///
    /// [`cdc_capability`](TableSource::cdc_capability) is `None`, exactly as the file
    /// source's is. A warehouse reached this way feeds bifröst for a **full build**,
    /// not for streaming.
    pub struct ArrowSource {
        /// A short name for the producer behind this source — it appears in every
        /// error, so an operator reads *which* warehouse came back empty.
        producer: String,
        batches: Box<dyn ArrowBatches>,
        empty: EmptyRead,
    }

    impl ArrowSource {
        /// Wrap any [`ArrowBatches`] producer. `producer` names it for diagnostics
        /// (`"skade-remote"`, `"iceberg-rest"`, `"spark41"`, …).
        pub fn new(producer: impl Into<String>, batches: impl ArrowBatches + 'static) -> Self {
            Self {
                producer: producer.into(),
                batches: Box::new(batches),
                empty: EmptyRead::default(),
            }
        }

        /// A one-shot source over batches already in hand, for exactly one table.
        /// Sugar over [`crate::rel2graph::ArrowTables`] — the same holder the offline
        /// engine reads from, not a second one.
        pub fn from_batches(
            producer: impl Into<String>,
            table: impl Into<String>,
            batches: Vec<RecordBatch>,
        ) -> Self {
            Self::new(
                producer,
                crate::rel2graph::ArrowTables::new().with_table(table, batches),
            )
        }

        /// Declare that a zero-row read is legitimate for this source. Say it on
        /// purpose; the default refuses. See [`EmptyRead`].
        #[must_use]
        pub fn allowing_empty(mut self) -> Self {
            self.empty = EmptyRead::Allow;
            self
        }

        /// The producer name this source reports in its errors.
        pub fn producer(&self) -> &str {
            &self.producer
        }

        /// What a zero-row read means here.
        pub fn empty_policy(&self) -> EmptyRead {
            self.empty
        }
    }

    #[async_trait]
    impl TableSource for ArrowSource {
        async fn fetch_new(
            &self,
            spec: &TableSpec,
            state: &TableState,
        ) -> Result<Option<(Vec<RecordBatch>, TableState)>> {
            // CDC is refused BEFORE the read: a source with no snapshot lineage must
            // not answer a changelog ask with an insert-only batch that the listener
            // would then treat as the whole truth.
            if matches!(spec.mode, SyncMode::Cdc) {
                return Err(BifrostError::Source(format!(
                    "table '{}': an Arrow batch source ({}) has no change feed — it is a \
                     BATCH read with no snapshot lineage to diff. Use SyncMode::Full for a \
                     complete rebuild, or a source with an engine behind it \
                     (SparkIcebergCdc, SparkNativeCdc, or IcebergSource over a local skade \
                     warehouse) for CDC.",
                    spec.table, self.producer
                )));
            }

            // Incremental over a batch source ships exactly once: there is no delta.
            if matches!(spec.mode, SyncMode::Incremental) && state.synced_at.is_some() {
                return Ok(None);
            }

            let batches = self.batches.batches(spec).await?;
            let rows: usize = batches.iter().map(RecordBatch::num_rows).sum();

            if rows == 0 {
                // **The gate.** An empty read is not a built graph.
                return match self.empty {
                    EmptyRead::Refuse => {
                        let holds = match self.batches.known_tables() {
                            Some(t) if !t.is_empty() => format!(" It holds {t:?}."),
                            Some(_) => " It holds no tables at all.".to_string(),
                            None => String::new(),
                        };
                        Err(BifrostError::Source(format!(
                            "table '{}': the Arrow producer ({}) returned {} batch(es) and 0 \
                             rows.{holds} An empty read is refused here rather than sunk as an \
                             empty graph that looks like a successful build — check the \
                             catalog, the namespace and the snapshot. If this table really is \
                             empty, say so with ArrowSource::allowing_empty().",
                            spec.table,
                            self.producer,
                            batches.len()
                        )))
                    }
                    EmptyRead::Allow => Ok(None),
                };
            }

            // The state is handed back unchanged: a batch producer has no snapshot
            // id, no watermark and no file ledger to record. `SyncState::advance`
            // stamps `synced_at` after the sink succeeds, and THAT is what the
            // Incremental "ship once" check above reads — so the marker is written by
            // the round that actually landed, not by the read that merely started.
            Ok(Some((batches, state.clone())))
        }

        /// No CDC story at all — the same honest `None` the file source gives.
        fn cdc_capability(&self, _table: &str) -> Option<crate::cdc::CdcCapability> {
            None
        }
    }
}

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()
    }

    /// The projected + filtered `SELECT` over a materialised Iceberg changelog
    /// view — the CDC path's half of the pushdown that landed for `Full`.
    ///
    /// An empty projection **and** no filter reproduces the old `SELECT * FROM
    /// view` byte for byte, so the narrowing can only ever apply where it was
    /// asked for. When the spec does project, the changelog metadata columns
    /// ride along regardless (`_change_type` is load-bearing for delete
    /// routing; `_change_ordinal` / `_commit_snapshot_id` carry the order and
    /// the snapshot a renderable feed needs), as does the filter's own column.
    ///
    /// Built on the shared [`crate::cdc`] vocabulary rather than a second copy
    /// of the SQL: this is the same statement `cdc::SparkIcebergCdc` issues.
    fn changelog_select(spec: &TableSpec, view: &str) -> Result<String> {
        crate::cdc::SparkIcebergCdc::select_statement(&change_request(spec), view)
    }

    /// The [`crate::cdc::ChangeRequest`] a [`TableSpec`] describes — one place
    /// the two pushdown vocabularies meet, so the plain-SQL path and the CDC
    /// seam can never drift apart on what "projected" means.
    fn change_request(spec: &TableSpec) -> crate::cdc::ChangeRequest {
        crate::cdc::ChangeRequest::new(spec.table.clone(), None)
            .with_columns(spec.columns.clone())
            .with_filter(spec.filter.clone())
    }

    /// `spec.columns` as a Spark select-list, or `*` when unprojected.
    fn select_list(spec: &TableSpec) -> String {
        if spec.columns.is_empty() {
            return "*".into();
        }
        let mut cols = spec.columns.clone();
        // The predicate is evaluated by Spark against this same relation, so its
        // column must survive the projection.
        if let Some(f) = &spec.filter {
            if !cols.iter().any(|c| c == &f.column) {
                cols.push(f.column.clone());
            }
        }
        cols.iter()
            .map(|c| format!("`{}`", c.replace('`', "``")))
            .collect::<Vec<_>>()
            .join(", ")
    }

    /// `spec.filter` as a ` WHERE …` fragment (empty when there is none).
    fn where_clause(spec: &TableSpec) -> Result<String> {
        crate::cdc::spark::where_clause_for(&change_request(spec))
    }

    /// 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: std::sync::Arc<SparkSession>,
        /// Per-table CDC verdict, resolved ONCE on the first changelog round
        /// (detection costs a version request plus a zero-row probe, so it must
        /// not run every poll) and reused after that. Recorded into
        /// [`TableState::cdc`] so `state_json` shows which backend ran.
        cdc: std::sync::Mutex<std::collections::HashMap<String, crate::cdc::CdcCapability>>,
    }

    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: std::sync::Arc::new(session),
                cdc: std::sync::Mutex::new(std::collections::HashMap::new()),
            })
        }

        /// **The CDC verdict for `table`** — detected on first use, cached after.
        ///
        /// [`crate::cdc::detect`] asks the server its version and then *probes*
        /// the table with a zero-row `CHANGES` read, because a 4.2 engine does
        /// not imply a 4.2-capable connector. The verdict is logged once (with
        /// `fell_back` set when the native path was expected but unusable) and
        /// then reused, so a poll loop pays for detection exactly once.
        pub async fn cdc_backend(
            &self,
            table: &str,
        ) -> (
            std::sync::Arc<dyn crate::cdc::CdcBackend>,
            crate::cdc::CdcCapability,
        ) {
            if let Some(cap) = self.cdc.lock().ok().and_then(|m| m.get(table).cloned()) {
                let backend: std::sync::Arc<dyn crate::cdc::CdcBackend> = match cap.chosen {
                    crate::cdc::CdcKind::SparkNative => std::sync::Arc::new(
                        crate::cdc::SparkNativeCdc::new(self.session.clone(), cap.clone()),
                    ),
                    _ => std::sync::Arc::new(crate::cdc::SparkIcebergCdc::new(
                        self.session.clone(),
                        cap.clone(),
                    )),
                };
                return (backend, cap);
            }
            let (backend, cap) = crate::cdc::backend_for(self.session.clone(), table).await;
            if let Ok(mut m) = self.cdc.lock() {
                m.insert(table.to_string(), cap.clone());
            }
            (backend, cap)
        }

        /// **The renderable change feed** a UI consumes: `(op, before, after,
        /// snapshot, ts)` batches over a [`crate::cdc::ChangeStream`], served by
        /// whichever backend detection chose. This is the UI-facing twin of the
        /// listener's wide-batch [`Self::read_changelog`] path.
        pub async fn changes(
            &self,
            req: &crate::cdc::ChangeRequest,
        ) -> Result<crate::cdc::ChangeStream> {
            let (backend, _) = self.cdc_backend(&req.table).await;
            backend.changes(req).await
        }

        /// Run arbitrary SQL on the Spark Connect session (DDL, admin,
        /// test fixtures). Returns the result batches directly — the
        /// `spark-connect-arrow59` fork shares the workspace arrow major, so
        /// `collect()` already yields workspace `RecordBatch`es (no bridge).
        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}")))?;
            self.session
                .collect(plan)
                .await
                .map_err(|e| BifrostError::Source(format!("{query}: {e}")))
        }

        /// 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.
        ///
        /// **Projected and filtered.** `spec.columns` / `spec.filter` are pushed
        /// into the `SELECT` over the materialised view instead of the old
        /// `SELECT *` — closing the gap where `SyncMode::Cdc` stayed full-width
        /// while `SyncMode::Full` had gained pushdown. `_change_type` is always
        /// retained: without it the listener cannot route a delete. An
        /// unprojected, unfiltered spec still produces exactly `SELECT *`.
        async fn read_changelog(
            &self,
            spec: &TableSpec,
            prev: i64,
            latest: i64,
        ) -> Result<Vec<RecordBatch>> {
            let table = &spec.table;
            // WHICH CDC. Detected from the live server (version + a real probe),
            // not from config — and cached, so this costs one round trip ever.
            let (_, cap) = self.cdc_backend(table).await;
            let req = crate::cdc::ChangeRequest::new(table.clone(), Some(prev))
                .to(Some(latest))
                .with_columns(spec.columns.clone())
                .with_filter(spec.filter.clone());

            let batches = if cap.chosen == crate::cdc::CdcKind::SparkNative {
                // Spark 4.2's own CDC: one statement, no temp view. The rows come
                // back with Spark's lower-case `_change_type` spelling, which the
                // listener's router understands because both dialects go through
                // `cdc::ChangeOp::parse`.
                self.sql(&crate::cdc::SparkNativeCdc::statement(&req)?)
                    .await?
            } else {
                // Iceberg's CDC: materialise the changelog view, then read it
                // PROJECTED + FILTERED (it used to be `SELECT *`).
                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 out = self.sql(&changelog_select(spec, &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;
                out
            };

            let rows: usize = batches.iter().map(|b| b.num_rows()).sum();
            tracing::info!(
                target: "bifrost::spark",
                event = "changelog_read",
                table = %table,
                backend = cap.chosen.as_str(),
                fell_back = cap.fell_back,
                from = prev,
                to = latest,
                projected = spec.columns.len(),
                filtered = spec.filter.is_some(),
                change_rows = rows,
                "read changelog (pushdown)",
            );
            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 {
        /// The cached verdict for `table` (`None` until the first changelog
        /// round has run detection — a `Full`-only table never probes).
        fn cdc_capability(&self, table: &str) -> Option<crate::cdc::CdcCapability> {
            self.cdc.lock().ok().and_then(|m| m.get(table).cloned())
        }

        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, prev, latest).await?;
                let new_state = TableState {
                    snapshot_id: Some(latest),
                    ..state.clone()
                };
                return Ok(Some((batches, new_state)));
            }

            // The projection, as a SQL select-list. Empty projection → `*`, so an
            // unprojected mapping issues exactly the query it always did. The
            // `EXCEPT ALL` incremental diff **must** use the same list on both
            // sides or the set difference compares different shapes — building it
            // once here is what guarantees that.
            let cols = select_list(spec);
            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 {cols}, 'INSERT' AS {ct} FROM {t} VERSION AS OF {latest}{w}",
                    ct = SyncMode::CHANGE_TYPE_COLUMN,
                    t = spec.table,
                    w = where_clause(spec)?,
                ),
                (SyncMode::Full, _) | (SyncMode::Incremental, None) => format!(
                    "SELECT {cols} FROM {t} VERSION AS OF {latest}{w}",
                    t = spec.table,
                    w = where_clause(spec)?,
                ),
                // 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 {cols} FROM {t} VERSION AS OF {latest}{w} \
                     EXCEPT ALL \
                     SELECT {cols} FROM {t} VERSION AS OF {prev}{w}",
                    t = spec.table,
                    w = where_clause(spec)?,
                ),
                // 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)))
        }
    }

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

        /// RED-WHEN-BROKEN version guard (container-free): the vendored
        /// `spark-connect` (the `spark-connect-arrow59` fork) must yield knut's
        /// **workspace** `arrow_array::RecordBatch` from `SparkSession::collect`,
        /// so `SparkConnectSource` hands Spark batches straight to the sink with
        /// no IPC / arrow-version bridge. If the fork ever regresses to an older
        /// arrow major, `collect`'s payload becomes a *foreign* `RecordBatch`
        /// type and this guard fails to compile — the whole no-bridge spark path
        /// breaks with it. Never called; it only has to type-check.
        #[allow(dead_code)]
        fn _spark_collect_is_workspace_arrow59() -> impl std::future::Future<Output = ()> {
            async {
                // A generic never-returning helper rather than `unreachable!()` in
                // each initializer: `unreachable!()` diverges, so everything after
                // the first binding was unreachable code and `session` counted as
                // unused. The guard type-checked anyway (unreachable code still
                // is), but it warned on every build, and a type guard nobody can
                // build cleanly is one people start ignoring. The bindings below
                // are what actually pins the types; this only produces them.
                fn nowhere<T>() -> T {
                    unreachable!("compile-only type guard — never called")
                }
                let session: SparkSession = nowhere();
                let plan: spark_connect::spark::Plan = nowhere();
                // The binding pins the element type to the workspace's arrow-59
                // `RecordBatch`; only compiles while spark-connect shares it.
                let _batches: Vec<arrow_array::RecordBatch> = session.collect(plan).await.unwrap();
            }
        }

        /// Runtime mirror of the compile guard: the `RecordBatch` type
        /// `spark-connect` names (`arrow::record_batch::RecordBatch`) is the same
        /// type as knut's `arrow_array::RecordBatch` at the shared arrow-59 major.
        #[test]
        fn spark_connect_shares_workspace_arrow59() {
            use std::any::TypeId;
            let ok = TypeId::of::<arrow::record_batch::RecordBatch>()
                == TypeId::of::<arrow_array::RecordBatch>();
            assert!(
                ok,
                "spark-connect's arrow RecordBatch must equal knut's workspace \
                 arrow_array::RecordBatch (both arrow 59) — no bridge on the spark path",
            );
            crate::testmatrix::emit(
                crate::testmatrix::COMPONENT,
                "source::spark_connect_shares_workspace_arrow59",
                ok,
                "spark-connect (arrow59 fork) yields the workspace arrow-59 RecordBatch, no bridge",
            );
        }
    }
}

#[cfg(feature = "skade")]
pub use iceberg::IcebergSource;
/// The delta plan an [`IcebergSource`] returns IS skade's: bifrost re-exports the
/// canonical [`skade::DeltaPlan`] rather than forking its own copy, so the shared
/// CDC changelog vocabulary never diverges.
#[cfg(feature = "skade")]
pub use skade::DeltaPlan;

/// 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::delivery::IcebergLake;
    use crate::error::BifrostError;
    use skade::iceberg::table::Table as IceTable;
    // skade's canonical CDC changelog surface — the vocabulary bifrost used to
    // keep a private copy of. `Table::read_changelog` builds the `ChangelogBatch`
    // (INSERT/DELETE tagging, delete-identity widening, full-reload fallback);
    // `change_type::COLUMN` is the `_change_type` sentinel; `DeltaPlan` is the
    // engine-side snapshot diff bifrost re-exports.
    use skade::{ChangelogBatch, DeltaPlan, change_type};

    /// 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 and knut both build on Apache Arrow 58 (one workspace major), so
    /// `skade::arrow_array::RecordBatch` *is* the workspace `RecordBatch`: the
    /// read batches hand straight through, no version bridge (see
    /// [`from_skade_batches`], now an identity clone).
    ///
    /// **Where the tables are** is an [`IcebergLake`]: an embedded warehouse
    /// ([`Self::new`]), or — [`Self::from_lake`] — Iceberg tables on S3/MinIO read
    /// with no catalog, or an Iceberg REST catalog such as MinIO AIStor's. Every
    /// read below works on the loaded `iceberg` table, so all three stream alike.
    pub struct IcebergSource {
        lake: IcebergLake,
    }

    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::from_lake(IcebergLake::Warehouse(dir.into()))
        }

        /// Read tables from any [`IcebergLake`]; a `TableSpec.table` is resolved
        /// by [`IcebergLake::locate`].
        pub fn from_lake(lake: IcebergLake) -> Self {
            Self { lake }
        }

        /// Load the table at its current metadata. Loaded again on every call,
        /// so each poll sees the lake's latest commit; an embedded warehouse is
        /// opened and released inside the call (redb is single-open).
        async fn open_table(&self, table: &str) -> Result<IceTable> {
            self.lake.locate(table).load().await
        }

        /// **The delta primitive.** Read the changes to `table` between the
        /// snapshots strictly after `from_snapshot` up to (and including)
        /// `to_snapshot` into Arrow-58 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?;
            let spec = TableSpec::new(table_name, SyncMode::Cdc);
            Self::delta_from_table(&table, &spec, from_snapshot, to_snapshot).await
        }

        /// The delta over an already-loaded iceberg 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).
        ///
        /// **Projected.** `spec.columns` narrows the changelog through skade's own
        /// [`skade::ChangelogBatch::project_columns`] — the pushdown
        /// `SyncMode::Incremental` / `SyncMode::Cdc` never had, because
        /// `read_changelog` takes no projection and knut called it unprojected.
        /// An empty projection reproduces the previous behaviour exactly.
        ///
        /// **`_change_type` survives the projection**, which is what makes this
        /// safe for `Cdc`: `ChangelogBatch::project_columns` re-adds the sentinel
        /// itself, so the listener's `partition_change_rows` still sees which
        /// rows are deletes. (`TableMapping::projected_columns` does not name it —
        /// skade does.) `tests/cdc_roundtrip.rs::source_cdc_mode_is_projected_and_agrees_with_the_full_width_read`
        /// asserts the `_change_type` column is byte-identical projected vs full.
        ///
        /// Accounting, stated rather than implied: `project_columns` is a
        /// **post-decode** slice (skade's `read_delta` decodes whole parquet
        /// files). The win is in the encode, the transforms, the sink and the
        /// memory — *not* in parquet I/O. Pushing the projection into the
        /// manifest-delta scan is skade-side work. The predicate is applied by
        /// the caller (`fetch_new`), which owns the null policy.
        async fn delta_from_table(
            table: &IceTable,
            spec: &TableSpec,
            from_snapshot: Option<i64>,
            to_snapshot: i64,
        ) -> Result<(Vec<RecordBatch>, DeltaPlan)> {
            let table_name = &spec.table;
            // skade's changelog encoder is the single source of truth for the
            // snapshot diff AND the CDC re-tagging: `read_changelog` walks the
            // manifest lineage (`read_delta`), tags appended rows INSERT and
            // equality-deleted identities DELETE (widening them to the table
            // schema), and on an overwrite/position-delete window re-reads the
            // whole `to` snapshot as an INSERT stream. knut no longer keeps a
            // private copy of any of that.
            let cl: ChangelogBatch = skade::read_changelog(table, from_snapshot, to_snapshot)
                .await
                .map_err(|e| BifrostError::Source(format!("{table_name}: {e}")))?;
            let plan = cl.plan.clone();

            // THE PROJECTION. skade keeps `_change_type` for us (its
            // `project_columns` re-adds the sentinel), so the listener's delete
            // routing survives the narrowing. The filter's column is carried in
            // by `engine_columns` so the row prune below can still see it.
            let req = crate::cdc::ChangeRequest::new(table_name.clone(), from_snapshot)
                .with_columns(spec.columns.clone())
                .with_filter(spec.filter.clone());
            let engine_cols = req.engine_columns();
            let cl = if engine_cols.is_empty() {
                cl
            } else {
                let refs: Vec<&str> = engine_cols.iter().map(String::as_str).collect();
                cl.project_columns(&refs)
                    .map_err(|e| BifrostError::Source(format!("{table_name}: {e}")))?
            };
            // The PREDICATE is deliberately NOT applied here: `fetch_new` owns
            // the residual filter (one implementation, `mapping::apply_row_filter_with`)
            // and it is the caller that knows which null policy this batch needs.
            let cl_rows: Vec<RecordBatch> = cl.rows;

            // bifrost's contract: a pure-append (or full-reload) window flows
            // through UNTAGGED — the rows go straight to the upsert pipeline, so
            // strip the `_change_type` column skade tags every row with.
            //
            // WHICH window carries deletes is read off the ROWS, not off
            // `plan.delete_files`: those are only the equality-delete files, and
            // a copy-on-write rewrite, a position delete or a deletion vector
            // deletes rows with none — its DELETE rows then lost their tag, became
            // upserts, and put the deleted row back into the graph.
            //
            // Both delta modes apply deletes: `incremental` has deleted on an
            // equality delete since the skade source existed
            // (tests/delta_iceberg_falkordb.rs), and now does on every kind of
            // delete — on an Iceberg source it reads the same changelog as `cdc`.
            let has_deletes = cl_rows.iter().any(has_non_insert);
            if !has_deletes {
                let out: Vec<RecordBatch> = cl_rows
                    .iter()
                    .map(strip_change_type)
                    .collect::<Result<_>>()?;
                let rows: usize = out.iter().map(|b| b.num_rows()).sum();
                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(),
                    full_reload = plan.needs_full_reload,
                    projected = spec.columns.len(),
                    post_filtered = spec.filter.is_some(),
                    rows,
                    "read iceberg delta (appended/full rows, no CDC deletes)",
                );
                return Ok((out, plan));
            }

            // CDC delta: keep skade's INSERT/DELETE tagging verbatim — the
            // listener's `partition_change_rows` splits upserts from deletes on
            // the `_change_type` column (skade already widened DELETE rows to the
            // table schema so the vertex transform projects them).
            let change_rows: usize = cl_rows.iter().map(|b| b.num_rows()).sum();
            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(),
                projected = spec.columns.len(),
                post_filtered = spec.filter.is_some(),
                change_rows,
                "read iceberg CDC delta (inserts + identity deletes, tagged _change_type)",
            );
            Ok((cl_rows, plan))
        }

        /// **The projected / filtered snapshot read** — the `SyncMode::Full`
        /// path, and the point where knut stopped paying for a full-width scan.
        ///
        /// Before this, `Full` called `Table::read()`: every column of every data
        /// file, always. skade has carried a pruning read surface for a while
        /// (`read_columns` pushes the projection into the row-group readers so the
        /// dropped columns are never decompressed or allocated; `read_filtered`
        /// pushes a predicate into the scan planner so non-matching data files are
        /// never opened) and knut called NONE of it. This routes the ask:
        ///
        /// | `spec.columns` | `spec.filter` | skade call |
        /// |---|---|---|
        /// | empty | `None` | `read()` — unchanged full scan |
        /// | set | `None` | `read_columns(cols)` |
        /// | any | `Some(f)` | `read_filtered(f, cols)` (empty cols = all) |
        ///
        /// So a mapping that names no columns and no filter behaves EXACTLY as
        /// before; the pruning only ever narrows a read that was provably safe to
        /// narrow.
        ///
        /// When a filter is present the plan is also costed with
        /// [`skade::Table::plan_stats`] — the filtered plan against the
        /// whole-table baseline — so the data-skipping win is *observable*
        /// (files/rows skipped) rather than merely asserted. That is plan-time
        /// metadata work; it opens no data file.
        async fn read_snapshot(
            table: &IceTable,
            spec: &TableSpec,
            snapshot: Option<i64>,
        ) -> Result<Vec<RecordBatch>> {
            let cols = spec.column_refs();
            let src = |e: skade::SkadeError| BifrostError::Source(format!("{}: {e}", spec.table));

            let (skade_batches, planned) = match &spec.filter {
                Some(f) => {
                    let sf = to_scan_filter(f)?;
                    // Plan-only: how much of the table survives the predicate,
                    // and how much the unfiltered plan would have read.
                    let after = skade::plan_stats(&table, Some(&sf)).await.map_err(src)?;
                    let before = skade::plan_stats(&table, None).await.map_err(src)?;
                    let batches = skade::read_filtered(&table, &sf, &cols).await.map_err(src)?;
                    (batches, Some((before, after)))
                }
                None if !cols.is_empty() => (skade::read_columns(&table, &cols).await.map_err(src)?, None),
                None => (skade::read_all(&table).await.map_err(src)?, None),
            };

            let batches = from_skade_batches(&skade_batches)?;
            let rows: usize = batches.iter().map(|b| b.num_rows()).sum();
            let width = batches.first().map(|b| b.num_columns()).unwrap_or(0);

            match planned {
                Some((before, after)) => {
                    tracing::info!(
                        target: "bifrost::iceberg",
                        event = "iceberg_read",
                        table = %spec.table,
                        snapshot = ?snapshot,
                        rows,
                        columns = width,
                        projected = cols.len(),
                        files_planned = after.data_files,
                        files_baseline = before.data_files,
                        files_skipped = before.data_files.saturating_sub(after.data_files),
                        rows_planned = after.rows_planned,
                        rows_baseline = before.rows_planned,
                        "read iceberg snapshot (full, pushdown)",
                    );
                    crate::testmatrix::emit(
                        crate::testmatrix::COMPONENT,
                        "source::iceberg_pushdown_read",
                        true,
                        &format!(
                            "{} data file(s) skipped of {}, {} column(s) projected",
                            before.data_files.saturating_sub(after.data_files),
                            before.data_files,
                            cols.len()
                        ),
                    );
                }
                None => tracing::info!(
                    target: "bifrost::iceberg",
                    event = "iceberg_read",
                    table = %spec.table,
                    snapshot = ?snapshot,
                    rows,
                    columns = width,
                    projected = cols.len(),
                    "read iceberg snapshot (full)",
                ),
            }
            Ok(batches)
        }

        /// **Plan-time data-skipping stats** for `table` under `filter` — how many
        /// data files / rows a scan would open, WITHOUT reading any of them
        /// (`filter = None` plans the whole table, the baseline to diff against).
        ///
        /// Exposed so a caller — a test, a bench, a planner UI — can measure the
        /// pruning win directly instead of inferring it from wall time.
        pub async fn plan_stats(
            &self,
            table_name: &str,
            filter: Option<&RowFilter>,
        ) -> Result<skade::ScanPlanStats> {
            let table = self.open_table(table_name).await?;
            let sf = filter.map(to_scan_filter).transpose()?;
            skade::plan_stats(&table, sf.as_ref())
                .await
                .map_err(|e| BifrostError::Source(format!("{table_name}: {e}")))
        }

        /// 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
                .metadata()
                .current_snapshot()
                .map(|s| s.snapshot_id()))
        }
    }

    /// Lower bifrost's backend-neutral [`RowFilter`] onto skade's
    /// [`skade::ScanFilter`]. The spec type is engine-free (bifrost's core has
    /// no Iceberg dependency); this is the ONE place the two vocabularies meet.
    fn to_scan_filter(f: &RowFilter) -> Result<skade::ScanFilter> {
        f.validate()?;
        if let Some(v) = &f.eq {
            return Ok(skade::ScanFilter::eq(&f.column, to_scalar(v)));
        }
        if let Some(vs) = &f.is_in {
            return Ok(skade::ScanFilter::is_in(
                &f.column,
                vs.iter().map(to_scalar).collect::<Vec<_>>(),
            ));
        }
        Ok(skade::ScanFilter::range(
            &f.column,
            f.ge.as_ref().map(to_scalar),
            f.le.as_ref().map(to_scalar),
        ))
    }

    fn to_scalar(v: &crate::mapping::FilterValue) -> skade::Scalar {
        use crate::mapping::FilterValue as F;
        match v {
            F::Str(s) => skade::Scalar::Str(s.clone()),
            F::Int(i) => skade::Scalar::I64(*i),
            F::Float(f) => skade::Scalar::F64(*f),
            F::Bool(b) => skade::Scalar::Bool(*b),
        }
    }

    /// Drop the `_change_type` column skade's changelog tags every row with, so a
    /// pure-append / full-reload window returns batches shaped exactly like an
    /// ordinary table read — they flow straight to the upsert pipeline, matching
    /// bifrost's pre-cutover behaviour. A batch without the column (defensive)
    /// passes through untouched.
    /// Whether `batch` holds a row that is not an insert (a DELETE, or an update's
    /// pre-image).
    fn has_non_insert(batch: &RecordBatch) -> bool {
        let Some(col) = batch.column_by_name(change_type::COLUMN) else { return false };
        let Some(ops) = col.as_any().downcast_ref::<arrow_array::StringArray>() else { return false };
        ops.iter().any(|op| !matches!(op, Some(change_type::INSERT) | Some(change_type::UPDATE_AFTER)))
    }

    fn strip_change_type(batch: &RecordBatch) -> Result<RecordBatch> {
        let Ok(idx) = batch.schema().index_of(change_type::COLUMN) else {
            return Ok(batch.clone());
        };
        let mut fields: Vec<arrow_schema::Field> = batch
            .schema()
            .fields()
            .iter()
            .map(|f| f.as_ref().clone())
            .collect();
        fields.remove(idx);
        let mut columns: Vec<arrow_array::ArrayRef> = batch.columns().to_vec();
        columns.remove(idx);
        RecordBatch::try_new(
            std::sync::Arc::new(arrow_schema::Schema::new(fields)),
            columns,
        )
        .map_err(|e| BifrostError::Source(format!("strip _change_type: {e}")))
    }

    #[async_trait]
    impl TableSource for IcebergSource {
        /// Always skade: this source holds an embedded warehouse, so there is no
        /// Spark to ask and nothing to fall back from. Stated explicitly rather
        /// than left `None`, so `state_json` names the path on this route too.
        fn cdc_capability(&self, _table: &str) -> Option<crate::cdc::CdcCapability> {
            Some(crate::cdc::CdcCapability::skade())
        }

        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
                .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 table at the new snapshot — PROJECTED
                // and FILTERED to what the round asked for (see `read_snapshot`).
                SyncMode::Full => Self::read_snapshot(&table, spec, snapshot).await?,
                // 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, state.snapshot_id, latest).await?;
                    // ── the predicate means the SAME thing in every mode ──────────
                    //
                    // The delta scan plans on the manifest (which files were ADDED
                    // since the cursor), not on the predicate, so it cannot prune on
                    // `spec.filter` the way `read_snapshot` does. Before this it
                    // simply DROPPED the filter — and nothing downstream re-applies
                    // it, so rows the operator excluded were ingested anyway. The
                    // same mapping selected 1000 rows under `Incremental` and 250
                    // under `Full` (tests/iceberg_pushdown.rs).
                    //
                    // Falling back to the in-memory residual keeps the MEANING
                    // identical and only changes the cost.
                    //
                    // UPDATED: the PROJECTION is no longer dropped here either —
                    // `delta_from_table` now pushes `spec.columns` through skade's
                    // `ChangelogBatch::project_columns`, which re-adds the
                    // `_change_type` sentinel itself, so Cdc's
                    // `partition_change_rows` still sees which rows are deletes.
                    // (The earlier note that projecting the delta path "would
                    // strip it" was true of a naive projection, not of skade's.)
                    //
                    // NULL POLICY. A changelog batch's `DELETE` rows are widened
                    // to the table schema with **nulls in every non-identity
                    // column**, so the scan-planner default (`FilterNulls::Drop`,
                    // NULL fails the predicate) would silently remove deletions —
                    // the one CDC failure that must never happen quietly. A batch
                    // carrying `_change_type` therefore keeps what the predicate
                    // cannot judge; an ordinary delta keeps the planner semantics
                    // so `Full` and `Incremental` select the same rows.
                    match &spec.filter {
                        Some(f) => {
                            let before: usize = batches.iter().map(|b| b.num_rows()).sum();
                            let nulls = if batches.iter().any(crate::transform::is_changelog_batch)
                            {
                                crate::mapping::FilterNulls::Keep
                            } else {
                                crate::mapping::FilterNulls::Drop
                            };
                            let kept = crate::mapping::apply_row_filter_with(&batches, f, nulls)?;
                            let after: usize = kept.iter().map(|b| b.num_rows()).sum();
                            tracing::info!(
                                target: "bifrost::iceberg",
                                event = "delta_residual_filter",
                                table = %spec.table,
                                mode = ?spec.mode,
                                column = %f.column,
                                rows_before = before,
                                rows_after = after,
                                rows_dropped = before.saturating_sub(after),
                                "applied the mapping's row filter to the delta batches \
                                 (residual — the delta scan plans on the manifest, not the \
                                 predicate)",
                            );
                            crate::testmatrix::emit(
                                crate::testmatrix::COMPONENT,
                                "source::delta_residual_filter",
                                true,
                                &format!(
                                    "{} rows -> {} ({} dropped) by `{}` in {:?} mode",
                                    before,
                                    after,
                                    before.saturating_sub(after),
                                    f.column,
                                    spec.mode
                                ),
                            );
                            kept
                        }
                        None => 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 59, 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 `to_skade_batch`, is gone: a plain `clone()` (Arc bump,
    /// no buffer copy, no FFI) hands the batch straight through.
    fn from_skade_batches(
        skade_batches: &[skade::arrow_array::RecordBatch],
    ) -> Result<Vec<RecordBatch>> {
        skade_batches.iter().map(from_skade_batch).collect()
    }

    /// Identity clone of one skade (arrow-59) batch to the workspace's arrow-59
    /// `RecordBatch` — same type, no FFI.
    fn from_skade_batch(skade_batch: &skade::arrow_array::RecordBatch) -> Result<RecordBatch> {
        Ok(skade_batch.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 SkadeBatch, StringArray};
            use skade::arrow_schema::{DataType, Field, Schema};

            // Stage an Iceberg table via skade (arrow 59).
            let dir = tempfile::tempdir().unwrap();
            let batch = SkadeBatch::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 59).
            let src = IcebergSource::new(dir.path());
            let spec = TableSpec {
                table: "people".into(),
                mode: SyncMode::Full,
                ..Default::default()
            };
            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 SkadeBatch, StringArray};
        use skade::arrow_schema::{DataType, Field, Schema};

        fn people_batch(ids: &[i64]) -> SkadeBatch {
            let names: Vec<String> = ids.iter().map(|i| format!("p{i}")).collect();
            SkadeBatch::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 = people_batch(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,
                ..Default::default()
            };

            // 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 = SkadeBatch::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.