knut-bifrost 0.3.0

//! The two **Spark-side** CDC backends and the capability probe that chooses
//! between them.
//!
//! * [`SparkNativeCdc`] — Spark 4.2's native `CHANGES FROM VERSION … TO VERSION …`
//!   (SPARK-55668 / SPARK-55948; Spark Connect bindings SPARK-55949).
//! * [`SparkIcebergCdc`] — Iceberg's `system.create_changelog_view` procedure,
//!   which is what every Spark **before** 4.2 has (and what a 4.2 server still
//!   has to use for an Iceberg table, because Iceberg implements no
//!   `org.apache.spark.sql.connector.catalog.Changelog`).
//!
//! [`detect`] picks one by **asking the server**, then **probing the table** —
//! never by config an operator has to get right.

use std::sync::Arc;

use arrow_array::RecordBatch;
use spark_connect::SparkSession;

use super::{
    CdcBackend, CdcCapability, CdcKind, ChangeRequest, ChangeStream, change_row_schema,
    encode_change_rows, fold_updates, has_native_cdc_engine, meta, parse_major_minor, post_filter,
};
use crate::error::{BifrostError, Result};

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

/// Run `query` and collect its batches. Shared by both backends.
async fn sql(session: &SparkSession, query: &str) -> Result<Vec<RecordBatch>> {
    let plan = session
        .sql(query, vec![])
        .await
        .map_err(|e| BifrostError::Source(format!("{query}: {e}")))?;
    session
        .collect(plan)
        .await
        .map_err(|e| BifrostError::Source(format!("{query}: {e}")))
}

/// The `SELECT` list for a projected changelog read: the asked-for columns plus
/// every metadata column the engine emits, so the encoder can still read `op` /
/// `snapshot` / `ts`. An empty projection stays `*` (the pre-pushdown shape).
///
/// `available` names the metadata columns *this* engine produces; asking for one
/// the engine does not have is a hard SQL error, so the two backends pass their
/// own list.
fn select_list(req: &ChangeRequest, available_meta: &[&str]) -> String {
    let cols = req.engine_columns();
    if cols.is_empty() {
        return "*".into();
    }
    let mut parts: Vec<String> = cols.iter().map(|c| quote_ident(c)).collect();
    for m in available_meta {
        parts.push(quote_ident(m));
    }
    parts.join(", ")
}

/// Backtick-quote one identifier (Spark's quoting), escaping embedded backticks.
fn quote_ident(name: &str) -> String {
    format!("`{}`", name.replace('`', "``"))
}

/// The `WHERE` fragment for the request's filter, pushed into the SQL over the
/// changelog view/relation rather than applied in knut. Returns `""` when there
/// is no filter.
///
/// This is a genuine engine-side filter (Spark evaluates it), unlike
/// [`post_filter`], which is knut's own last-resort for the no-Spark path.
pub(crate) fn where_clause_for(req: &ChangeRequest) -> Result<String> {
    let Some(f) = &req.filter else {
        return Ok(String::new());
    };
    f.validate()?;
    let c = quote_ident(&f.column);
    if let Some(v) = &f.eq {
        return Ok(format!(" WHERE {c} = {}", sql_literal(v)));
    }
    if let Some(set) = &f.is_in {
        let vals: Vec<String> = set.iter().map(sql_literal).collect();
        return Ok(format!(" WHERE {c} IN ({})", vals.join(", ")));
    }
    let mut preds = Vec::new();
    if let Some(lo) = &f.ge {
        preds.push(format!("{c} >= {}", sql_literal(lo)));
    }
    if let Some(hi) = &f.le {
        preds.push(format!("{c} <= {}", sql_literal(hi)));
    }
    Ok(format!(" WHERE {}", preds.join(" AND ")))
}

/// Render a filter value as a Spark SQL literal (strings single-quoted with
/// doubled quotes, so a value can never break out of the literal).
fn sql_literal(v: &crate::mapping::FilterValue) -> String {
    use crate::mapping::FilterValue as F;
    match v {
        F::Str(s) => format!("'{}'", s.replace('\'', "''")),
        F::Int(i) => i.to_string(),
        F::Float(f) => f.to_string(),
        F::Bool(b) => b.to_string(),
    }
}

/// Encode + fold + post-filter a set of wide changelog batches into the
/// renderable feed, then hand them over as a [`ChangeStream`].
///
/// `engine_filtered` says the SQL already applied the predicate, so knut does
/// not re-apply it (and cannot wrongly claim to).
async fn finish(
    wide: Vec<RecordBatch>,
    req: &ChangeRequest,
    capability: CdcCapability,
    default_snapshot: Option<i64>,
    engine_filtered: bool,
) -> Result<ChangeStream> {
    let mut rows = Vec::with_capacity(wide.len());
    let mut schema = None;
    for b in &wide {
        let b = if engine_filtered {
            b.clone()
        } else {
            match &req.filter {
                Some(f) => post_filter(b, f)?,
                None => b.clone(),
            }
        };
        let enc = encode_change_rows(&b, default_snapshot)?;
        let enc = fold_updates(&enc, &req.identity)?;
        schema.get_or_insert_with(|| enc.schema());
        rows.push(enc);
    }
    // An empty window still needs a schema for the UI to bind its columns to.
    let schema = match schema {
        Some(s) => s,
        None => change_row_schema(&arrow_schema::Schema::empty()),
    };
    let total: usize = rows.iter().map(|b| b.num_rows()).sum();
    tracing::info!(
        target: "bifrost::cdc",
        event = "cdc_read",
        table = %req.table,
        backend = capability.chosen.as_str(),
        from = ?req.from_snapshot,
        to = ?req.to_snapshot,
        projected = req.columns.len(),
        filtered = req.filter.is_some(),
        engine_filtered,
        change_rows = total,
        batches = rows.len(),
        "read change feed",
    );
    Ok(ChangeStream::ready(schema, capability, rows))
}

// ── Spark 4.2 native ────────────────────────────────────────────────────────

/// **Spark 4.2's own CDC.** Reads `… CHANGES FROM VERSION a TO VERSION b`.
///
/// The clause is engine-level (SPARK-55948) — Spark resolves it against the
/// table's DSv2 connector, which must implement
/// `org.apache.spark.sql.connector.catalog.Changelog` and expose `_change_type`
/// / `_commit_version` / `_commit_timestamp`. Spark itself does the carry-over
/// removal, update detection and (with `netChanges`) the net-change collapse, so
/// this backend issues one statement and encodes the result.
///
/// Reachable over Spark Connect: the clause travels as ordinary SQL through
/// `SparkSession::sql`, independent of the DataFrame `changes()` bindings that
/// SPARK-55949 added.
pub struct SparkNativeCdc {
    session: Arc<SparkSession>,
    capability: CdcCapability,
}

impl SparkNativeCdc {
    /// Wrap a session whose [`detect`] verdict already chose the native path.
    pub fn new(session: Arc<SparkSession>, capability: CdcCapability) -> Self {
        Self {
            session,
            capability,
        }
    }

    /// The statement this backend issues — exposed so a test can assert the SQL
    /// without a live 4.2 server (which, per the module docs, cannot serve an
    /// Iceberg table today anyway).
    pub fn statement(req: &ChangeRequest) -> Result<String> {
        // Spark's grammar: `CHANGES FROM VERSION v [TO VERSION v]`. `from` is the
        // last *applied* snapshot, so the window must EXCLUDE it — the grammar's
        // own `EXCLUSIVE` marker says exactly that, no off-by-one arithmetic.
        let from = match req.from_snapshot {
            Some(v) => format!("CHANGES FROM VERSION {v} EXCLUSIVE"),
            // No cursor: from the beginning of the table's history.
            None => "CHANGES FROM VERSION 0".to_string(),
        };
        let to = match req.to_snapshot {
            Some(v) => format!(" TO VERSION {v} INCLUSIVE"),
            None => String::new(),
        };
        let select = select_list(
            req,
            &[
                meta::CHANGE_TYPE,
                meta::COMMIT_VERSION,
                meta::COMMIT_TIMESTAMP,
            ],
        );
        Ok(format!(
            "SELECT {select} FROM (SELECT * FROM {t} {from}{to}){w}",
            t = req.table,
            w = where_clause_for(req)?,
        ))
    }
}

#[async_trait::async_trait]
impl CdcBackend for SparkNativeCdc {
    fn kind(&self) -> CdcKind {
        CdcKind::SparkNative
    }

    async fn changes(&self, req: &ChangeRequest) -> Result<ChangeStream> {
        let stmt = Self::statement(req)?;
        let wide = sql(&self.session, &stmt).await?;
        finish(wide, req, self.capability.clone(), req.to_snapshot, true).await
    }
}

// ── Iceberg changelog view ──────────────────────────────────────────────────

/// **Iceberg's CDC**, the path every Spark before 4.2 has — and the path a 4.2
/// server must still use for an Iceberg table.
///
/// `CALL system.create_changelog_view(table => …, options => map('start-snapshot-id', …,
/// 'end-snapshot-id', …), net_changes => …, changelog_view => …)` materialises a
/// temp view whose rows carry `_change_type` (`INSERT` / `DELETE` /
/// `UPDATE_BEFORE` / `UPDATE_AFTER`), `_change_ordinal` and
/// `_commit_snapshot_id`. This backend then reads that view **projected and
/// filtered** rather than `SELECT *` — the pushdown the CDC path never had.
pub struct SparkIcebergCdc {
    session: Arc<SparkSession>,
    capability: CdcCapability,
}

impl SparkIcebergCdc {
    pub fn new(session: Arc<SparkSession>, capability: CdcCapability) -> Self {
        Self {
            session,
            capability,
        }
    }

    /// The `CALL system.create_changelog_view(...)` statement for `req`.
    pub fn call_statement(req: &ChangeRequest, view: &str) -> String {
        let mut opts: Vec<String> = Vec::new();
        if let Some(from) = req.from_snapshot {
            opts.push(format!("'start-snapshot-id','{from}'"));
        }
        if let Some(to) = req.to_snapshot {
            opts.push(format!("'end-snapshot-id','{to}'"));
        }
        let options = if opts.is_empty() {
            String::new()
        } else {
            format!("options => map({}), ", opts.join(","))
        };
        // `identifier_columns` lets the procedure emit real UPDATE_BEFORE /
        // UPDATE_AFTER pairs instead of delete+insert — the shape the renderable
        // row wants — so pass the identity when the caller named one.
        let ids = if req.identity.is_empty() {
            String::new()
        } else {
            format!(
                "identifier_columns => array({}), ",
                req.identity
                    .iter()
                    .map(|c| format!("'{}'", c.replace('\'', "''")))
                    .collect::<Vec<_>>()
                    .join(",")
            )
        };
        format!(
            "CALL system.create_changelog_view(table => '{t}', {options}{ids}\
             net_changes => {net}, changelog_view => '{view}')",
            t = req.table,
            net = req.net_changes,
        )
    }

    /// The projected/filtered `SELECT` over the materialised view — the point
    /// where `SyncMode::Cdc` stopped being full-width.
    pub fn select_statement(req: &ChangeRequest, view: &str) -> Result<String> {
        let select = select_list(
            req,
            &[
                meta::CHANGE_TYPE,
                meta::CHANGE_ORDINAL,
                meta::COMMIT_SNAPSHOT_ID,
            ],
        );
        Ok(format!(
            "SELECT {select} FROM {view}{w} ORDER BY {ord}",
            w = where_clause_for(req)?,
            ord = meta::CHANGE_ORDINAL,
        ))
    }
}

#[async_trait::async_trait]
impl CdcBackend for SparkIcebergCdc {
    fn kind(&self) -> CdcKind {
        CdcKind::IcebergChangelogView
    }

    async fn changes(&self, req: &ChangeRequest) -> Result<ChangeStream> {
        let view = format!("bifrost_changes_{}", sanitise_view(&req.table));
        sql(&self.session, &Self::call_statement(req, &view)).await?;
        let wide = sql(&self.session, &Self::select_statement(req, &view)?).await?;
        // Best-effort cleanup; a leftover temp view is replaced next round.
        let _ = sql(&self.session, &format!("DROP VIEW IF EXISTS {view}")).await;
        finish(wide, req, self.capability.clone(), req.to_snapshot, true).await
    }
}

// ── detection ───────────────────────────────────────────────────────────────

/// **Pick the backend by what the server actually is.**
///
/// Two steps, both observable:
///
/// 1. **Ask.** `SparkSession::version()` (the Connect `AnalyzePlan` SparkVersion
///    request). `>= 4.2` means the native CDC *engine* exists — SPARK-55668
///    landed in 4.2.0.
/// 2. **Probe.** Even on 4.2, `CHANGES` only resolves when the table's connector
///    implements `Changelog`. So run a zero-row `CHANGES` read against the real
///    table. If it throws, the native engine is present but unusable *here*
///    (which is the state of the world for Iceberg today) — record the engine's
///    own error and fall back.
///
/// The result is returned, logged and stored in `state_json`; a fallback is
/// always marked [`CdcCapability::fell_back`]. A version string we cannot parse
/// is treated as "no native CDC" — conservative, and said out loud.
pub async fn detect(session: &SparkSession, table: &str) -> CdcCapability {
    let version = session.version().await.ok();
    let mm = version.as_deref().and_then(parse_major_minor);

    let Some(mm) = mm else {
        return CdcCapability {
            chosen: CdcKind::IcebergChangelogView,
            server_version: version.clone(),
            server_major_minor: None,
            native_expected: false,
            native_usable: false,
            fell_back: false,
            reason: format!(
                "could not parse a Spark version from {version:?} — assuming no native CDC \
                 and using Iceberg's create_changelog_view"
            ),
        };
    };

    if !has_native_cdc_engine(mm) {
        return CdcCapability {
            chosen: CdcKind::IcebergChangelogView,
            server_version: version,
            server_major_minor: Some(mm),
            native_expected: false,
            native_usable: false,
            fell_back: false,
            reason: format!(
                "Spark {}.{} predates the native CDC engine (SPARK-55668 landed in 4.2.0) — \
                 using Iceberg's create_changelog_view",
                mm.0, mm.1
            ),
        };
    }

    // 4.2+: the engine is there. Does it resolve for THIS table?
    let probe = format!("SELECT * FROM {table} CHANGES FROM VERSION 0 LIMIT 0");
    match sql(session, &probe).await {
        Ok(_) => CdcCapability {
            chosen: CdcKind::SparkNative,
            server_version: version,
            server_major_minor: Some(mm),
            native_expected: true,
            native_usable: true,
            fell_back: false,
            reason: format!(
                "Spark {}.{} native CDC probe resolved for `{table}` — using the CHANGES clause",
                mm.0, mm.1
            ),
        },
        Err(e) => CdcCapability {
            chosen: CdcKind::IcebergChangelogView,
            server_version: version,
            server_major_minor: Some(mm),
            native_expected: true,
            native_usable: false,
            fell_back: true,
            reason: format!(
                "Spark {}.{} has the native CDC engine but the CHANGES clause did not resolve \
                 for `{table}` (the connector implements no \
                 org.apache.spark.sql.connector.catalog.Changelog) — FELL BACK to Iceberg's \
                 create_changelog_view. Engine said: {e}",
                mm.0, mm.1
            ),
        },
    }
}

/// Detect, log, and hand back the backend to use — the one call a caller needs.
///
/// Boxed because the two arms are different types by design; the seam is the
/// trait, not the concrete backend.
pub async fn backend_for(
    session: Arc<SparkSession>,
    table: &str,
) -> (Arc<dyn CdcBackend>, CdcCapability) {
    let cap = detect(&session, table).await;
    cap.log();
    let backend: Arc<dyn CdcBackend> = match cap.chosen {
        CdcKind::SparkNative => Arc::new(SparkNativeCdc::new(session, cap.clone())),
        // `SkadeDelta` cannot be chosen from a Spark session; the Iceberg view
        // is the correct Spark-side answer either way.
        CdcKind::IcebergChangelogView | CdcKind::SkadeDelta => {
            Arc::new(SparkIcebergCdc::new(session, cap.clone()))
        }
    };
    (backend, cap)
}

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

    fn req() -> ChangeRequest {
        ChangeRequest::new("lake.social.people", Some(1234))
            .to(Some(5678))
            .with_columns(vec!["id".into(), "name".into()])
    }

    /// RED-when-broken: the native statement must carry the projection AND the
    /// three `Changelog` metadata columns. A projection that drops
    /// `_change_type` would make every change row un-typable — the encoder
    /// fails loudly, but the SQL is where it must not happen in the first place.
    #[test]
    fn native_statement_projects_and_keeps_metadata() {
        let s = SparkNativeCdc::statement(&req()).unwrap();
        assert!(s.contains("CHANGES FROM VERSION 1234 EXCLUSIVE"), "{s}");
        assert!(s.contains("TO VERSION 5678 INCLUSIVE"), "{s}");
        assert!(s.contains("`id`") && s.contains("`name`"), "{s}");
        for m in [
            meta::CHANGE_TYPE,
            meta::COMMIT_VERSION,
            meta::COMMIT_TIMESTAMP,
        ] {
            assert!(s.contains(m), "native statement dropped `{m}`: {s}");
        }
        assert!(
            !s.contains("SELECT *,"),
            "projection must not be SELECT *: {s}"
        );
        crate::testmatrix::emit(
            crate::testmatrix::COMPONENT,
            "cdc::native_statement_projects_and_keeps_metadata",
            true,
            &format!("Spark 4.2 CHANGES statement projected + metadata kept: {s}"),
        );
    }

    /// An unprojected request stays `SELECT *` — the pre-pushdown shape, so the
    /// pushdown can only ever narrow a read it was asked to narrow.
    #[test]
    fn unprojected_native_statement_is_select_star() {
        let s = SparkNativeCdc::statement(&ChangeRequest::new("t", None)).unwrap();
        assert!(s.starts_with("SELECT * FROM"), "{s}");
        assert!(s.contains("CHANGES FROM VERSION 0"), "{s}");
    }

    /// The Iceberg path: the CALL carries the snapshot window and the identity
    /// columns, and the SELECT is projected + filtered + ordered — not `*`.
    #[test]
    fn iceberg_changelog_view_statements_push_down() {
        let r = req()
            .with_filter(Some(RowFilter::eq("country", "SE")))
            .with_identity(vec!["id".into()]);
        let call = SparkIcebergCdc::call_statement(&r, "v");
        assert!(call.contains("'start-snapshot-id','1234'"), "{call}");
        assert!(call.contains("'end-snapshot-id','5678'"), "{call}");
        assert!(call.contains("identifier_columns => array('id')"), "{call}");

        let sel = SparkIcebergCdc::select_statement(&r, "v").unwrap();
        assert!(!sel.contains("SELECT * "), "must not be full-width: {sel}");
        assert!(sel.contains("`id`") && sel.contains("`name`"), "{sel}");
        // The filter column rides along even though the mapping never named it,
        // otherwise the WHERE would reference a column the projection dropped.
        assert!(sel.contains("`country`"), "{sel}");
        assert!(sel.contains("WHERE `country` = 'SE'"), "{sel}");
        for m in [
            meta::CHANGE_TYPE,
            meta::CHANGE_ORDINAL,
            meta::COMMIT_SNAPSHOT_ID,
        ] {
            assert!(sel.contains(m), "iceberg select dropped `{m}`: {sel}");
        }
        crate::testmatrix::emit(
            crate::testmatrix::COMPONENT,
            "cdc::iceberg_changelog_view_statements_push_down",
            true,
            &format!("changelog view SELECT is projected+filtered: {sel}"),
        );
    }

    /// A string literal can never break out of its quotes.
    #[test]
    fn sql_literals_are_escaped() {
        let r = ChangeRequest::new("t", None)
            .with_columns(vec!["a".into()])
            .with_filter(Some(RowFilter::eq("c", "O'Brien")));
        let sel = SparkIcebergCdc::select_statement(&r, "v").unwrap();
        assert!(sel.contains("'O''Brien'"), "{sel}");
    }

    /// A formless filter is rejected rather than silently pushed as a no-op.
    #[test]
    fn formless_filter_is_rejected() {
        let r = ChangeRequest::new("t", None).with_filter(Some(RowFilter {
            column: "c".into(),
            eq: None,
            is_in: None,
            ge: None,
            le: None,
        }));
        assert!(SparkNativeCdc::statement(&r).is_err());
    }

    /// Version gating is exactly "4.2 or newer".
    #[test]
    fn native_engine_gate_is_4_2() {
        assert!(!has_native_cdc_engine(parse_major_minor("4.1.2").unwrap()));
        assert!(has_native_cdc_engine(parse_major_minor("4.2.0").unwrap()));
        assert!(has_native_cdc_engine(parse_major_minor("5.0.0").unwrap()));
        assert!(!has_native_cdc_engine(parse_major_minor("3.5.7").unwrap()));
    }

    /// Compile-time proof that both Spark backends satisfy the ONE seam, so a
    /// caller can hold either behind `Arc<dyn CdcBackend>`.
    #[allow(dead_code)]
    fn _both_are_backends(a: SparkNativeCdc, b: SparkIcebergCdc) {
        let _: Arc<dyn CdcBackend> = Arc::new(a);
        let _: Arc<dyn CdcBackend> = Arc::new(b);
    }
}