knut-bifrost 0.3.0

//! The **pure-Rust** CDC backend: no Spark, no JVM.
//!
//! `skade::Table::read_changelog` walks the Iceberg manifest lineage
//! (`read_delta`), tags appended rows `INSERT` and equality-deleted identities
//! `DELETE`, and falls back to a whole-snapshot re-read on an overwrite /
//! position-delete window. knut **consumes** that — it does not keep a second
//! incremental reader (L5).
//!
//! ## Where the projection lands
//!
//! Honest accounting, because "pushdown" means two different things here:
//!
//! * `ChangelogBatch::project_columns` is skade's own pruning API and it is a
//!   **post-decode** slice — the parquet files behind `read_delta` are still
//!   decoded whole (`ParquetRecordBatchReaderBuilder::try_new(bytes)?.build()?`
//!   takes no projection mask). So the projection here saves the *encode, the
//!   transform, the sink and the memory*, not the parquet I/O.
//! * The predicate is likewise applied to the change rows
//!   ([`super::post_filter`]), not to the scan plan. skade's `read_filtered` /
//!   `plan_stats` exist only for whole-snapshot reads.
//!
//! Pushing either into the manifest-delta scan itself is skade-side work and is
//! logged as such rather than faked here.

use std::path::PathBuf;

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

/// A [`CdcBackend`] over an embedded skade Iceberg warehouse.
///
/// ## What an update looks like here
///
/// Iceberg v2 has no "update" primitive: an update is an equality-delete plus an
/// append. skade's `ChangelogBatch::collapsed` re-tags such a pair as
/// `UPDATE_BEFORE` / `UPDATE_AFTER` **only on a primary-key table** (one whose
/// Iceberg schema declares `identifier_field_ids`); on a log table it cancels
/// the pair instead, which is the correct *net-change* answer and the wrong
/// answer for a live feed. So the collapse is driven by
/// [`ChangeRequest::net_changes`], not by a constructor flag — ask for net
/// changes and you get them, otherwise every raw event reaches the feed and
/// nothing is ever cancelled behind your back.
pub struct SkadeCdc {
    warehouse_dir: PathBuf,
}

impl SkadeCdc {
    /// Read changes from the warehouse rooted at `dir`.
    pub fn new(dir: impl Into<PathBuf>) -> Self {
        Self {
            warehouse_dir: dir.into(),
        }
    }
}

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

    async fn changes(&self, req: &ChangeRequest) -> Result<ChangeStream> {
        let src = |e: skade::SkadeError| BifrostError::Source(format!("{}: {e}", req.table));
        let wh = skade::Warehouse::open(&self.warehouse_dir)
            .await
            .map_err(src)?;
        let table = wh.table(&req.table).await.map_err(src)?;

        let to = match req.to_snapshot {
            Some(v) => v,
            None => match table.inner().metadata().current_snapshot() {
                Some(s) => s.snapshot_id(),
                // No snapshots yet: an empty, correctly-shaped feed.
                None => {
                    return Ok(ChangeStream::ready(
                        change_row_schema(&arrow_schema::Schema::empty()),
                        CdcCapability::skade(),
                        Vec::new(),
                    ));
                }
            },
        };

        let cl = table
            .read_changelog(req.from_snapshot, to)
            .await
            .map_err(src)?;
        let plan = cl.plan.clone();
        // Net-change collapse must run BEFORE the projection: it joins on the
        // delete files' identity columns, which the projection may drop.
        let cl = if req.net_changes {
            cl.collapsed().map_err(src)?
        } else {
            cl
        };
        // THE PROJECTION — skade's own pruning API, consumed not twinned. This
        // is what `SyncMode::Cdc` never had: before it, the CDC path was
        // full-width by construction.
        let engine_cols = req.engine_columns();
        let refs: Vec<&str> = engine_cols.iter().map(String::as_str).collect();
        let cl = cl.project_columns(&refs).map_err(src)?;

        let mut rows = Vec::with_capacity(cl.rows.len());
        let mut schema = None;
        for b in &cl.rows {
            let b = match &req.filter {
                Some(f) => post_filter(b, f)?,
                None => b.clone(),
            };
            let enc = encode_change_rows(&b, Some(to))?;
            let enc = fold_updates(&enc, &req.identity)?;
            schema.get_or_insert_with(|| enc.schema());
            rows.push(enc);
        }
        let schema = schema.unwrap_or_else(|| 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 = CdcKind::SkadeDelta.as_str(),
            from = ?req.from_snapshot,
            to,
            snapshots = plan.snapshots.len(),
            added_files = plan.added_files.len(),
            delete_files = plan.delete_files.len(),
            full_reload = plan.needs_full_reload,
            projected = req.columns.len(),
            // Said plainly: knut filtered the rows, the scan planner did not.
            post_filtered = req.filter.is_some(),
            change_rows = total,
            batches = rows.len(),
            "read change feed",
        );
        Ok(ChangeStream::ready(schema, CdcCapability::skade(), rows))
    }
}