crates/knut-bifrost/src/cdc.rs
//! **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;