knut-bifrost 0.3.0

//! **The CDC seam** — one change-data-capture vocabulary, three backends, and
//! the Arrow row shape a UI renders change-by-change.
//!
//! # Why a seam at all
//!
//! knut can learn "what changed in this table since snapshot N" three different
//! ways, and which one is *available* is a property of the server it is talking
//! to, not of anything an operator types into a YAML file:
//!
//! | backend | how | needs |
//! |---|---|---|
//! | [`CdcKind::SparkNative`] | Spark 4.2's `… CHANGES FROM VERSION a TO VERSION b` | a Spark **4.2+** Connect server **and** a DSv2 connector implementing `Changelog` |
//! | [`CdcKind::IcebergChangelogView`] | Iceberg's `system.create_changelog_view` procedure | any Spark with the Iceberg runtime loaded (4.0 / 4.1 / …) |
//! | [`CdcKind::SkadeDelta`] | `skade::Table::read_changelog` — manifest-level delta, pure Rust | no Spark at all |
//!
//! ## What Spark 4.2 actually added (and what it does not give us *yet*)
//!
//! Spark 4.2.0 (GA 2026-07-14) landed first-class CDC as SPARK-55668: a SQL
//! `CHANGES` clause, a DSv2 connector interface
//! `org.apache.spark.sql.connector.catalog.Changelog` (`@since 4.2.0`), and
//! DataFrame / PySpark / **Spark Connect** bindings (SPARK-55949). The grammar,
//! from Spark's own `SqlBaseParser.g4` at tag `v4.2.0`:
//!
//! ```text
//! changesClause
//!   : CHANGES FROM (SYSTEM_VERSION | VERSION) startingVersion (INCLUSIVE | EXCLUSIVE)?
//!       (TO (SYSTEM_VERSION | VERSION) endingVersion (INCLUSIVE | EXCLUSIVE)?)?
//!   | CHANGES FROM (SYSTEM_TIME | TIMESTAMP) startingTimestamp (INCLUSIVE | EXCLUSIVE)?
//!       (TO (SYSTEM_TIME | TIMESTAMP) endingTimestamp (INCLUSIVE | EXCLUSIVE)?)?
//!   ;
//! ```
//!
//! Its changelog rows carry `_change_type` (`insert` / `delete` /
//! `update_preimage` / `update_postimage` — **lower case**), `_commit_version`
//! and `_commit_timestamp`.
//!
//! **But**: the `CHANGES` clause only resolves for a table whose connector
//! implements `Changelog`. Apache Iceberg does not — as of 2026-07-22 Iceberg
//! 1.11.0 publishes `iceberg-spark-runtime-{3.4,3.5,4.0,4.1}` and **no 4.2
//! runtime at all** (apache/iceberg#14984 "Spark: Add support for 4.2.0" is
//! still open; #16751 was closed unmerged), and no Iceberg source references
//! `org.apache.spark.sql.connector.catalog.Changelog`. So on a *4.2 server
//! holding Iceberg tables* the native path is present in the engine and
//! unreachable for our data. That is not a reason to pretend: it is exactly why
//! [`detect`] **probes** instead of trusting the version string, and why a
//! failed probe records [`CdcCapability::fell_back`] rather than silently
//! choosing the other path.
//!
//! ## The renderable row
//!
//! Every backend is normalised onto ONE Arrow shape — [`change_row_schema`]:
//!
//! ```text
//! op       : Utf8                          non-null   insert | delete | update_before | update_after | update
//! before   : Struct<data columns, nullable> nullable  the pre-image  (null for an insert)
//! after    : Struct<data columns, nullable> nullable  the post-image (null for a delete)
//! snapshot : Int64                          nullable  Iceberg snapshot id / Spark _commit_version
//! ts       : Timestamp(Microsecond, UTC)    nullable  commit timestamp
//! ```
//!
//! One Arrow row per change event, delivered batch-by-batch over a
//! [`ChangeStream`] so a UI pane can paint changes as they arrive instead of
//! blocking on a whole scan.

use std::sync::Arc;

use arrow_array::{
    Array, ArrayRef, Int64Array, RecordBatch, StringArray, StructArray, TimestampMicrosecondArray,
};
use arrow_schema::{DataType, Field, Fields, Schema, SchemaRef, TimeUnit};

use crate::{
    error::{BifrostError, Result},
    mapping::{RowFilter, SyncMode},
};

// ── vocabulary ──────────────────────────────────────────────────────────────

/// The kind of a single change event, normalised across backends.
///
/// The two engines spell these differently and knut refuses to guess: Iceberg's
/// changelog view emits SCREAMING `INSERT` / `DELETE` / `UPDATE_BEFORE` /
/// `UPDATE_AFTER`, Spark 4.2's `Changelog` contract emits lower-case `insert` /
/// `delete` / `update_preimage` / `update_postimage`. [`ChangeOp::parse`]
/// accepts **both** dialects; [`ChangeOp::as_str`] emits knut's canonical one.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, serde::Serialize, serde::Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum ChangeOp {
    /// A row was added.
    Insert,
    /// A row was removed.
    Delete,
    /// The pre-image of an in-place update (Spark: `update_preimage`).
    UpdateBefore,
    /// The post-image of an in-place update (Spark: `update_postimage`).
    UpdateAfter,
    /// A pre/post-image **pair** folded into one row by [`fold_updates`]; both
    /// `before` and `after` are populated. No engine emits this — knut does.
    Update,
}

impl ChangeOp {
    /// knut's canonical spelling (what lands in the `op` column).
    pub const fn as_str(self) -> &'static str {
        match self {
            ChangeOp::Insert => "insert",
            ChangeOp::Delete => "delete",
            ChangeOp::UpdateBefore => "update_before",
            ChangeOp::UpdateAfter => "update_after",
            ChangeOp::Update => "update",
        }
    }

    /// Parse either engine's spelling (case-insensitive), plus knut's own.
    ///
    /// Returns `None` for anything unrecognised — the caller must fail loudly
    /// rather than drop the row, because a silently dropped `delete` is the
    /// single worst failure a change feed can have.
    pub fn parse(s: &str) -> Option<ChangeOp> {
        match s.to_ascii_lowercase().as_str() {
            "insert" => Some(ChangeOp::Insert),
            "delete" => Some(ChangeOp::Delete),
            // Iceberg spelling            | Spark 4.2 spelling
            "update_before" | "update_preimage" => Some(ChangeOp::UpdateBefore),
            "update_after" | "update_postimage" => Some(ChangeOp::UpdateAfter),
            "update" => Some(ChangeOp::Update),
            _ => None,
        }
    }

    /// Does this event carry a pre-image (fills the `before` struct)?
    pub const fn has_before(self) -> bool {
        matches!(
            self,
            ChangeOp::Delete | ChangeOp::UpdateBefore | ChangeOp::Update
        )
    }

    /// Does this event carry a post-image (fills the `after` struct)?
    pub const fn has_after(self) -> bool {
        matches!(
            self,
            ChangeOp::Insert | ChangeOp::UpdateAfter | ChangeOp::Update
        )
    }
}

/// Column names of the renderable change-row schema. Public so the UI does not
/// have to string-match knut's internals.
pub mod col {
    /// The [`super::ChangeOp`] as a canonical lower-case string.
    pub const OP: &str = "op";
    /// The pre-image struct (null when the event has no pre-image).
    pub const BEFORE: &str = "before";
    /// The post-image struct (null when the event has no post-image).
    pub const AFTER: &str = "after";
    /// Iceberg snapshot id / Spark `_commit_version`.
    pub const SNAPSHOT: &str = "snapshot";
    /// Commit timestamp, microseconds UTC.
    pub const TS: &str = "ts";
}

/// Spark 4.2's `Changelog` metadata columns, and Iceberg's changelog-view ones.
/// Named here so the two SQL builders and the encoder agree on one spelling.
pub mod meta {
    /// Both engines agree on this one (values differ — see [`super::ChangeOp`]).
    pub const CHANGE_TYPE: &str = "_change_type";
    /// Spark 4.2 (`Changelog` contract): the commit version of the change.
    pub const COMMIT_VERSION: &str = "_commit_version";
    /// Spark 4.2 (`Changelog` contract): the commit timestamp of the change.
    pub const COMMIT_TIMESTAMP: &str = "_commit_timestamp";
    /// Iceberg `create_changelog_view`: the snapshot the change was committed in.
    pub const COMMIT_SNAPSHOT_ID: &str = "_commit_snapshot_id";
    /// Iceberg `create_changelog_view`: the ordering of changes.
    pub const CHANGE_ORDINAL: &str = "_change_ordinal";

    /// Every metadata column either engine may add — these are stripped out of
    /// the `before` / `after` structs so a UI renders only real table columns.
    pub const ALL: [&str; 5] = [
        CHANGE_TYPE,
        COMMIT_VERSION,
        COMMIT_TIMESTAMP,
        COMMIT_SNAPSHOT_ID,
        CHANGE_ORDINAL,
    ];

    /// Is `name` one of the changelog metadata columns?
    pub fn is_meta(name: &str) -> bool {
        ALL.contains(&name)
    }
}

// ── the renderable row shape ────────────────────────────────────────────────

/// The data-only fields of `wide` — every column that is **not** changelog
/// metadata, made nullable (a `delete` row populates only its identity columns,
/// an `insert`'s `before` is entirely absent).
fn data_fields(wide: &Schema) -> Fields {
    wide.fields()
        .iter()
        .filter(|f| !meta::is_meta(f.name()))
        .map(|f| Arc::new(f.as_ref().clone().with_nullable(true)))
        .collect()
}

/// Build the renderable change-row schema for a table whose data columns are
/// the non-metadata columns of `wide`.
///
/// See the module docs for the shape. The `before` / `after` structs share ONE
/// [`Fields`] instance, so the two columns are guaranteed field-identical — a UI
/// can render a diff by zipping them positionally.
pub fn change_row_schema(wide: &Schema) -> SchemaRef {
    let fields = data_fields(wide);
    Arc::new(Schema::new(vec![
        Field::new(col::OP, DataType::Utf8, false),
        Field::new(col::BEFORE, DataType::Struct(fields.clone()), true),
        Field::new(col::AFTER, DataType::Struct(fields), true),
        Field::new(col::SNAPSHOT, DataType::Int64, true),
        Field::new(
            col::TS,
            DataType::Timestamp(TimeUnit::Microsecond, Some("UTC".into())),
            true,
        ),
    ]))
}

/// The data (non-metadata) column names of a wide changelog batch, in order.
pub fn data_columns(wide: &Schema) -> Vec<String> {
    wide.fields()
        .iter()
        .filter(|f| !meta::is_meta(f.name()))
        .map(|f| f.name().clone())
        .collect()
}

/// Take `col` at the rows in `idx` (a gather), returning a new array.
fn take_rows(array: &ArrayRef, idx: &[u32]) -> Result<ArrayRef> {
    let indices = arrow_array::UInt32Array::from(idx.to_vec());
    arrow::compute::take(array.as_ref(), &indices, None)
        .map_err(|e| BifrostError::Source(format!("cdc take: {e}")))
}

/// Build a `Struct` array over `fields` whose row *i* is `wide` row `rows[i]`,
/// null wherever `valid[i]` is false.
fn image_struct(
    wide: &RecordBatch,
    fields: &Fields,
    rows: &[u32],
    valid: &[bool],
) -> Result<ArrayRef> {
    let mut cols: Vec<ArrayRef> = Vec::with_capacity(fields.len());
    for f in fields.iter() {
        let src = wide.column_by_name(f.name()).ok_or_else(|| {
            BifrostError::Source(format!("changelog batch is missing column `{}`", f.name()))
        })?;
        let gathered = take_rows(src, rows)?;
        let gathered = if gathered.data_type() == f.data_type() {
            gathered
        } else {
            arrow::compute::cast(gathered.as_ref(), f.data_type())
                .map_err(|e| BifrostError::Source(format!("cdc cast {}: {e}", f.name())))?
        };
        cols.push(gathered);
    }
    let nulls = arrow::buffer::NullBuffer::from(valid.to_vec());
    Ok(Arc::new(
        StructArray::try_new(fields.clone(), cols, Some(nulls))
            .map_err(|e| BifrostError::Source(format!("cdc struct: {e}")))?,
    ))
}

/// Read the change kind of every row of a wide changelog batch.
///
/// Fails on a missing / non-string / null `_change_type`, and on any spelling
/// neither engine defines — never drops the row.
fn ops_of(wide: &RecordBatch) -> Result<Vec<ChangeOp>> {
    let idx = wide.schema().index_of(meta::CHANGE_TYPE).map_err(|_| {
        BifrostError::Source(format!(
            "changelog batch has no `{}` column",
            meta::CHANGE_TYPE
        ))
    })?;
    let kinds = wide
        .column(idx)
        .as_any()
        .downcast_ref::<StringArray>()
        .ok_or_else(|| {
            BifrostError::Source(format!("`{}` must be a string column", meta::CHANGE_TYPE))
        })?;
    (0..wide.num_rows())
        .map(|r| {
            if kinds.is_null(r) {
                return Err(BifrostError::Source(format!(
                    "null `{}` at row {r}",
                    meta::CHANGE_TYPE
                )));
            }
            ChangeOp::parse(kinds.value(r)).ok_or_else(|| {
                BifrostError::Source(format!(
                    "unknown `{}` `{}` at row {r} — refusing to drop a change row",
                    meta::CHANGE_TYPE,
                    kinds.value(r)
                ))
            })
        })
        .collect()
}

/// Pull the per-row snapshot id: `_commit_snapshot_id` (Iceberg) or
/// `_commit_version` (Spark 4.2), falling back to `default_snapshot` for the
/// pure-Rust skade path, which carries the cursor out-of-band.
fn snapshots_of(wide: &RecordBatch, default_snapshot: Option<i64>) -> Result<ArrayRef> {
    for name in [meta::COMMIT_SNAPSHOT_ID, meta::COMMIT_VERSION] {
        let Some(c) = wide.column_by_name(name) else {
            continue;
        };
        // `_commit_version` is LongType *or* StringType per Spark's contract.
        return arrow::compute::cast(c.as_ref(), &DataType::Int64)
            .map_err(|e| BifrostError::Source(format!("cdc snapshot cast {name}: {e}")));
    }
    Ok(Arc::new(Int64Array::from(vec![
        default_snapshot;
        wide.num_rows()
    ])))
}

/// Pull the per-row commit timestamp: `_commit_timestamp` (Spark 4.2) cast to
/// microseconds UTC, else all-null (Iceberg's changelog view has no timestamp
/// column — the ordinal is its ordering, and knut carries the snapshot id).
fn timestamps_of(wide: &RecordBatch) -> Result<ArrayRef> {
    let want = DataType::Timestamp(TimeUnit::Microsecond, Some("UTC".into()));
    match wide.column_by_name(meta::COMMIT_TIMESTAMP) {
        Some(c) => arrow::compute::cast(c.as_ref(), &want)
            .map_err(|e| BifrostError::Source(format!("cdc ts cast: {e}"))),
        None => Ok(Arc::new(
            TimestampMicrosecondArray::from(vec![None::<i64>; wide.num_rows()])
                .with_timezone("UTC"),
        )),
    }
}

/// **The encoder.** Turn one *wide* changelog batch (data columns +
/// `_change_type`, optionally `_commit_version` / `_commit_timestamp` /
/// `_commit_snapshot_id`) into the renderable [`change_row_schema`] shape.
///
/// `default_snapshot` fills the `snapshot` column when the batch carries no
/// per-row snapshot metadata (the skade path, whose cursor is out-of-band).
///
/// Every input row becomes exactly one output row — the encoder never merges,
/// filters or reorders. Call [`fold_updates`] afterwards to pair pre/post-images.
pub fn encode_change_rows(
    wide: &RecordBatch,
    default_snapshot: Option<i64>,
) -> Result<RecordBatch> {
    let schema = change_row_schema(wide.schema_ref());
    let DataType::Struct(fields) = schema.field(1).data_type().clone() else {
        unreachable!("change_row_schema builds `before` as a Struct");
    };
    let ops = ops_of(wide)?;

    let rows: Vec<u32> = (0..wide.num_rows() as u32).collect();
    let before_valid: Vec<bool> = ops.iter().map(|o| o.has_before()).collect();
    let after_valid: Vec<bool> = ops.iter().map(|o| o.has_after()).collect();

    let op_col = StringArray::from(ops.iter().map(|o| o.as_str()).collect::<Vec<_>>());
    let before = image_struct(wide, &fields, &rows, &before_valid)?;
    let after = image_struct(wide, &fields, &rows, &after_valid)?;
    let snapshot = snapshots_of(wide, default_snapshot)?;
    let ts = timestamps_of(wide)?;

    RecordBatch::try_new(schema, vec![Arc::new(op_col), before, after, snapshot, ts])
        .map_err(|e| BifrostError::Source(format!("cdc encode: {e}")))
}

/// Pair an `update_before` with its `update_after` into a single `update` row
/// carrying **both** images — the shape a diff pane actually wants.
///
/// Pairing is keyed on `(snapshot, identity columns of the pre-image)`. Rows
/// that do not pair (an unmatched pre-image, an insert, a delete) pass through
/// untouched and IN ORDER; an empty `identity` disables folding entirely and
/// returns the batch unchanged. Nothing is ever dropped.
pub fn fold_updates(rows: &RecordBatch, identity: &[String]) -> Result<RecordBatch> {
    if identity.is_empty() || rows.num_rows() == 0 {
        return Ok(rows.clone());
    }
    let op = rows
        .column_by_name(col::OP)
        .and_then(|c| c.as_any().downcast_ref::<StringArray>())
        .ok_or_else(|| BifrostError::Source("change rows missing `op`".into()))?;
    let before = struct_col(rows, col::BEFORE)?;
    let after = struct_col(rows, col::AFTER)?;
    let snapshot = rows
        .column_by_name(col::SNAPSHOT)
        .and_then(|c| c.as_any().downcast_ref::<Int64Array>())
        .ok_or_else(|| BifrostError::Source("change rows missing `snapshot`".into()))?;

    // key → index of a pending `update_before` awaiting its `update_after`.
    let mut pending: std::collections::HashMap<String, usize> = std::collections::HashMap::new();
    // Output plan: (op, row supplying `before`, row supplying `after`).
    let mut plan: Vec<(ChangeOp, Option<usize>, Option<usize>)> = Vec::new();
    // Where in `plan` a pending pre-image sits, so the matching post-image can
    // upgrade it in place instead of appending a second row.
    let mut slot: std::collections::HashMap<String, usize> = std::collections::HashMap::new();

    for r in 0..rows.num_rows() {
        let kind = ChangeOp::parse(op.value(r))
            .ok_or_else(|| BifrostError::Source(format!("unknown op `{}`", op.value(r))))?;
        match kind {
            ChangeOp::UpdateBefore => {
                let key = image_key(before, identity, r, snapshot, r)?;
                slot.insert(key.clone(), plan.len());
                pending.insert(key, r);
                plan.push((ChangeOp::UpdateBefore, Some(r), None));
            }
            ChangeOp::UpdateAfter => {
                let key = image_key(after, identity, r, snapshot, r)?;
                match (pending.remove(&key), slot.remove(&key)) {
                    (Some(pre), Some(at)) => plan[at] = (ChangeOp::Update, Some(pre), Some(r)),
                    _ => plan.push((ChangeOp::UpdateAfter, None, Some(r))),
                }
            }
            other => plan.push((
                other,
                other.has_before().then_some(r),
                other.has_after().then_some(r),
            )),
        }
    }
    if plan.len() == rows.num_rows() {
        return Ok(rows.clone()); // nothing paired — keep the original allocation
    }

    // Gather. A `None` side becomes a null image; `take` needs *some* index, so
    // point it at row 0 and mask it off with the validity buffer.
    let ops: Vec<&str> = plan.iter().map(|(o, _, _)| o.as_str()).collect();
    let b_idx: Vec<u32> = plan.iter().map(|(_, b, _)| b.unwrap_or(0) as u32).collect();
    let b_valid: Vec<bool> = plan.iter().map(|(_, b, _)| b.is_some()).collect();
    let a_idx: Vec<u32> = plan.iter().map(|(_, _, a)| a.unwrap_or(0) as u32).collect();
    let a_valid: Vec<bool> = plan.iter().map(|(_, _, a)| a.is_some()).collect();
    // Snapshot/ts come from whichever side exists (they agree within a pair).
    let m_idx: Vec<u32> = plan
        .iter()
        .map(|(_, b, a)| b.or(*a).unwrap_or(0) as u32)
        .collect();

    let fields = struct_fields(rows, col::BEFORE)?;
    RecordBatch::try_new(
        rows.schema(),
        vec![
            Arc::new(StringArray::from(ops)),
            gather_struct(before, &fields, &b_idx, &b_valid)?,
            gather_struct(after, &fields, &a_idx, &a_valid)?,
            take_rows(rows.column(3), &m_idx)?,
            take_rows(rows.column(4), &m_idx)?,
        ],
    )
    .map_err(|e| BifrostError::Source(format!("cdc fold: {e}")))
}

fn struct_col<'a>(rows: &'a RecordBatch, name: &str) -> Result<&'a StructArray> {
    rows.column_by_name(name)
        .and_then(|c| c.as_any().downcast_ref::<StructArray>())
        .ok_or_else(|| BifrostError::Source(format!("change rows missing `{name}` struct")))
}

fn struct_fields(rows: &RecordBatch, name: &str) -> Result<Fields> {
    match rows
        .schema()
        .field_with_name(name)
        .map(|f| f.data_type().clone())
    {
        Ok(DataType::Struct(f)) => Ok(f),
        _ => Err(BifrostError::Source(format!("`{name}` is not a struct"))),
    }
}

/// `(snapshot, identity values)` of row `r` inside an image struct — the fold key.
fn image_key(
    image: &StructArray,
    identity: &[String],
    r: usize,
    snapshot: &Int64Array,
    snap_row: usize,
) -> Result<String> {
    let mut key = String::new();
    if snapshot.is_valid(snap_row) {
        key.push_str(&snapshot.value(snap_row).to_string());
    }
    for name in identity {
        key.push('\u{1f}');
        let Some(c) = image.column_by_name(name) else {
            continue;
        };
        key.push_str(
            &arrow::util::display::array_value_to_string(c, r)
                .map_err(|e| BifrostError::Source(format!("cdc key {name}: {e}")))?,
        );
    }
    Ok(key)
}

fn gather_struct(
    src: &StructArray,
    fields: &Fields,
    idx: &[u32],
    valid: &[bool],
) -> Result<ArrayRef> {
    let mut cols: Vec<ArrayRef> = Vec::with_capacity(fields.len());
    for f in fields.iter() {
        let c = src
            .column_by_name(f.name())
            .ok_or_else(|| BifrostError::Source(format!("image struct missing `{}`", f.name())))?;
        cols.push(take_rows(c, idx)?);
    }
    let nulls = arrow::buffer::NullBuffer::from(valid.to_vec());
    Ok(Arc::new(
        StructArray::try_new(fields.clone(), cols, Some(nulls))
            .map_err(|e| BifrostError::Source(format!("cdc gather: {e}")))?,
    ))
}

// ── the ask ─────────────────────────────────────────────────────────────────

/// What a caller asks a [`CdcBackend`] for.
///
/// `columns` and `filter` are the **pushdown** half: they narrow what the CDC
/// read produces, exactly as [`crate::source::TableSpec`] does for a snapshot
/// read. Before this existed, `SyncMode::Cdc` was full-width by construction —
/// `SELECT *` from the changelog view / an unprojected `read_changelog` — so the
/// projection that landed for `SyncMode::Full` did not reach the CDC path at all.
#[derive(Debug, Clone, Default)]
pub struct ChangeRequest {
    /// Fully qualified table name as the engine sees it.
    pub table: String,
    /// Exclusive lower bound: the last snapshot/version already applied.
    /// `None` = from the beginning of the table's history.
    pub from_snapshot: Option<i64>,
    /// Inclusive upper bound. `None` = the table's current snapshot/version.
    pub to_snapshot: Option<i64>,
    /// The projection. **Empty = every column** (the pre-pushdown behaviour).
    /// The changelog metadata columns are always retained regardless.
    pub columns: Vec<String>,
    /// A row predicate applied to the change rows. `None` = no filter.
    pub filter: Option<RowFilter>,
    /// Collapse intermediate states so the window reports the *net* effect per
    /// row identity (Iceberg `net_changes => true`; Spark 4.2 `netChanges`).
    pub net_changes: bool,
    /// Identity columns used to pair pre/post-images in [`fold_updates`]. Empty
    /// leaves `update_before` / `update_after` as separate rows.
    pub identity: Vec<String>,
}

impl ChangeRequest {
    /// A window over `table` from `from` (exclusive) to the current snapshot.
    pub fn new(table: impl Into<String>, from: Option<i64>) -> Self {
        Self {
            table: table.into(),
            from_snapshot: from,
            ..Default::default()
        }
    }

    /// 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 change read.
    pub fn with_filter(mut self, filter: Option<RowFilter>) -> Self {
        self.filter = filter;
        self
    }

    /// Set the upper bound (inclusive).
    pub fn to(mut self, to: Option<i64>) -> Self {
        self.to_snapshot = to;
        self
    }

    /// Pair pre/post-images on these identity columns.
    pub fn with_identity(mut self, identity: Vec<String>) -> Self {
        self.identity = identity;
        self
    }

    /// The projection as the engine must see it: the asked-for columns plus
    /// every changelog metadata column, which must survive so the encoder can
    /// read `op` / `snapshot` / `ts`. Empty stays empty (= all columns).
    pub fn engine_columns(&self) -> Vec<String> {
        if self.columns.is_empty() {
            return Vec::new();
        }
        let mut out = self.columns.clone();
        for id in &self.identity {
            if !out.iter().any(|c| c == id) {
                out.push(id.clone());
            }
        }
        if let Some(f) = &self.filter
            && !out.iter().any(|c| c == &f.column)
        {
            out.push(f.column.clone());
        }
        out
    }
}

// ── which backend, and why ──────────────────────────────────────────────────

/// Which CDC implementation a feed is actually running on.
#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum CdcKind {
    /// Spark 4.2's native `CHANGES FROM VERSION … TO VERSION …` (SPARK-55668).
    SparkNative,
    /// Iceberg's `system.create_changelog_view` procedure, over Spark Connect.
    IcebergChangelogView,
    /// `skade::Table::read_changelog` — pure Rust, no Spark.
    SkadeDelta,
}

impl CdcKind {
    pub const fn as_str(self) -> &'static str {
        match self {
            CdcKind::SparkNative => "spark_native",
            CdcKind::IcebergChangelogView => "iceberg_changelog_view",
            CdcKind::SkadeDelta => "skade_delta",
        }
    }
}

/// **What detection concluded** — the observable half of "pick the backend by
/// what the server actually is".
///
/// This is logged as a `cdc_backend` tracing event and serialised into the
/// listener's `state_json`, so an operator can read *which* path ran and, when
/// it is not the one they expected, *why* — instead of inferring it from
/// timings. A fallback is never silent: [`fell_back`](Self::fell_back) is true
/// and [`reason`](Self::reason) carries the engine's own error text.
#[derive(Debug, Clone, PartialEq, serde::Serialize, serde::Deserialize)]
pub struct CdcCapability {
    /// The backend that will run.
    pub chosen: CdcKind,
    /// The Spark Connect server's version string, verbatim (`None` off-Spark).
    pub server_version: Option<String>,
    /// Parsed `(major, minor)` of `server_version`.
    pub server_major_minor: Option<(u32, u32)>,
    /// Is the server new enough that the native CDC *engine* exists (>= 4.2)?
    pub native_expected: bool,
    /// Did the native path actually resolve for this table? Spark 4.2 only
    /// answers `CHANGES` for a connector implementing
    /// `org.apache.spark.sql.connector.catalog.Changelog`; Iceberg does not.
    pub native_usable: bool,
    /// True when [`native_expected`](Self::native_expected) held but
    /// [`native_usable`](Self::native_usable) did not — the honest degrade.
    pub fell_back: bool,
    /// Human-readable justification, including the probe's error when it failed.
    pub reason: String,
}

impl CdcCapability {
    /// The off-Spark, pure-Rust conclusion: skade reads the changelog directly.
    pub fn skade() -> Self {
        Self {
            chosen: CdcKind::SkadeDelta,
            server_version: None,
            server_major_minor: None,
            native_expected: false,
            native_usable: false,
            fell_back: false,
            reason: "no Spark Connect session — reading Iceberg changelog in-process via skade"
                .into(),
        }
    }

    /// The `state_json` / UI projection of this conclusion.
    pub fn to_json(&self) -> serde_json::Value {
        serde_json::json!({
            "cdc_backend": self.chosen.as_str(),
            "server_version": self.server_version,
            "server_major_minor": self.server_major_minor.map(|(a, b)| format!("{a}.{b}")),
            "native_expected": self.native_expected,
            "native_usable": self.native_usable,
            "fell_back": self.fell_back,
            "reason": self.reason,
        })
    }

    /// Emit the decision as a structured tracing event (target `bifrost::cdc`).
    pub fn log(&self) {
        if self.fell_back {
            tracing::warn!(
                target: "bifrost::cdc",
                event = "cdc_backend",
                backend = self.chosen.as_str(),
                server_version = ?self.server_version,
                native_expected = self.native_expected,
                native_usable = self.native_usable,
                fell_back = true,
                reason = %self.reason,
                "native Spark CDC was expected but is not usable here — FELL BACK",
            );
        } else {
            tracing::info!(
                target: "bifrost::cdc",
                event = "cdc_backend",
                backend = self.chosen.as_str(),
                server_version = ?self.server_version,
                native_expected = self.native_expected,
                native_usable = self.native_usable,
                fell_back = false,
                reason = %self.reason,
                "selected CDC backend",
            );
        }
    }
}

/// Parse `"4.2.0"` / `"4.1.2-SNAPSHOT"` → `(4, 2)` / `(4, 1)`.
pub fn parse_major_minor(v: &str) -> Option<(u32, u32)> {
    let mut it = v.split(['.', '-', '+']);
    let major = it.next()?.trim().parse().ok()?;
    let minor = it.next()?.trim().parse().ok()?;
    Some((major, minor))
}

/// Is this Spark new enough to have the native CDC engine (SPARK-55668, 4.2.0)?
pub fn has_native_cdc_engine(major_minor: (u32, u32)) -> bool {
    major_minor >= (4, 2)
}

// ── the seam ────────────────────────────────────────────────────────────────

/// **The one CDC seam.** Every backend answers the same ask with the same
/// renderable rows, so the listener/UI never branches on which engine ran.
#[async_trait::async_trait]
pub trait CdcBackend: Send + Sync {
    /// Which implementation this is.
    fn kind(&self) -> CdcKind;

    /// Open a change feed for `req`. The stream yields
    /// [`change_row_schema`]-shaped batches as they become available.
    async fn changes(&self, req: &ChangeRequest) -> Result<ChangeStream>;
}

/// An incremental change feed: renderable batches delivered as they are
/// produced, so a UI pane paints arriving changes instead of waiting for the
/// whole scan.
///
/// Backed by a bounded channel, so a slow consumer applies backpressure to the
/// producer rather than growing an unbounded buffer.
pub struct ChangeStream {
    /// The renderable schema every batch carries (see [`change_row_schema`]).
    pub schema: SchemaRef,
    /// Which backend produced this feed and why (see [`CdcCapability`]).
    pub capability: CdcCapability,
    rx: tokio::sync::mpsc::Receiver<Result<RecordBatch>>,
}

/// The producer half of a [`ChangeStream`].
pub struct ChangeSink(tokio::sync::mpsc::Sender<Result<RecordBatch>>);

impl ChangeSink {
    /// Publish one renderable batch. Returns `false` once the consumer is gone.
    pub async fn send(&self, batch: Result<RecordBatch>) -> bool {
        self.0.send(batch).await.is_ok()
    }
}

impl ChangeStream {
    /// Create a stream/sink pair with a `capacity`-batch buffer.
    pub fn channel(
        schema: SchemaRef,
        capability: CdcCapability,
        capacity: usize,
    ) -> (ChangeSink, ChangeStream) {
        let (tx, rx) = tokio::sync::mpsc::channel(capacity.max(1));
        (
            ChangeSink(tx),
            ChangeStream {
                schema,
                capability,
                rx,
            },
        )
    }

    /// An already-complete feed — the batches are handed over as-is. Used by
    /// backends whose engine materialises the whole result before returning
    /// (Spark Connect's `collect` does); the consumer still sees them arrive
    /// batch-by-batch.
    pub fn ready(
        schema: SchemaRef,
        capability: CdcCapability,
        batches: Vec<RecordBatch>,
    ) -> ChangeStream {
        let (tx, rx) = tokio::sync::mpsc::channel(batches.len().max(1));
        for b in batches {
            // Capacity was sized to the batch count, so this cannot block.
            let _ = tx.try_send(Ok(b));
        }
        drop(tx);
        ChangeStream {
            schema,
            capability,
            rx,
        }
    }

    /// The next renderable batch, or `None` once the feed is exhausted.
    pub async fn next(&mut self) -> Option<Result<RecordBatch>> {
        self.rx.recv().await
    }

    /// Drain the whole feed (tests, benches, and any caller that genuinely
    /// wants the full window before acting).
    pub async fn collect(mut self) -> Result<Vec<RecordBatch>> {
        let mut out = Vec::new();
        while let Some(b) = self.next().await {
            out.push(b?);
        }
        Ok(out)
    }
}

// ── predicate over change rows ──────────────────────────────────────────────

/// Apply a [`RowFilter`] to a *wide* changelog batch, keeping only rows whose
/// filter column matches.
///
/// This is a **post-scan** filter, and is named as such everywhere it is logged:
/// neither Iceberg's changelog view nor skade's manifest delta accepts a scan
/// predicate, so knut cannot claim plan-time file skipping here the way
/// `SyncMode::Full` legitimately can. What it does buy is real: the rows never
/// reach the encoder, the transforms or the graph sink.
///
/// A `delete` row whose non-identity columns are null (skade widens them) is
/// **kept** when the filter column is null, because dropping it would silently
/// lose a deletion — the one CDC failure that must never happen quietly.
pub fn post_filter(wide: &RecordBatch, filter: &RowFilter) -> Result<RecordBatch> {
    let out = crate::mapping::apply_row_filter_with(
        std::slice::from_ref(wide),
        filter,
        crate::mapping::FilterNulls::Keep,
    )?;
    Ok(out.into_iter().next().unwrap_or_else(|| wide.clone()))
}

/// Bridge: the legacy `_change_type` sentinel name [`SyncMode`] exposes is the
/// same column this module calls [`meta::CHANGE_TYPE`]. Asserted, not assumed —
/// the listener's `partition_change_rows` and the CDC encoder must agree or a
/// delete goes to the wrong pipeline.
const _: () = assert!(
    matches!(SyncMode::CHANGE_TYPE_COLUMN.as_bytes(), b"_change_type"),
    "SyncMode::CHANGE_TYPE_COLUMN must equal cdc::meta::CHANGE_TYPE"
);

/// The Spark-side backends (`CHANGES` clause and `create_changelog_view`) plus
/// the capability probe that chooses between them.
#[cfg(feature = "spark")]
pub mod spark;
#[cfg(feature = "spark")]
pub use spark::{SparkIcebergCdc, SparkNativeCdc, backend_for, detect};

/// The pure-Rust backend: skade reads the Iceberg changelog in-process.
#[cfg(feature = "skade")]
pub mod skade;
#[cfg(feature = "skade")]
pub use skade::SkadeCdc;

#[cfg(test)]
mod tests;