crates/knut-bifrost/src/source.rs
//! Table sources. `SparkConnectSource` (feature `spark`) polls Iceberg
//! tables through a Spark Connect 4.x endpoint.
use arrow_array::RecordBatch;
use async_trait::async_trait;
use crate::{
error::Result,
mapping::{RowFilter, SyncMode},
state::TableState,
};
/// What the listener asks a source each round.
///
/// `columns` + `filter` are the **pushdown** half of the ask: they say what the
/// round actually needs, so a source that can prune (the Iceberg one) reads only
/// that. A source that cannot prune ignores them and returns everything — the
/// batches are then a superset, which every downstream transform already
/// tolerates (they project what they name).
#[derive(Debug, Clone, Default)]
pub struct TableSpec {
pub table: String,
pub mode: SyncMode,
/// The projection: columns the round needs. **Empty = every column**, which
/// is the pre-pushdown behaviour. Derived from the mapping's pipeline by
/// [`TableMapping::projected_columns`](crate::mapping::TableMapping::projected_columns).
pub columns: Vec<String>,
/// An optional row predicate to push into the scan planner (partition / data
/// file / row-group pruning). `None` = no filter.
pub filter: Option<RowFilter>,
}
impl TableSpec {
/// The pre-pushdown ask: this table, this mode, every column, no filter.
pub fn new(table: impl Into<String>, mode: SyncMode) -> Self {
Self {
table: table.into(),
mode,
columns: Vec::new(),
filter: None,
}
}
/// Narrow the read to `columns` (empty leaves it unprojected).
pub fn with_columns(mut self, columns: Vec<String>) -> Self {
self.columns = columns;
self
}
/// Push `filter` into the scan planner.
pub fn with_filter(mut self, filter: Option<RowFilter>) -> Self {
self.filter = filter;
self
}
/// `columns` as the `&[&str]` the skade read API takes.
pub fn column_refs(&self) -> Vec<&str> {
self.columns.iter().map(String::as_str).collect()
}
}
/// A place rows come from.
#[async_trait]
pub trait TableSource: Send + Sync {
/// Return rows new since `state` and the state to persist once they are
/// sunk, or `None` when nothing changed.
async fn fetch_new(
&self,
spec: &TableSpec,
state: &TableState,
) -> Result<Option<(Vec<RecordBatch>, TableState)>>;
/// **Which CDC backend this source resolved to for `table`, and why.**
///
/// The listener stamps this onto the table's [`TableState`] after every
/// changelog round, so the selection — and any *fallback* — is readable out
/// of `state_json` rather than only out of the log. `None` means the source
/// has no CDC story at all (the file source); it never means "unknown".
///
/// Default: `None`, so an out-of-tree source keeps compiling.
fn cdc_capability(&self, table: &str) -> Option<crate::cdc::CdcCapability> {
let _ = table;
None
}
}
pub use arrow_batches::{ArrowBatches, ArrowSource, EmptyRead};
mod arrow_batches {
use super::*;
use crate::error::BifrostError;
/// **A caller-supplied producer of Arrow rows for one table.**
///
/// This is the seam that lets *anything that already speaks Arrow* feed bifröst:
/// a warehouse client, a Flight stream, an ADBC read, a query result held in
/// memory. Implement it, wrap it in an [`ArrowSource`], and the whole mapping →
/// transform → graph-sink pipeline runs unchanged.
///
/// `spec.columns` is the round's projection ask. Honour it if the producer can
/// prune; ignoring it is fine — the batches are then a superset, which every
/// downstream transform already tolerates (they project what they name).
///
/// [`crate::rel2graph::ArrowTables`] — the in-memory holder the OFFLINE engine
/// already reads from — implements this, so one bag of Arrow batches drives both
/// the offline `build_mutation` and the live [`crate::Listener`].
#[async_trait]
pub trait ArrowBatches: Send + Sync {
/// Every row of `spec.table`, as Arrow.
async fn batches(&self, spec: &TableSpec) -> Result<Vec<RecordBatch>>;
/// The tables this producer knows about, when it can say. Used ONLY to make
/// an empty read's error useful ("you asked for `orders`; I hold `Orders`").
/// `None` means "cannot enumerate", which is honest for a live catalog.
fn known_tables(&self) -> Option<Vec<String>> {
None
}
}
/// The in-memory Arrow holder is a live-listener producer too — ONE type, both
/// paths. `rel2graph::ArrowTables` was already the crate's answer to "some Arrow
/// I already have"; this is what lets it drive a round loop as well as a one-shot
/// `build_mutation`.
#[async_trait]
impl ArrowBatches for crate::rel2graph::ArrowTables {
async fn batches(&self, spec: &TableSpec) -> Result<Vec<RecordBatch>> {
Ok(crate::rel2graph::TableSource::batches(self, &spec.table).to_vec())
}
fn known_tables(&self) -> Option<Vec<String>> {
Some(self.table_names().map(str::to_string).collect())
}
}
/// **What an Arrow read of ZERO rows means.**
///
/// It is a policy, not a detail, because the two readings are opposite and both
/// look identical on the wire:
///
/// * "the table really is empty" — a legitimate `Ok`, nothing to build;
/// * "I asked for a full build of `orders` and the catalog handed me nothing" —
/// a mis-pointed warehouse, a wrong namespace, an unrefreshed snapshot. Shipping
/// that as a successful round writes an EMPTY GRAPH that looks exactly like a
/// built one, and nobody finds out until someone opens the viewer.
///
/// The default is [`EmptyRead::Refuse`], because the second reading is the common
/// one and the first is easy to declare.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub enum EmptyRead {
/// A zero-row read is a typed [`BifrostError::Source`] naming the table and
/// the producer. **The default.**
#[default]
Refuse,
/// A zero-row read is `Ok(None)` — the caller has said this table may
/// legitimately be empty.
Allow,
}
/// **A [`TableSource`] over Arrow batches a caller already has.**
///
/// `SparkConnectSource` needs a Spark Connect endpoint and `IcebergSource` needs
/// a local skade warehouse *directory*. Neither can be pointed at a warehouse
/// reached some other way — a remote skade over its Iceberg-REST front, an
/// Iceberg REST catalog, a Spark whose client lives in the calling process. All
/// of those already produce Arrow, and this is the source that takes it.
///
/// ```no_run
/// # async fn demo() -> knut_bifrost::Result<()> {
/// use knut_bifrost::source::ArrowSource;
/// # let batches = Vec::new();
/// // A one-shot build from batches already in hand.
/// let source = ArrowSource::from_batches("my-warehouse", "lake.db.people", batches);
/// # let _ = source;
/// # Ok(()) }
/// ```
///
/// ## What it is NOT
///
/// **It is a batch read, not change-data-capture.** CDC and delta-watching in
/// bifröst (`SparkIcebergCdc`, `SparkNativeCdc`, `watch_iceberg_delta`,
/// `IcebergSource`'s incremental path) all rest on a *snapshot lineage* — a
/// before-id and an after-id the engine can diff. A bag of Arrow batches has no
/// such lineage and no engine behind it to ask, so:
///
/// | [`SyncMode`] | what this source does |
/// |---|---|
/// | `Full` | re-reads the producer every round (idempotent MERGE in the sink makes that safe) |
/// | `Incremental` | ships once, then reports nothing — there is no delta to compute |
/// | `Cdc` | **refuses**, typed, naming the reason and the sources that can |
///
/// [`cdc_capability`](TableSource::cdc_capability) is `None`, exactly as the file
/// source's is. A warehouse reached this way feeds bifröst for a **full build**,
/// not for streaming.
pub struct ArrowSource {
/// A short name for the producer behind this source — it appears in every
/// error, so an operator reads *which* warehouse came back empty.
producer: String,
batches: Box<dyn ArrowBatches>,
empty: EmptyRead,
}
impl ArrowSource {
/// Wrap any [`ArrowBatches`] producer. `producer` names it for diagnostics
/// (`"skade-remote"`, `"iceberg-rest"`, `"spark41"`, …).
pub fn new(producer: impl Into<String>, batches: impl ArrowBatches + 'static) -> Self {
Self {
producer: producer.into(),
batches: Box::new(batches),
empty: EmptyRead::default(),
}
}
/// A one-shot source over batches already in hand, for exactly one table.
/// Sugar over [`crate::rel2graph::ArrowTables`] — the same holder the offline
/// engine reads from, not a second one.
pub fn from_batches(
producer: impl Into<String>,
table: impl Into<String>,
batches: Vec<RecordBatch>,
) -> Self {
Self::new(
producer,
crate::rel2graph::ArrowTables::new().with_table(table, batches),
)
}
/// Declare that a zero-row read is legitimate for this source. Say it on
/// purpose; the default refuses. See [`EmptyRead`].
#[must_use]
pub fn allowing_empty(mut self) -> Self {
self.empty = EmptyRead::Allow;
self
}
/// The producer name this source reports in its errors.
pub fn producer(&self) -> &str {
&self.producer
}
/// What a zero-row read means here.
pub fn empty_policy(&self) -> EmptyRead {
self.empty
}
}
#[async_trait]
impl TableSource for ArrowSource {
async fn fetch_new(
&self,
spec: &TableSpec,
state: &TableState,
) -> Result<Option<(Vec<RecordBatch>, TableState)>> {
// CDC is refused BEFORE the read: a source with no snapshot lineage must
// not answer a changelog ask with an insert-only batch that the listener
// would then treat as the whole truth.
if matches!(spec.mode, SyncMode::Cdc) {
return Err(BifrostError::Source(format!(
"table '{}': an Arrow batch source ({}) has no change feed — it is a \
BATCH read with no snapshot lineage to diff. Use SyncMode::Full for a \
complete rebuild, or a source with an engine behind it \
(SparkIcebergCdc, SparkNativeCdc, or IcebergSource over a local skade \
warehouse) for CDC.",
spec.table, self.producer
)));
}
// Incremental over a batch source ships exactly once: there is no delta.
if matches!(spec.mode, SyncMode::Incremental) && state.synced_at.is_some() {
return Ok(None);
}
let batches = self.batches.batches(spec).await?;
let rows: usize = batches.iter().map(RecordBatch::num_rows).sum();
if rows == 0 {
// **The gate.** An empty read is not a built graph.
return match self.empty {
EmptyRead::Refuse => {
let holds = match self.batches.known_tables() {
Some(t) if !t.is_empty() => format!(" It holds {t:?}."),
Some(_) => " It holds no tables at all.".to_string(),
None => String::new(),
};
Err(BifrostError::Source(format!(
"table '{}': the Arrow producer ({}) returned {} batch(es) and 0 \
rows.{holds} An empty read is refused here rather than sunk as an \
empty graph that looks like a successful build — check the \
catalog, the namespace and the snapshot. If this table really is \
empty, say so with ArrowSource::allowing_empty().",
spec.table,
self.producer,
batches.len()
)))
}
EmptyRead::Allow => Ok(None),
};
}
// The state is handed back unchanged: a batch producer has no snapshot
// id, no watermark and no file ledger to record. `SyncState::advance`
// stamps `synced_at` after the sink succeeds, and THAT is what the
// Incremental "ship once" check above reads — so the marker is written by
// the round that actually landed, not by the read that merely started.
Ok(Some((batches, state.clone())))
}
/// No CDC story at all — the same honest `None` the file source gives.
fn cdc_capability(&self, _table: &str) -> Option<crate::cdc::CdcCapability> {
None
}
}
}
pub use file::FileSource;
mod file {
use std::path::{Path, PathBuf};
use graphar::FileType;
use super::*;
use crate::error::BifrostError;
/// A [`TableSource`] over a watched directory tree of GraphAr/columnar
/// file drops — the local half of the Airflow AIP-82 story.
///
/// Each mapped table maps to a subdirectory `root/<table>/`; every file
/// in it with a recognised extension (`parquet`, `orc`, `csv`, `json`,
/// `arrow`) is a chunk. The set of already-ingested file names lives in
/// the table's [`TableState::done_files`] ledger, so a new file drop
/// triggers exactly one sync round and nothing is re-read.
///
/// ```no_run
/// use knut_bifrost::source::FileSource;
/// let source = FileSource::new("/data/drops");
/// ```
///
/// [`TableState::done_files`]: crate::state::TableState::done_files
pub struct FileSource {
root: PathBuf,
}
impl FileSource {
/// Watch `root`; table `lake.foo.bar` reads from `root/lake.foo.bar/`.
pub fn new(root: impl Into<PathBuf>) -> Self {
Self { root: root.into() }
}
fn table_dir(&self, table: &str) -> PathBuf {
self.root.join(table)
}
/// All readable chunk files in `dir`, sorted by name for determinism.
fn list_files(dir: &Path) -> Result<Vec<String>> {
let read = match std::fs::read_dir(dir) {
Ok(r) => r,
// A table whose drop directory does not exist yet is simply
// empty — not an error (it may appear on a later round).
Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(vec![]),
Err(e) => return Err(e.into()),
};
let mut names = Vec::new();
for entry in read {
let entry = entry?;
if !entry.file_type()?.is_file() {
continue;
}
let name = entry.file_name().to_string_lossy().into_owned();
if file_type_of(&name).is_some() {
names.push(name);
}
}
names.sort();
Ok(names)
}
}
/// Recognise a chunk file by extension; `None` means "ignore this file".
fn file_type_of(name: &str) -> Option<FileType> {
match name.rsplit_once('.').map(|(_, ext)| ext) {
Some("parquet") => Some(FileType::Parquet),
Some("orc") => Some(FileType::Orc),
Some("csv") => Some(FileType::Csv),
Some("json") => Some(FileType::Json),
Some("arrow") => Some(graph_ar_ipc::file_type()),
_ => None,
}
}
#[async_trait]
impl TableSource for FileSource {
async fn fetch_new(
&self,
spec: &TableSpec,
state: &TableState,
) -> Result<Option<(Vec<RecordBatch>, TableState)>> {
let dir = self.table_dir(&spec.table);
let present = Self::list_files(&dir)?;
// Full mode re-reads every file each round (idempotent MERGE);
// incremental ships only files absent from the ledger.
let to_read: Vec<&String> = match spec.mode {
SyncMode::Full => present.iter().collect(),
// File drops have no changelog view, so CDC degrades to the
// append-only file ledger (new files only) — same as
// Incremental. (Per-row CDC would need a `_change_type` column
// inside the dropped files, which the Iceberg source supplies.)
SyncMode::Incremental | SyncMode::Cdc => present
.iter()
.filter(|f| !state.done_files.contains(f))
.collect(),
};
if to_read.is_empty() {
tracing::debug!(
target: "bifrost::file",
table = %spec.table,
present = present.len(),
"no new files",
);
return Ok(None);
}
let mut batches = Vec::new();
let mut rows = 0usize;
for name in &to_read {
let ft = file_type_of(name).expect("filtered to known types");
let chunk = graphar::io::read_chunk(dir.join(name), &ft)
.map_err(|e| BifrostError::Source(format!("{name}: {e}")))?;
let n: usize = chunk.iter().map(|b| b.num_rows()).sum();
rows += n;
// One event per detected-and-read file — the "file detected"
// signal a test/bencher asserts on without a human.
tracing::info!(
target: "bifrost::file",
event = "file_detected",
table = %spec.table,
file = %name,
rows = n,
"detected new file",
);
batches.extend(chunk);
}
tracing::info!(
target: "bifrost::file",
event = "files_read",
table = %spec.table,
files = to_read.len(),
rows,
"read new files",
);
let new_state = TableState {
done_files: present, // everything now present is done
..state.clone()
};
Ok(Some((batches, new_state)))
}
}
}
#[cfg(feature = "spark")]
pub use spark::SparkConnectSource;
#[cfg(feature = "spark")]
mod spark {
use super::*;
use arrow_array::cast::AsArray;
use arrow_array::types::Int64Type;
use spark_connect::{SparkSession, SparkSessionBuilder};
use crate::error::BifrostError;
/// Make a table name safe to embed in a temp-view identifier: keep
/// alphanumerics, map everything else (`.`, `-`, …) to `_`.
fn sanitise_view(table: &str) -> String {
table
.chars()
.map(|c| if c.is_ascii_alphanumeric() { c } else { '_' })
.collect()
}
/// The projected + filtered `SELECT` over a materialised Iceberg changelog
/// view — the CDC path's half of the pushdown that landed for `Full`.
///
/// An empty projection **and** no filter reproduces the old `SELECT * FROM
/// view` byte for byte, so the narrowing can only ever apply where it was
/// asked for. When the spec does project, the changelog metadata columns
/// ride along regardless (`_change_type` is load-bearing for delete
/// routing; `_change_ordinal` / `_commit_snapshot_id` carry the order and
/// the snapshot a renderable feed needs), as does the filter's own column.
///
/// Built on the shared [`crate::cdc`] vocabulary rather than a second copy
/// of the SQL: this is the same statement `cdc::SparkIcebergCdc` issues.
fn changelog_select(spec: &TableSpec, view: &str) -> Result<String> {
crate::cdc::SparkIcebergCdc::select_statement(&change_request(spec), view)
}
/// The [`crate::cdc::ChangeRequest`] a [`TableSpec`] describes — one place
/// the two pushdown vocabularies meet, so the plain-SQL path and the CDC
/// seam can never drift apart on what "projected" means.
fn change_request(spec: &TableSpec) -> crate::cdc::ChangeRequest {
crate::cdc::ChangeRequest::new(spec.table.clone(), None)
.with_columns(spec.columns.clone())
.with_filter(spec.filter.clone())
}
/// `spec.columns` as a Spark select-list, or `*` when unprojected.
fn select_list(spec: &TableSpec) -> String {
if spec.columns.is_empty() {
return "*".into();
}
let mut cols = spec.columns.clone();
// The predicate is evaluated by Spark against this same relation, so its
// column must survive the projection.
if let Some(f) = &spec.filter {
if !cols.iter().any(|c| c == &f.column) {
cols.push(f.column.clone());
}
}
cols.iter()
.map(|c| format!("`{}`", c.replace('`', "``")))
.collect::<Vec<_>>()
.join(", ")
}
/// `spec.filter` as a ` WHERE …` fragment (empty when there is none).
fn where_clause(spec: &TableSpec) -> Result<String> {
crate::cdc::spark::where_clause_for(&change_request(spec))
}
/// Polls Iceberg tables via Spark Connect (`sc://host:15002`).
///
/// Snapshot detection queries the table's `.snapshots` metadata table;
/// incremental mode reads only the delta between two snapshots.
pub struct SparkConnectSource {
session: std::sync::Arc<SparkSession>,
/// Per-table CDC verdict, resolved ONCE on the first changelog round
/// (detection costs a version request plus a zero-row probe, so it must
/// not run every poll) and reused after that. Recorded into
/// [`TableState::cdc`] so `state_json` shows which backend ran.
cdc: std::sync::Mutex<std::collections::HashMap<String, crate::cdc::CdcCapability>>,
}
impl SparkConnectSource {
pub async fn connect(url: &str) -> Result<Self> {
let session = SparkSessionBuilder::new(url)
.build()
.await
.map_err(|e| BifrostError::Source(e.to_string()))?;
Ok(Self {
session: std::sync::Arc::new(session),
cdc: std::sync::Mutex::new(std::collections::HashMap::new()),
})
}
/// **The CDC verdict for `table`** — detected on first use, cached after.
///
/// [`crate::cdc::detect`] asks the server its version and then *probes*
/// the table with a zero-row `CHANGES` read, because a 4.2 engine does
/// not imply a 4.2-capable connector. The verdict is logged once (with
/// `fell_back` set when the native path was expected but unusable) and
/// then reused, so a poll loop pays for detection exactly once.
pub async fn cdc_backend(
&self,
table: &str,
) -> (
std::sync::Arc<dyn crate::cdc::CdcBackend>,
crate::cdc::CdcCapability,
) {
if let Some(cap) = self.cdc.lock().ok().and_then(|m| m.get(table).cloned()) {
let backend: std::sync::Arc<dyn crate::cdc::CdcBackend> = match cap.chosen {
crate::cdc::CdcKind::SparkNative => std::sync::Arc::new(
crate::cdc::SparkNativeCdc::new(self.session.clone(), cap.clone()),
),
_ => std::sync::Arc::new(crate::cdc::SparkIcebergCdc::new(
self.session.clone(),
cap.clone(),
)),
};
return (backend, cap);
}
let (backend, cap) = crate::cdc::backend_for(self.session.clone(), table).await;
if let Ok(mut m) = self.cdc.lock() {
m.insert(table.to_string(), cap.clone());
}
(backend, cap)
}
/// **The renderable change feed** a UI consumes: `(op, before, after,
/// snapshot, ts)` batches over a [`crate::cdc::ChangeStream`], served by
/// whichever backend detection chose. This is the UI-facing twin of the
/// listener's wide-batch [`Self::read_changelog`] path.
pub async fn changes(
&self,
req: &crate::cdc::ChangeRequest,
) -> Result<crate::cdc::ChangeStream> {
let (backend, _) = self.cdc_backend(&req.table).await;
backend.changes(req).await
}
/// Run arbitrary SQL on the Spark Connect session (DDL, admin,
/// test fixtures). Returns the result batches directly — the
/// `spark-connect-arrow59` fork shares the workspace arrow major, so
/// `collect()` already yields workspace `RecordBatch`es (no bridge).
pub async fn sql(&self, query: &str) -> Result<Vec<RecordBatch>> {
let plan = self
.session
.sql(query, vec![])
.await
.map_err(|e| BifrostError::Source(format!("{query}: {e}")))?;
self.session
.collect(plan)
.await
.map_err(|e| BifrostError::Source(format!("{query}: {e}")))
}
/// Read Iceberg's changelog between two snapshots via the
/// `system.create_changelog_view` procedure.
///
/// The procedure materialises a temp view whose rows carry a
/// `_change_type` column (`INSERT` / `UPDATE_AFTER` / `UPDATE_BEFORE` /
/// `DELETE`); we then `SELECT *` from it. `net-changes` collapses an
/// update into a single `INSERT`/`DELETE` pair and drops rows that an
/// in-window insert+delete cancels out — exactly the upsert/delete
/// split the graph wants.
///
/// `prev` is the *start* snapshot (already applied, so the changelog is
/// open-ended after it) and `latest` the end snapshot.
///
/// **Projected and filtered.** `spec.columns` / `spec.filter` are pushed
/// into the `SELECT` over the materialised view instead of the old
/// `SELECT *` — closing the gap where `SyncMode::Cdc` stayed full-width
/// while `SyncMode::Full` had gained pushdown. `_change_type` is always
/// retained: without it the listener cannot route a delete. An
/// unprojected, unfiltered spec still produces exactly `SELECT *`.
async fn read_changelog(
&self,
spec: &TableSpec,
prev: i64,
latest: i64,
) -> Result<Vec<RecordBatch>> {
let table = &spec.table;
// WHICH CDC. Detected from the live server (version + a real probe),
// not from config — and cached, so this costs one round trip ever.
let (_, cap) = self.cdc_backend(table).await;
let req = crate::cdc::ChangeRequest::new(table.clone(), Some(prev))
.to(Some(latest))
.with_columns(spec.columns.clone())
.with_filter(spec.filter.clone());
let batches = if cap.chosen == crate::cdc::CdcKind::SparkNative {
// Spark 4.2's own CDC: one statement, no temp view. The rows come
// back with Spark's lower-case `_change_type` spelling, which the
// listener's router understands because both dialects go through
// `cdc::ChangeOp::parse`.
self.sql(&crate::cdc::SparkNativeCdc::statement(&req)?)
.await?
} else {
// Iceberg's CDC: materialise the changelog view, then read it
// PROJECTED + FILTERED (it used to be `SELECT *`).
let view = format!("bifrost_changes_{}", sanitise_view(table));
self.sql(&format!(
"CALL system.create_changelog_view(\
table => '{table}', \
options => map('start-snapshot-id','{prev}','end-snapshot-id','{latest}'), \
net_changes => true, \
changelog_view => '{view}')"
))
.await?;
let out = self.sql(&changelog_select(spec, &view)?).await?;
// Best-effort cleanup; a leftover temp view is harmless and gets
// replaced next round, so a drop failure must not fail the round.
let _ = self.sql(&format!("DROP VIEW IF EXISTS {view}")).await;
out
};
let rows: usize = batches.iter().map(|b| b.num_rows()).sum();
tracing::info!(
target: "bifrost::spark",
event = "changelog_read",
table = %table,
backend = cap.chosen.as_str(),
fell_back = cap.fell_back,
from = prev,
to = latest,
projected = spec.columns.len(),
filtered = spec.filter.is_some(),
change_rows = rows,
"read changelog (pushdown)",
);
Ok(batches)
}
async fn latest_snapshot(&self, table: &str) -> Result<Option<i64>> {
// Spark caches Iceberg table metadata per session; without a
// refresh, commits from other writers stay invisible to a
// long-lived polling session.
self.sql(&format!("REFRESH TABLE {table}")).await?;
let batches = self
.sql(&format!(
"SELECT snapshot_id FROM {table}.snapshots \
ORDER BY committed_at DESC LIMIT 1"
))
.await?;
for batch in &batches {
if batch.num_rows() > 0 {
let col = batch.column(0).as_primitive::<Int64Type>();
return Ok(Some(col.value(0)));
}
}
Ok(None)
}
}
#[async_trait]
impl TableSource for SparkConnectSource {
/// The cached verdict for `table` (`None` until the first changelog
/// round has run detection — a `Full`-only table never probes).
fn cdc_capability(&self, table: &str) -> Option<crate::cdc::CdcCapability> {
self.cdc.lock().ok().and_then(|m| m.get(table).cloned())
}
async fn fetch_new(
&self,
spec: &TableSpec,
state: &TableState,
) -> Result<Option<(Vec<RecordBatch>, TableState)>> {
let latest = self.latest_snapshot(&spec.table).await?;
let Some(latest) = latest else {
return Ok(None); // table has no snapshots yet
};
if state.snapshot_id == Some(latest) {
return Ok(None); // nothing new
}
// CDC reads the changelog *view* (a separate SQL call that creates
// a temp view), so it can't share the single-`SELECT` path below.
// (First CDC round has no prior snapshot to diff against — it falls
// through to the whole-table seed load tagged INSERT, below.)
if let (SyncMode::Cdc, Some(prev)) = (spec.mode, state.snapshot_id) {
let batches = self.read_changelog(spec, prev, latest).await?;
let new_state = TableState {
snapshot_id: Some(latest),
..state.clone()
};
return Ok(Some((batches, new_state)));
}
// The projection, as a SQL select-list. Empty projection → `*`, so an
// unprojected mapping issues exactly the query it always did. The
// `EXCEPT ALL` incremental diff **must** use the same list on both
// sides or the set difference compares different shapes — building it
// once here is what guarantees that.
let cols = select_list(spec);
let query = match (spec.mode, state.snapshot_id) {
// First sync, full mode, or first CDC round: whole table at the
// new snapshot. For CDC, tag every row INSERT so the listener's
// partitioning treats the seed load as upserts.
(SyncMode::Cdc, None) => format!(
"SELECT {cols}, 'INSERT' AS {ct} FROM {t} VERSION AS OF {latest}{w}",
ct = SyncMode::CHANGE_TYPE_COLUMN,
t = spec.table,
w = where_clause(spec)?,
),
(SyncMode::Full, _) | (SyncMode::Incremental, None) => format!(
"SELECT {cols} FROM {t} VERSION AS OF {latest}{w}",
t = spec.table,
w = where_clause(spec)?,
),
// Appended rows only. EXCEPT ALL is portable SQL; CDC mode above
// sees deletes/updates via Iceberg's changelog view instead.
(SyncMode::Incremental, Some(prev)) => format!(
"SELECT {cols} FROM {t} VERSION AS OF {latest}{w} \
EXCEPT ALL \
SELECT {cols} FROM {t} VERSION AS OF {prev}{w}",
t = spec.table,
w = where_clause(spec)?,
),
// Unreachable: CDC with a prior snapshot returned above.
(SyncMode::Cdc, Some(_)) => unreachable!("CDC delta handled above"),
};
let batches = self.sql(&query).await?;
let new_state = TableState {
snapshot_id: Some(latest),
..state.clone()
};
Ok(Some((batches, new_state)))
}
}
#[cfg(test)]
mod tests {
use super::*;
/// RED-WHEN-BROKEN version guard (container-free): the vendored
/// `spark-connect` (the `spark-connect-arrow59` fork) must yield knut's
/// **workspace** `arrow_array::RecordBatch` from `SparkSession::collect`,
/// so `SparkConnectSource` hands Spark batches straight to the sink with
/// no IPC / arrow-version bridge. If the fork ever regresses to an older
/// arrow major, `collect`'s payload becomes a *foreign* `RecordBatch`
/// type and this guard fails to compile — the whole no-bridge spark path
/// breaks with it. Never called; it only has to type-check.
#[allow(dead_code)]
fn _spark_collect_is_workspace_arrow59() -> impl std::future::Future<Output = ()> {
async {
// A generic never-returning helper rather than `unreachable!()` in
// each initializer: `unreachable!()` diverges, so everything after
// the first binding was unreachable code and `session` counted as
// unused. The guard type-checked anyway (unreachable code still
// is), but it warned on every build, and a type guard nobody can
// build cleanly is one people start ignoring. The bindings below
// are what actually pins the types; this only produces them.
fn nowhere<T>() -> T {
unreachable!("compile-only type guard — never called")
}
let session: SparkSession = nowhere();
let plan: spark_connect::spark::Plan = nowhere();
// The binding pins the element type to the workspace's arrow-59
// `RecordBatch`; only compiles while spark-connect shares it.
let _batches: Vec<arrow_array::RecordBatch> = session.collect(plan).await.unwrap();
}
}
/// Runtime mirror of the compile guard: the `RecordBatch` type
/// `spark-connect` names (`arrow::record_batch::RecordBatch`) is the same
/// type as knut's `arrow_array::RecordBatch` at the shared arrow-59 major.
#[test]
fn spark_connect_shares_workspace_arrow59() {
use std::any::TypeId;
let ok = TypeId::of::<arrow::record_batch::RecordBatch>()
== TypeId::of::<arrow_array::RecordBatch>();
assert!(
ok,
"spark-connect's arrow RecordBatch must equal knut's workspace \
arrow_array::RecordBatch (both arrow 59) — no bridge on the spark path",
);
crate::testmatrix::emit(
crate::testmatrix::COMPONENT,
"source::spark_connect_shares_workspace_arrow59",
ok,
"spark-connect (arrow59 fork) yields the workspace arrow-59 RecordBatch, no bridge",
);
}
}
}
#[cfg(feature = "skade")]
pub use iceberg::IcebergSource;
/// The delta plan an [`IcebergSource`] returns IS skade's: bifrost re-exports the
/// canonical [`skade::DeltaPlan`] rather than forking its own copy, so the shared
/// CDC changelog vocabulary never diverges.
#[cfg(feature = "skade")]
pub use skade::DeltaPlan;
/// Iceberg source via `skade` (feature `skade`) — read an Iceberg table
/// directly from an embedded warehouse, no Spark.
#[cfg(feature = "skade")]
mod iceberg {
use std::path::PathBuf;
use super::*;
use crate::delivery::IcebergLake;
use crate::error::BifrostError;
use skade::iceberg::table::Table as IceTable;
// skade's canonical CDC changelog surface — the vocabulary bifrost used to
// keep a private copy of. `Table::read_changelog` builds the `ChangelogBatch`
// (INSERT/DELETE tagging, delete-identity widening, full-reload fallback);
// `change_type::COLUMN` is the `_change_type` sentinel; `DeltaPlan` is the
// engine-side snapshot diff bifrost re-exports.
use skade::{ChangelogBatch, DeltaPlan, change_type};
/// A [`TableSource`] over an Iceberg table read through `skade`'s embedded
/// catalog. Each poll compares the table's current snapshot id to the
/// recorded one; on change it ships rows to the listener. The behaviour
/// depends on the [`SyncMode`] the listener asks for:
///
/// * [`SyncMode::Full`] — re-read the **whole** table at the new snapshot
/// (idempotent MERGE in the sink makes re-reads safe). Robust to any
/// commit kind (overwrite, delete, compaction) at the cost of re-scanning
/// every row each change.
/// * [`SyncMode::Incremental`] / [`SyncMode::Cdc`] — read **only the files
/// added since the recorded snapshot** (a manifest-level delta over the
/// snapshot lineage; see [`Self::ingest_delta`]). This is the
/// streaming/delta path: a continuous poll injects just the changes. It is
/// **CDC-aware**: appended data files become inserts, and a `Delete`
/// snapshot's **equality-delete** files become `_change_type = DELETE`
/// rows that remove the matching graph nodes (pure Rust, no Spark — skade
/// writes the deletes and surfaces them, knut tags + routes them through
/// the listener's changelog split). A position-delete or overwrite/replace
/// commit falls back to a full re-read so the graph never goes stale.
///
/// For append-only tables this replaces `SparkConnectSource` — no Spark
/// cluster needed, no JVM, pure Rust.
///
/// skade and knut both build on Apache Arrow 58 (one workspace major), so
/// `skade::arrow_array::RecordBatch` *is* the workspace `RecordBatch`: the
/// read batches hand straight through, no version bridge (see
/// [`from_skade_batches`], now an identity clone).
///
/// **Where the tables are** is an [`IcebergLake`]: an embedded warehouse
/// ([`Self::new`]), or — [`Self::from_lake`] — Iceberg tables on S3/MinIO read
/// with no catalog, or an Iceberg REST catalog such as MinIO AIStor's. Every
/// read below works on the loaded `iceberg` table, so all three stream alike.
pub struct IcebergSource {
lake: IcebergLake,
}
impl IcebergSource {
/// Read tables from the embedded Iceberg warehouse rooted at `dir`
/// (`<dir>/catalog.redb` + `warehouse/`). A `TableSpec.table` of
/// `name` resolves to the Iceberg table `name` in the default namespace.
pub fn new(dir: impl Into<PathBuf>) -> Self {
Self::from_lake(IcebergLake::Warehouse(dir.into()))
}
/// Read tables from any [`IcebergLake`]; a `TableSpec.table` is resolved
/// by [`IcebergLake::locate`].
pub fn from_lake(lake: IcebergLake) -> Self {
Self { lake }
}
/// Load the table at its current metadata. Loaded again on every call,
/// so each poll sees the lake's latest commit; an embedded warehouse is
/// opened and released inside the call (redb is single-open).
async fn open_table(&self, table: &str) -> Result<IceTable> {
self.lake.locate(table).load().await
}
/// **The delta primitive.** Read the changes to `table` between the
/// snapshots strictly after `from_snapshot` up to (and including)
/// `to_snapshot` into Arrow-58 batches — only the files those snapshots
/// added, never the whole table.
///
/// One-shot, pure Rust, no Spark. The manifest-level snapshot diff is
/// built by [`skade::Table::read_delta`] (iceberg-rust 0.9.1 has no
/// first-class incremental scan): it walks the snapshot lineage and
/// classifies each added file — appended **data** files are inserts,
/// **equality-delete** files are CDC deletes.
///
/// `from_snapshot = None` means "from the beginning" → every data file
/// up to `to` (equivalent to a full read, but file-by-file).
///
/// Returns the read batches **and** the [`DeltaPlan`] so a caller can
/// assert on the plan (snapshots, added/delete files) without a human.
/// When the window has identity deletes the batches carry a
/// `_change_type` column (`INSERT`/`DELETE`) for the listener's changelog
/// routing; a pure-append window returns untagged batches. A position-
/// delete or overwrite/replace commit sets `needs_full_reload` and the
/// batches are a **full** re-read of `to`.
pub async fn ingest_delta(
&self,
table_name: &str,
from_snapshot: Option<i64>,
to_snapshot: i64,
) -> Result<(Vec<RecordBatch>, DeltaPlan)> {
// Opens the warehouse once (redb is single-open); the delta runs
// against this same handle so we never reopen it mid-call.
let table = self.open_table(table_name).await?;
let spec = TableSpec::new(table_name, SyncMode::Cdc);
Self::delta_from_table(&table, &spec, from_snapshot, to_snapshot).await
}
/// The delta over an already-loaded iceberg table — the shared core of
/// [`Self::ingest_delta`] (which opens its own handle) and the
/// [`TableSource::fetch_new`] Incremental path (which reuses the handle
/// it already opened; redb is single-open, so reopening would deadlock).
///
/// **Projected.** `spec.columns` narrows the changelog through skade's own
/// [`skade::ChangelogBatch::project_columns`] — the pushdown
/// `SyncMode::Incremental` / `SyncMode::Cdc` never had, because
/// `read_changelog` takes no projection and knut called it unprojected.
/// An empty projection reproduces the previous behaviour exactly.
///
/// **`_change_type` survives the projection**, which is what makes this
/// safe for `Cdc`: `ChangelogBatch::project_columns` re-adds the sentinel
/// itself, so the listener's `partition_change_rows` still sees which
/// rows are deletes. (`TableMapping::projected_columns` does not name it —
/// skade does.) `tests/cdc_roundtrip.rs::source_cdc_mode_is_projected_and_agrees_with_the_full_width_read`
/// asserts the `_change_type` column is byte-identical projected vs full.
///
/// Accounting, stated rather than implied: `project_columns` is a
/// **post-decode** slice (skade's `read_delta` decodes whole parquet
/// files). The win is in the encode, the transforms, the sink and the
/// memory — *not* in parquet I/O. Pushing the projection into the
/// manifest-delta scan is skade-side work. The predicate is applied by
/// the caller (`fetch_new`), which owns the null policy.
async fn delta_from_table(
table: &IceTable,
spec: &TableSpec,
from_snapshot: Option<i64>,
to_snapshot: i64,
) -> Result<(Vec<RecordBatch>, DeltaPlan)> {
let table_name = &spec.table;
// skade's changelog encoder is the single source of truth for the
// snapshot diff AND the CDC re-tagging: `read_changelog` walks the
// manifest lineage (`read_delta`), tags appended rows INSERT and
// equality-deleted identities DELETE (widening them to the table
// schema), and on an overwrite/position-delete window re-reads the
// whole `to` snapshot as an INSERT stream. knut no longer keeps a
// private copy of any of that.
let cl: ChangelogBatch = skade::read_changelog(table, from_snapshot, to_snapshot)
.await
.map_err(|e| BifrostError::Source(format!("{table_name}: {e}")))?;
let plan = cl.plan.clone();
// THE PROJECTION. skade keeps `_change_type` for us (its
// `project_columns` re-adds the sentinel), so the listener's delete
// routing survives the narrowing. The filter's column is carried in
// by `engine_columns` so the row prune below can still see it.
let req = crate::cdc::ChangeRequest::new(table_name.clone(), from_snapshot)
.with_columns(spec.columns.clone())
.with_filter(spec.filter.clone());
let engine_cols = req.engine_columns();
let cl = if engine_cols.is_empty() {
cl
} else {
let refs: Vec<&str> = engine_cols.iter().map(String::as_str).collect();
cl.project_columns(&refs)
.map_err(|e| BifrostError::Source(format!("{table_name}: {e}")))?
};
// The PREDICATE is deliberately NOT applied here: `fetch_new` owns
// the residual filter (one implementation, `mapping::apply_row_filter_with`)
// and it is the caller that knows which null policy this batch needs.
let cl_rows: Vec<RecordBatch> = cl.rows;
// bifrost's contract: a pure-append (or full-reload) window flows
// through UNTAGGED — the rows go straight to the upsert pipeline, so
// strip the `_change_type` column skade tags every row with.
//
// WHICH window carries deletes is read off the ROWS, not off
// `plan.delete_files`: those are only the equality-delete files, and
// a copy-on-write rewrite, a position delete or a deletion vector
// deletes rows with none — its DELETE rows then lost their tag, became
// upserts, and put the deleted row back into the graph.
//
// Both delta modes apply deletes: `incremental` has deleted on an
// equality delete since the skade source existed
// (tests/delta_iceberg_falkordb.rs), and now does on every kind of
// delete — on an Iceberg source it reads the same changelog as `cdc`.
let has_deletes = cl_rows.iter().any(has_non_insert);
if !has_deletes {
let out: Vec<RecordBatch> = cl_rows
.iter()
.map(strip_change_type)
.collect::<Result<_>>()?;
let rows: usize = out.iter().map(|b| b.num_rows()).sum();
tracing::info!(
target: "bifrost::iceberg",
event = "delta_read",
table = %table_name,
from = ?from_snapshot,
to = to_snapshot,
snapshots = plan.snapshots.len(),
files = plan.added_files.len(),
full_reload = plan.needs_full_reload,
projected = spec.columns.len(),
post_filtered = spec.filter.is_some(),
rows,
"read iceberg delta (appended/full rows, no CDC deletes)",
);
return Ok((out, plan));
}
// CDC delta: keep skade's INSERT/DELETE tagging verbatim — the
// listener's `partition_change_rows` splits upserts from deletes on
// the `_change_type` column (skade already widened DELETE rows to the
// table schema so the vertex transform projects them).
let change_rows: usize = cl_rows.iter().map(|b| b.num_rows()).sum();
tracing::info!(
target: "bifrost::iceberg",
event = "delta_read_cdc",
table = %table_name,
from = ?from_snapshot,
to = to_snapshot,
snapshots = plan.snapshots.len(),
added_files = plan.added_files.len(),
delete_files = plan.delete_files.len(),
projected = spec.columns.len(),
post_filtered = spec.filter.is_some(),
change_rows,
"read iceberg CDC delta (inserts + identity deletes, tagged _change_type)",
);
Ok((cl_rows, plan))
}
/// **The projected / filtered snapshot read** — the `SyncMode::Full`
/// path, and the point where knut stopped paying for a full-width scan.
///
/// Before this, `Full` called `Table::read()`: every column of every data
/// file, always. skade has carried a pruning read surface for a while
/// (`read_columns` pushes the projection into the row-group readers so the
/// dropped columns are never decompressed or allocated; `read_filtered`
/// pushes a predicate into the scan planner so non-matching data files are
/// never opened) and knut called NONE of it. This routes the ask:
///
/// | `spec.columns` | `spec.filter` | skade call |
/// |---|---|---|
/// | empty | `None` | `read()` — unchanged full scan |
/// | set | `None` | `read_columns(cols)` |
/// | any | `Some(f)` | `read_filtered(f, cols)` (empty cols = all) |
///
/// So a mapping that names no columns and no filter behaves EXACTLY as
/// before; the pruning only ever narrows a read that was provably safe to
/// narrow.
///
/// When a filter is present the plan is also costed with
/// [`skade::Table::plan_stats`] — the filtered plan against the
/// whole-table baseline — so the data-skipping win is *observable*
/// (files/rows skipped) rather than merely asserted. That is plan-time
/// metadata work; it opens no data file.
async fn read_snapshot(
table: &IceTable,
spec: &TableSpec,
snapshot: Option<i64>,
) -> Result<Vec<RecordBatch>> {
let cols = spec.column_refs();
let src = |e: skade::SkadeError| BifrostError::Source(format!("{}: {e}", spec.table));
let (skade_batches, planned) = match &spec.filter {
Some(f) => {
let sf = to_scan_filter(f)?;
// Plan-only: how much of the table survives the predicate,
// and how much the unfiltered plan would have read.
let after = skade::plan_stats(&table, Some(&sf)).await.map_err(src)?;
let before = skade::plan_stats(&table, None).await.map_err(src)?;
let batches = skade::read_filtered(&table, &sf, &cols).await.map_err(src)?;
(batches, Some((before, after)))
}
None if !cols.is_empty() => (skade::read_columns(&table, &cols).await.map_err(src)?, None),
None => (skade::read_all(&table).await.map_err(src)?, None),
};
let batches = from_skade_batches(&skade_batches)?;
let rows: usize = batches.iter().map(|b| b.num_rows()).sum();
let width = batches.first().map(|b| b.num_columns()).unwrap_or(0);
match planned {
Some((before, after)) => {
tracing::info!(
target: "bifrost::iceberg",
event = "iceberg_read",
table = %spec.table,
snapshot = ?snapshot,
rows,
columns = width,
projected = cols.len(),
files_planned = after.data_files,
files_baseline = before.data_files,
files_skipped = before.data_files.saturating_sub(after.data_files),
rows_planned = after.rows_planned,
rows_baseline = before.rows_planned,
"read iceberg snapshot (full, pushdown)",
);
crate::testmatrix::emit(
crate::testmatrix::COMPONENT,
"source::iceberg_pushdown_read",
true,
&format!(
"{} data file(s) skipped of {}, {} column(s) projected",
before.data_files.saturating_sub(after.data_files),
before.data_files,
cols.len()
),
);
}
None => tracing::info!(
target: "bifrost::iceberg",
event = "iceberg_read",
table = %spec.table,
snapshot = ?snapshot,
rows,
columns = width,
projected = cols.len(),
"read iceberg snapshot (full)",
),
}
Ok(batches)
}
/// **Plan-time data-skipping stats** for `table` under `filter` — how many
/// data files / rows a scan would open, WITHOUT reading any of them
/// (`filter = None` plans the whole table, the baseline to diff against).
///
/// Exposed so a caller — a test, a bench, a planner UI — can measure the
/// pruning win directly instead of inferring it from wall time.
pub async fn plan_stats(
&self,
table_name: &str,
filter: Option<&RowFilter>,
) -> Result<skade::ScanPlanStats> {
let table = self.open_table(table_name).await?;
let sf = filter.map(to_scan_filter).transpose()?;
skade::plan_stats(&table, sf.as_ref())
.await
.map_err(|e| BifrostError::Source(format!("{table_name}: {e}")))
}
/// The table's current snapshot id (the delta target), or `None` if it
/// has no snapshots yet. The cursor a `watch` loop polls against.
pub async fn latest_snapshot(&self, table_name: &str) -> Result<Option<i64>> {
let table = self.open_table(table_name).await?;
Ok(table
.metadata()
.current_snapshot()
.map(|s| s.snapshot_id()))
}
}
/// Lower bifrost's backend-neutral [`RowFilter`] onto skade's
/// [`skade::ScanFilter`]. The spec type is engine-free (bifrost's core has
/// no Iceberg dependency); this is the ONE place the two vocabularies meet.
fn to_scan_filter(f: &RowFilter) -> Result<skade::ScanFilter> {
f.validate()?;
if let Some(v) = &f.eq {
return Ok(skade::ScanFilter::eq(&f.column, to_scalar(v)));
}
if let Some(vs) = &f.is_in {
return Ok(skade::ScanFilter::is_in(
&f.column,
vs.iter().map(to_scalar).collect::<Vec<_>>(),
));
}
Ok(skade::ScanFilter::range(
&f.column,
f.ge.as_ref().map(to_scalar),
f.le.as_ref().map(to_scalar),
))
}
fn to_scalar(v: &crate::mapping::FilterValue) -> skade::Scalar {
use crate::mapping::FilterValue as F;
match v {
F::Str(s) => skade::Scalar::Str(s.clone()),
F::Int(i) => skade::Scalar::I64(*i),
F::Float(f) => skade::Scalar::F64(*f),
F::Bool(b) => skade::Scalar::Bool(*b),
}
}
/// Drop the `_change_type` column skade's changelog tags every row with, so a
/// pure-append / full-reload window returns batches shaped exactly like an
/// ordinary table read — they flow straight to the upsert pipeline, matching
/// bifrost's pre-cutover behaviour. A batch without the column (defensive)
/// passes through untouched.
/// Whether `batch` holds a row that is not an insert (a DELETE, or an update's
/// pre-image).
fn has_non_insert(batch: &RecordBatch) -> bool {
let Some(col) = batch.column_by_name(change_type::COLUMN) else { return false };
let Some(ops) = col.as_any().downcast_ref::<arrow_array::StringArray>() else { return false };
ops.iter().any(|op| !matches!(op, Some(change_type::INSERT) | Some(change_type::UPDATE_AFTER)))
}
fn strip_change_type(batch: &RecordBatch) -> Result<RecordBatch> {
let Ok(idx) = batch.schema().index_of(change_type::COLUMN) else {
return Ok(batch.clone());
};
let mut fields: Vec<arrow_schema::Field> = batch
.schema()
.fields()
.iter()
.map(|f| f.as_ref().clone())
.collect();
fields.remove(idx);
let mut columns: Vec<arrow_array::ArrayRef> = batch.columns().to_vec();
columns.remove(idx);
RecordBatch::try_new(
std::sync::Arc::new(arrow_schema::Schema::new(fields)),
columns,
)
.map_err(|e| BifrostError::Source(format!("strip _change_type: {e}")))
}
#[async_trait]
impl TableSource for IcebergSource {
/// Always skade: this source holds an embedded warehouse, so there is no
/// Spark to ask and nothing to fall back from. Stated explicitly rather
/// than left `None`, so `state_json` names the path on this route too.
fn cdc_capability(&self, _table: &str) -> Option<crate::cdc::CdcCapability> {
Some(crate::cdc::CdcCapability::skade())
}
async fn fetch_new(
&self,
spec: &TableSpec,
state: &TableState,
) -> Result<Option<(Vec<RecordBatch>, TableState)>> {
let table = self.open_table(&spec.table).await?;
// The current Iceberg snapshot id; unchanged → nothing to do.
let snapshot = table
.metadata()
.current_snapshot()
.map(|s| s.snapshot_id());
let Some(latest) = snapshot else {
return Ok(None); // table has no snapshots yet
};
if Some(latest) == state.snapshot_id {
tracing::debug!(
target: "bifrost::iceberg",
table = %spec.table,
snapshot = ?snapshot,
"no snapshot change",
);
return Ok(None);
}
let batches = match spec.mode {
// Full mode: re-read the table at the new snapshot — PROJECTED
// and FILTERED to what the round asked for (see `read_snapshot`).
SyncMode::Full => Self::read_snapshot(&table, spec, snapshot).await?,
// Incremental / CDC: the streaming delta path — read only the
// data files added since the recorded snapshot. First round
// (no cursor) reads from the beginning (whole table).
SyncMode::Incremental | SyncMode::Cdc => {
// Reuse the table handle already opened above — reopening the
// same redb warehouse mid-call would deadlock (single-open).
let (batches, _plan) =
Self::delta_from_table(&table, spec, state.snapshot_id, latest).await?;
// ── the predicate means the SAME thing in every mode ──────────
//
// The delta scan plans on the manifest (which files were ADDED
// since the cursor), not on the predicate, so it cannot prune on
// `spec.filter` the way `read_snapshot` does. Before this it
// simply DROPPED the filter — and nothing downstream re-applies
// it, so rows the operator excluded were ingested anyway. The
// same mapping selected 1000 rows under `Incremental` and 250
// under `Full` (tests/iceberg_pushdown.rs).
//
// Falling back to the in-memory residual keeps the MEANING
// identical and only changes the cost.
//
// UPDATED: the PROJECTION is no longer dropped here either —
// `delta_from_table` now pushes `spec.columns` through skade's
// `ChangelogBatch::project_columns`, which re-adds the
// `_change_type` sentinel itself, so Cdc's
// `partition_change_rows` still sees which rows are deletes.
// (The earlier note that projecting the delta path "would
// strip it" was true of a naive projection, not of skade's.)
//
// NULL POLICY. A changelog batch's `DELETE` rows are widened
// to the table schema with **nulls in every non-identity
// column**, so the scan-planner default (`FilterNulls::Drop`,
// NULL fails the predicate) would silently remove deletions —
// the one CDC failure that must never happen quietly. A batch
// carrying `_change_type` therefore keeps what the predicate
// cannot judge; an ordinary delta keeps the planner semantics
// so `Full` and `Incremental` select the same rows.
match &spec.filter {
Some(f) => {
let before: usize = batches.iter().map(|b| b.num_rows()).sum();
let nulls = if batches.iter().any(crate::transform::is_changelog_batch)
{
crate::mapping::FilterNulls::Keep
} else {
crate::mapping::FilterNulls::Drop
};
let kept = crate::mapping::apply_row_filter_with(&batches, f, nulls)?;
let after: usize = kept.iter().map(|b| b.num_rows()).sum();
tracing::info!(
target: "bifrost::iceberg",
event = "delta_residual_filter",
table = %spec.table,
mode = ?spec.mode,
column = %f.column,
rows_before = before,
rows_after = after,
rows_dropped = before.saturating_sub(after),
"applied the mapping's row filter to the delta batches \
(residual — the delta scan plans on the manifest, not the \
predicate)",
);
crate::testmatrix::emit(
crate::testmatrix::COMPONENT,
"source::delta_residual_filter",
true,
&format!(
"{} rows -> {} ({} dropped) by `{}` in {:?} mode",
before,
after,
before.saturating_sub(after),
f.column,
spec.mode
),
);
kept
}
None => batches,
}
}
};
let new_state = TableState {
snapshot_id: Some(latest),
..state.clone()
};
Ok(Some((batches, new_state)))
}
}
/// Bridge skade's batches to the workspace's — now an **identity clone**.
///
/// knut and skade both build on arrow 59, so `skade::arrow_array::RecordBatch`
/// *is* the workspace `RecordBatch` (same crate, same major). The old
/// C-Data-Interface FFI transmute bridge (arrow 57 → arrow 56), the mirror of
/// `graphar-flight`'s `to_skade_batch`, is gone: a plain `clone()` (Arc bump,
/// no buffer copy, no FFI) hands the batch straight through.
fn from_skade_batches(
skade_batches: &[skade::arrow_array::RecordBatch],
) -> Result<Vec<RecordBatch>> {
skade_batches.iter().map(from_skade_batch).collect()
}
/// Identity clone of one skade (arrow-59) batch to the workspace's arrow-59
/// `RecordBatch` — same type, no FFI.
fn from_skade_batch(skade_batch: &skade::arrow_array::RecordBatch) -> Result<RecordBatch> {
Ok(skade_batch.clone())
}
#[cfg(test)]
mod tests {
use super::*;
use std::sync::Arc;
#[tokio::test]
async fn reads_iceberg_table_and_tracks_snapshot() {
use skade::arrow_array::{Int64Array, RecordBatch as SkadeBatch, StringArray};
use skade::arrow_schema::{DataType, Field, Schema};
// Stage an Iceberg table via skade (arrow 59).
let dir = tempfile::tempdir().unwrap();
let batch = SkadeBatch::try_new(
Arc::new(Schema::new(vec![
Field::new("id", DataType::Int64, false),
Field::new("name", DataType::Utf8, false),
])),
vec![
Arc::new(Int64Array::from(vec![1, 2, 3])),
Arc::new(StringArray::from(vec!["a", "b", "c"])),
],
)
.unwrap();
{
// Drop the setup warehouse before reading — redb is single-open
// and IcebergSource opens the same dir itself.
let wh = skade::Warehouse::open(dir.path()).await.unwrap();
let mut table = wh.table_or_create("people", &batch.schema()).await.unwrap();
table.append(&[batch]).await.unwrap();
}
// Read it back through IcebergSource (arrow 59).
let src = IcebergSource::new(dir.path());
let spec = TableSpec {
table: "people".into(),
mode: SyncMode::Full,
..Default::default()
};
let (batches, st) = src
.fetch_new(&spec, &TableState::default())
.await
.unwrap()
.expect("a snapshot to read");
let rows = batches.iter().map(|b| b.num_rows()).sum::<usize>();
assert_eq!(rows, 3);
assert!(st.snapshot_id.is_some(), "tracked the snapshot id");
// Re-poll with the recorded snapshot → no change.
let idle_none = src.fetch_new(&spec, &st).await.unwrap().is_none();
assert!(idle_none, "unchanged snapshot → None");
crate::testmatrix::emit(
crate::testmatrix::COMPONENT,
"source::reads_iceberg_table_and_tracks_snapshot",
rows == 3 && st.snapshot_id.is_some() && idle_none,
&format!(
"read {} rows, tracked snapshot {:?}, idle→None",
rows, st.snapshot_id
),
);
}
use skade::arrow_array::{Int64Array, RecordBatch as SkadeBatch, StringArray};
use skade::arrow_schema::{DataType, Field, Schema};
fn people_batch(ids: &[i64]) -> SkadeBatch {
let names: Vec<String> = ids.iter().map(|i| format!("p{i}")).collect();
SkadeBatch::try_new(
Arc::new(Schema::new(vec![
Field::new("id", DataType::Int64, false),
Field::new("name", DataType::Utf8, false),
])),
vec![
Arc::new(Int64Array::from(ids.to_vec())),
Arc::new(StringArray::from(names)),
],
)
.unwrap()
}
/// Append `ids` as one new snapshot; return its snapshot id.
async fn append_snapshot(dir: &std::path::Path, ids: &[i64]) -> i64 {
let wh = skade::Warehouse::open(dir).await.unwrap();
let batch = people_batch(ids);
let mut table = wh.table_or_create("people", &batch.schema()).await.unwrap();
table.append(&[batch]).await.unwrap();
table
.inner()
.metadata()
.current_snapshot()
.unwrap()
.snapshot_id()
}
/// LAW inject-assert: a real local Iceberg warehouse, two appends, and
/// the delta reads ONLY the second append's rows — not the whole table.
#[tokio::test]
async fn delta_reads_only_appended_rows() {
let dir = tempfile::tempdir().unwrap();
// Snapshot A: 3 rows.
let snap_a = append_snapshot(dir.path(), &[1, 2, 3]).await;
// Snapshot B: 2 MORE rows (a separate fast_append commit).
let snap_b = append_snapshot(dir.path(), &[4, 5]).await;
assert_ne!(snap_a, snap_b, "two distinct snapshots");
let src = IcebergSource::new(dir.path());
// From the beginning to A → all 3 rows.
let (batches, plan) = src.ingest_delta("people", None, snap_a).await.unwrap();
assert_eq!(batches.iter().map(|b| b.num_rows()).sum::<usize>(), 3);
assert!(!plan.needs_full_reload);
assert_eq!(plan.snapshots, vec![snap_a]);
assert_eq!(plan.added_files.len(), 1, "A added exactly one data file");
// THE DELTA: from A to B → ONLY the 2 new rows, not 5.
let (delta, plan) = src
.ingest_delta("people", Some(snap_a), snap_b)
.await
.unwrap();
assert_eq!(
delta.iter().map(|b| b.num_rows()).sum::<usize>(),
2,
"delta is just B's 2 appended rows (not a full 5-row reload)",
);
assert!(!plan.needs_full_reload);
assert_eq!(plan.snapshots, vec![snap_b], "only snapshot B is new");
assert_eq!(plan.added_files.len(), 1, "B added exactly one data file");
// The delta's ids are exactly {4,5}.
let col = delta[0]
.column(0)
.as_any()
.downcast_ref::<arrow_array::Int64Array>()
.unwrap();
let mut got: Vec<i64> = (0..col.len()).map(|i| col.value(i)).collect();
got.sort();
assert_eq!(got, vec![4, 5]);
// Cursor already at B → empty delta, no files.
let (none, plan) = src
.ingest_delta("people", Some(snap_b), snap_b)
.await
.unwrap();
assert!(none.is_empty(), "caught-up cursor → no rows");
assert!(plan.snapshots.is_empty() && plan.added_files.is_empty());
crate::testmatrix::emit(
crate::testmatrix::COMPONENT,
"source::delta_reads_only_appended_rows",
got == vec![4, 5] && none.is_empty(),
&format!("A→B delta = ids {got:?} (2 rows, not full 5); caught-up→empty"),
);
}
/// The `TableSource::fetch_new` Incremental path advances the cursor and
/// ships only the new snapshot's rows on the second poll.
#[tokio::test]
async fn incremental_fetch_new_advances_cursor_and_ships_delta() {
let dir = tempfile::tempdir().unwrap();
append_snapshot(dir.path(), &[1, 2, 3]).await;
let src = IcebergSource::new(dir.path());
let spec = TableSpec {
table: "people".into(),
mode: SyncMode::Incremental,
..Default::default()
};
// First poll (no cursor) → whole table (3 rows), cursor set.
let (b1, st1) = src
.fetch_new(&spec, &TableState::default())
.await
.unwrap()
.expect("first snapshot");
assert_eq!(b1.iter().map(|b| b.num_rows()).sum::<usize>(), 3);
assert!(st1.snapshot_id.is_some());
// Idle: unchanged snapshot → None.
assert!(src.fetch_new(&spec, &st1).await.unwrap().is_none());
// Append B (2 rows), poll again → ONLY the 2 new rows.
append_snapshot(dir.path(), &[4, 5]).await;
let (b2, st2) = src
.fetch_new(&spec, &st1)
.await
.unwrap()
.expect("delta snapshot");
let b2_rows = b2.iter().map(|b| b.num_rows()).sum::<usize>();
assert_eq!(b2_rows, 2, "incremental ships only the delta");
assert_ne!(st2.snapshot_id, st1.snapshot_id, "cursor advanced");
crate::testmatrix::emit(
crate::testmatrix::COMPONENT,
"source::incremental_fetch_new_advances_cursor_and_ships_delta",
b2_rows == 2 && st2.snapshot_id != st1.snapshot_id,
&format!("second poll shipped {b2_rows} delta rows; cursor advanced"),
);
}
/// A three-step append chain (A→B→C): the delta from A to C spans BOTH
/// new snapshots and reads only their files (not A's). Proves the
/// lineage walk collects multiple snapshots in one delta.
#[tokio::test]
async fn delta_spans_multiple_snapshots() {
let dir = tempfile::tempdir().unwrap();
let snap_a = append_snapshot(dir.path(), &[1, 2]).await;
let snap_b = append_snapshot(dir.path(), &[3, 4]).await;
let snap_c = append_snapshot(dir.path(), &[5, 6]).await;
let src = IcebergSource::new(dir.path());
let (delta, plan) = src
.ingest_delta("people", Some(snap_a), snap_c)
.await
.unwrap();
assert_eq!(
delta.iter().map(|b| b.num_rows()).sum::<usize>(),
4,
"B+C rows only (4), not A's 2 and not a full 6",
);
let delta_rows = delta.iter().map(|b| b.num_rows()).sum::<usize>();
assert_eq!(plan.snapshots, vec![snap_b, snap_c], "oldest-first lineage");
assert_eq!(plan.added_files.len(), 2, "B and C each added one file");
crate::testmatrix::emit(
crate::testmatrix::COMPONENT,
"source::delta_spans_multiple_snapshots",
delta_rows == 4
&& plan.snapshots == vec![snap_b, snap_c]
&& plan.added_files.len() == 2,
&format!("A→C delta spans 2 snapshots, {delta_rows} rows, 2 files"),
);
}
/// Delete `ids` from `people` via a real Iceberg equality-delete
/// snapshot (skade), returning its snapshot id.
async fn delete_snapshot(dir: &std::path::Path, ids: &[i64]) -> i64 {
let wh = skade::Warehouse::open(dir).await.unwrap();
let mut table = wh.table("people").await.unwrap();
let key = SkadeBatch::try_new(
Arc::new(skade::arrow_schema::Schema::new(vec![
skade::arrow_schema::Field::new(
"id",
skade::arrow_schema::DataType::Int64,
false,
),
])),
vec![Arc::new(Int64Array::from(ids.to_vec()))],
)
.unwrap();
table.delete_equality(&key, &["id"]).await.unwrap();
table.current_snapshot_id().unwrap()
}
/// Collect (id, change_type) from a CDC delta's batches.
fn id_change_pairs(batches: &[RecordBatch]) -> Vec<(i64, String)> {
use arrow_array::StringArray;
let mut out = Vec::new();
for b in batches {
let id = b
.column(b.schema().index_of("id").unwrap())
.as_any()
.downcast_ref::<arrow_array::Int64Array>()
.unwrap();
let ct = b
.column(b.schema().index_of(SyncMode::CHANGE_TYPE_COLUMN).unwrap())
.as_any()
.downcast_ref::<StringArray>()
.unwrap();
for i in 0..b.num_rows() {
out.push((id.value(i), ct.value(i).to_string()));
}
}
out
}
/// CDC inject-assert: snapshot A inserts {1,2,3}; snapshot B is a real
/// Iceberg equality-delete of id=2. The Incremental delta over A→B must
/// emit CDC `_change_type` rows — id=2 tagged DELETE — so the listener
/// removes that node. (The pure-Rust delete path, no Spark.)
#[tokio::test]
async fn cdc_delta_emits_change_type_delete_for_equality_delete() {
let dir = tempfile::tempdir().unwrap();
let snap_a = append_snapshot(dir.path(), &[1, 2, 3]).await;
let src = IcebergSource::new(dir.path());
// Pure-append delta carries NO _change_type (back-compat).
let (appended, plan) = src.ingest_delta("people", None, snap_a).await.unwrap();
assert!(plan.delete_files.is_empty(), "no deletes yet");
assert!(
appended
.iter()
.all(|b| b.schema().index_of(SyncMode::CHANGE_TYPE_COLUMN).is_err()),
"an append-only delta is untagged (flows straight to upserts)",
);
// Snapshot B: equality-delete id=2.
let snap_b = delete_snapshot(dir.path(), &[2]).await;
assert_ne!(snap_a, snap_b);
// THE CDC DELTA A→B: a _change_type column, id=2 = DELETE.
let (delta, plan) = src
.ingest_delta("people", Some(snap_a), snap_b)
.await
.unwrap();
assert!(!plan.needs_full_reload, "equality-delete needs no reload");
assert_eq!(plan.delete_files.len(), 1, "one equality-delete file");
assert!(
plan.added_files.is_empty(),
"the delete added no data files"
);
let pairs = id_change_pairs(&delta);
assert_eq!(
pairs,
vec![(2i64, "DELETE".to_string())],
"the delta is exactly: delete id=2",
);
// And it routes through the listener as a real vertex delete.
let change = crate::transform::partition_change_rows(&delta).unwrap();
assert!(change.upserts.iter().all(|b| b.num_rows() == 0));
assert_eq!(
change.deletes.iter().map(|b| b.num_rows()).sum::<usize>(),
1,
"one delete row routed to the delete pipeline",
);
let vt: Arc<dyn crate::transform::Transform> =
Arc::new(crate::transform::VertexTransform {
label: "Person".into(),
id_column: "id".into(),
properties: vec![],
rename: Default::default(),
});
let ctx = crate::transform::TableContext {
table: "people".into(),
};
let dels = crate::transform::deletes_from_change_rows(
std::slice::from_ref(&vt),
&ctx,
&change.deletes,
)
.unwrap();
assert_eq!(dels.len(), 1, "one VertexDelete emitted");
assert_eq!(dels[0].label, "Person");
crate::testmatrix::emit(
crate::testmatrix::COMPONENT,
"source::cdc_delta_emits_change_type_delete_for_equality_delete",
pairs == vec![(2i64, "DELETE".to_string())]
&& dels.len() == 1
&& dels[0].label == "Person",
&format!(
"equality-delete A→B tagged id=2 DELETE → 1 VertexDelete; pairs={pairs:?}"
),
);
}
}
}
// NOTE: row-level deletes now flow through the pure-Rust path as Iceberg
// *equality* deletes (skade's `Table::delete_equality` writes them, `read_delta`
// surfaces them, and `delta_from_table` tags them `_change_type = DELETE`). The
// `needs_full_reload` branch remains for *position* deletes (delete-by-offset,
// unmappable to a graph identity) and overwrite/replace commits, which an
// additive delta can't express. skade can't *produce* those via `fast_append`,
// so that branch is covered by the code path + the design doc, not a unit test.