crates/knut-bifrost/src/delivery.rs
//! **Iceberg → table delivery**: the loop that reads an Iceberg table's change
//! windows and hands them to a [`TableSink`] (Doris), one snapshot at a time.
//!
//! Feature `skade`. The pieces it drives already existed — skade's
//! `read_changelog`, [`partition_change_rows`], [`deliver_window`] — and
//! nothing ran them end to end. This is that driver, and nothing more.
//!
//! # Where the table is
//!
//! [`IcebergLocator`] names the table three ways, one per kind of lake:
//!
//! | locator | how the table is found | feature |
//! |---|---|---|
//! | `Warehouse` | an embedded skade warehouse directory | `skade` |
//! | `S3` | its S3 root, **no catalog** — `version-hint.text`, else the newest `*.metadata.json` | `skade-s3` |
//! | `Catalog` | any `dyn skade::connect::CatalogConnector`, `ns.name` in it | `skade` |
//!
//! The catalogs bifrost's flags build a `Catalog` locator from:
//!
//! | connector | catalog | flag | feature |
//! |---|---|---|---|
//! | `IcebergRest` | an Iceberg REST catalog; MinIO AIStor's `/_iceberg` with SigV4 service `s3tables` | `--rest-uri` | `skade-rest` |
//! | `SkadeGrpc` | a skade-katalog gRPC front (`skade-katalog serve --grpc`) | `--grpc-uri` | `skade-grpc` |
//!
//! The cursor key names the catalog by its connector's `catalog_id`, which
//! for a REST catalog is the `rest:<uri>#<warehouse>` form the key always had.
//!
//! Every round loads the table again, so a new commit is seen on the next round
//! whichever way the table is found.
//!
//! # One snapshot per window
//!
//! A window is `(previous snapshot, next snapshot]`, never a wider range:
//!
//! * **memory** — `read_changelog` materialises the whole window; one commit at
//! a time bounds it by the largest single commit, not by the backlog;
//! * **ordering** — inside one Iceberg commit an update is an equality delete
//! plus an append of the same key; the sink ships deletes before upserts, so
//! a one-commit window is always applied in the right order. A multi-commit
//! window could hold insert-then-delete of one key, which the sink cannot
//! order (`sink::doris` says so);
//! * **replay** — the label is derived from the snapshot pair, so a crash
//! replays exactly the commit that was in flight.
//!
//! # The initial load, and what is refused after it
//!
//! * **No cursor** (first run): the whole **current** snapshot as inserts, one
//! window `(None, current]` (`skade::read_snapshot_changelog`) — never a
//! replay of history. It is held in memory in full; the log says so.
//! * **Then one snapshot per window**, whatever the writer did — appends,
//! copy-on-write overwrites, merge-on-read deletes (v3 deletion vectors,
//! which Spark 4.1 + Iceberg 1.12 writes for `MERGE INTO`), equality deletes.
//! Each window carries its snapshot's data sequence number, which a Unique Key
//! destination versions every row by.
//! * A **`replace`** (compaction) changes no row: delivered as an empty window,
//! no read, no load, the cursor moves past it.
//! * A commit skade cannot read is refused and the cursor held
//! (`refuse_full_reload` in `sink::table`).
//! * The **cursor's snapshot expired** from the table: a
//! [`BifrostError::FullReloadRequired`] error, never a silent resync. The
//! operator empties the destination, deletes the cursor, and restarts.
//!
//! # Near-live
//!
//! [`IcebergDelivery::run`] polls (default [`DEFAULT_POLL`]): every round
//! re-loads the table metadata through its [`IcebergLocator`] and drains the
//! new snapshots, oldest first. A stop signal is honoured between snapshots,
//! after the in-flight one's cursor commit.
use std::sync::Arc;
use std::time::{Duration, Instant};
use skade::connect::CatalogConnector;
use skade::iceberg::spec::Operation;
use skade::iceberg::table::Table as IceTable;
use crate::{
error::{BifrostError, Result},
sink::table::{SnapshotWindow, TableSink, TableWindow, deliver_window},
state::SyncState,
transform::partition_change_rows,
};
/// Where an Iceberg table is. See the module docs.
#[derive(Debug, Clone)]
pub enum IcebergLocator {
/// An embedded skade warehouse at `dir`, table `table` (`ns.name` or `name`).
Warehouse { dir: std::path::PathBuf, table: String },
/// A table on S3/MinIO by its root, no catalog.
#[cfg(feature = "skade-s3")]
S3(skade::connect::S3Table),
/// A table behind a catalog — an Iceberg REST catalog (`IcebergRest`), a
/// skade-katalog gRPC front (`SkadeGrpc`), any [`CatalogConnector`] —
/// `table` = `ns.name`.
Catalog { connector: Arc<dyn CatalogConnector>, table: String },
}
impl IcebergLocator {
/// The table `table` (`ns.name`) behind `connector`.
pub fn catalog(connector: impl CatalogConnector + 'static, table: impl Into<String>) -> Self {
IcebergLocator::Catalog { connector: Arc::new(connector), table: table.into() }
}
/// A stable, secret-free name for the table — the cursor key's source half.
pub fn source_id(&self) -> String {
match self {
IcebergLocator::Warehouse { dir, table } => format!("skade:{}#{table}", dir.display()),
#[cfg(feature = "skade-s3")]
IcebergLocator::S3(t) => format!("s3:{}", t.location.trim_end_matches('/')),
IcebergLocator::Catalog { connector, table } => format!("{}#{table}", connector.catalog_id()),
}
}
/// Load the table at its current metadata.
pub async fn load(&self) -> Result<IceTable> {
let src = |e: skade::SkadeError| BifrostError::Source(format!("{}: {e}", self.source_id()));
match self {
IcebergLocator::Warehouse { dir, table } => {
let wh = skade::Warehouse::open(dir).await.map_err(src)?;
Ok(wh.table(table).await.map_err(src)?.inner().clone())
}
#[cfg(feature = "skade-s3")]
IcebergLocator::S3(t) => t.load().await.map_err(src),
IcebergLocator::Catalog { connector, table } => {
let ident = skade::iceberg::TableIdent::from_strs(table.split('.')).map_err(|e| {
BifrostError::Source(format!("{}: bad table name {table:?}: {e}", self.source_id()))
})?;
let cat = connector.open().await.map_err(src)?;
cat.load_table(&ident)
.await
.map_err(|e| BifrostError::Source(format!("{}: {e}", self.source_id())))
}
}
}
}
/// Where a lake of **many** Iceberg tables is — an [`IcebergLocator`] per table,
/// on demand. `bifrost watch` streams every table a mapping names, so it holds a
/// lake and asks it for each table by the mapping's name:
///
/// | lake | the mapping's `table:` | feature |
/// |---|---|---|
/// | `Warehouse` | `name` or `ns.name` in the embedded skade warehouse | `skade` |
/// | `S3` | an `s3://…` table root verbatim, else `ns.name` → `<root>/ns/name` | `skade-s3` |
/// | `Catalog` | `ns.name` in the catalog, a bare `name` in its `namespace` — REST (MinIO AIStor: `/_iceberg`, SigV4 `s3tables`) or skade-katalog gRPC | `skade-rest` / `skade-grpc` |
#[derive(Debug, Clone)]
pub enum IcebergLake {
/// An embedded skade warehouse directory.
Warehouse(std::path::PathBuf),
/// Iceberg tables on S3/MinIO read with no catalog: each table's current
/// metadata is found under its root (`version-hint.text`, else the newest
/// `*.metadata.json`).
#[cfg(feature = "skade-s3")]
S3 { s3: skade::S3Config, root: String },
/// A catalog (Iceberg REST, skade-katalog gRPC, …); the tables' files are
/// read the way its connector says. A bare table name (no `ns.`) is in
/// `namespace`, if one is given.
Catalog { connector: Arc<dyn CatalogConnector>, namespace: Option<String> },
}
impl IcebergLake {
/// The locator for the mapping's table `table`.
pub fn locate(&self, table: &str) -> IcebergLocator {
match self {
IcebergLake::Warehouse(dir) => IcebergLocator::Warehouse { dir: dir.clone(), table: table.to_string() },
#[cfg(feature = "skade-s3")]
IcebergLake::S3 { s3, root } => {
let location = if bucket_of(table).is_some() {
table.to_string()
} else {
format!("{}/{}", root.trim_end_matches('/'), table.replace('.', "/"))
};
IcebergLocator::S3(skade::connect::S3Table { s3: s3.clone(), location })
}
IcebergLake::Catalog { connector, namespace } => {
let table = match namespace {
Some(ns) if !table.contains('.') => format!("{ns}.{table}"),
_ => table.to_string(),
};
IcebergLocator::Catalog { connector: connector.clone(), table }
}
}
}
/// Where the lake is, for logs. Never a secret.
pub fn describe(&self) -> String {
match self {
IcebergLake::Warehouse(dir) => format!("skade warehouse {}", dir.display()),
#[cfg(feature = "skade-s3")]
IcebergLake::S3 { s3, root } => format!("s3 {root} via {}", s3.endpoint),
IcebergLake::Catalog { connector, namespace } => match namespace {
Some(ns) => format!("{} namespace {ns}", connector.describe()),
None => connector.describe(),
},
}
}
}
#[cfg(feature = "skade-s3")]
/// The bucket named in an `s3://bucket/…` URL (`s3a://` and `s3n://`, which Spark
/// writes, too).
pub fn bucket_of(url: &str) -> Option<&str> {
["s3://", "s3a://", "s3n://"]
.iter()
.find_map(|s| url.strip_prefix(s))
.and_then(|r| r.split('/').next())
.filter(|b| !b.is_empty())
}
#[cfg(feature = "skade-s3")]
fn env(name: &str) -> Option<String> {
std::env::var(name).ok().filter(|v| !v.is_empty())
}
#[cfg(feature = "skade-s3")]
fn need_env(name: &str) -> Result<String> {
env(name).ok_or_else(|| BifrostError::Other(format!("{name} is not set (secrets come from the environment, never a flag)")))
}
#[cfg(feature = "skade-s3")]
/// The CA to verify an `https://` S3/REST endpoint with: `flag` (a PEM file),
/// else `BIFROST_S3_CA_PEM` (the PEM text) written to `<state_dir>/s3-ca.pem`.
pub fn s3_ca(flag: Option<&std::path::Path>, state_dir: &std::path::Path) -> Result<Option<std::path::PathBuf>> {
if let Some(p) = flag {
return Ok(Some(p.to_path_buf()));
}
let Some(pem) = env("BIFROST_S3_CA_PEM") else { return Ok(None) };
std::fs::create_dir_all(state_dir)?;
let path = state_dir.join("s3-ca.pem");
std::fs::write(&path, pem)?;
Ok(Some(path))
}
#[cfg(feature = "skade-s3")]
/// The S3/MinIO config for a lake's files. The keys come from `MINIO_ACCESS_KEY`
/// and `MINIO_SECRET_KEY` only — a flag lands in `/proc/<pid>/cmdline`. They sign
/// the S3 reads and, for a REST catalog with SigV4 (AIStor), every catalog call.
pub fn s3_config(endpoint: String, region: String, bucket: String, ca: Option<std::path::PathBuf>) -> Result<skade::S3Config> {
Ok(skade::S3Config {
endpoint,
region,
bucket,
access_key_id: need_env("MINIO_ACCESS_KEY")?,
secret_access_key: need_env("MINIO_SECRET_KEY")?,
path_style: true,
tls: ca.map(|ca| skade::tls::ClientTls { ca_pem: Some(ca), ..Default::default() }),
})
}
/// Knobs for [`IcebergDelivery`].
#[derive(Debug, Clone)]
pub struct DeliveryOptions {
/// At most this many one-snapshot windows per [`IcebergDelivery::round`];
/// the rest wait for the next round, so one round never runs unbounded.
pub max_windows_per_round: usize,
/// The row identity, when the Iceberg table declares no identifier fields
/// (a plain Spark table) — the destination's key columns. Empty = what the
/// changelog knows (identifier fields, equality-delete ids).
pub key_columns: Vec<String>,
}
impl Default for DeliveryOptions {
fn default() -> Self {
Self { max_windows_per_round: 64, key_columns: Vec::new() }
}
}
/// Default seconds between polls while healthy: near-live, and cheap — a poll
/// that finds no new snapshot reads one metadata file.
pub const DEFAULT_POLL: Duration = Duration::from_secs(5);
/// What one [`IcebergDelivery::round`] did.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct RoundOutcome {
/// Windows delivered (and cursor advances made) this round.
pub windows: usize,
pub rows_loaded: usize,
pub rows_deleted: usize,
/// Rows the store refused as an already-seen label: a replay, not an error.
pub rows_duplicate: usize,
/// The snapshot the cursor stands on after the round.
pub at_snapshot: Option<i64>,
/// Snapshots still to deliver after this round (0 = caught up).
pub behind: usize,
/// Wall time reading changelogs (table load + read_changelog + partition).
pub read: Duration,
/// Wall time inside the sink (encode + load + cursor commit).
pub load: Duration,
}
/// The delivery loop: one Iceberg table into one table sink.
pub struct IcebergDelivery<S: TableSink> {
locator: IcebergLocator,
sink: S,
state: SyncState,
key: String,
opts: DeliveryOptions,
probed: bool,
}
impl<S: TableSink> IcebergDelivery<S> {
/// `destination` names where the rows go (e.g. `doris:db.table`); it is
/// part of the cursor key, so one source delivered to two destinations
/// keeps two cursors.
pub fn new(
locator: IcebergLocator,
destination: &str,
sink: S,
state: SyncState,
opts: DeliveryOptions,
) -> Self {
let key = format!("{} -> {destination}", locator.source_id());
Self { locator, sink, state, key, opts, probed: false }
}
/// The cursor key this delivery reads and advances.
pub fn cursor_key(&self) -> &str {
&self.key
}
/// The snapshot the cursor stands on.
pub fn cursor(&self) -> Option<i64> {
self.state.get(&self.key).snapshot_id
}
/// Deliver what is waiting, up to [`DeliveryOptions::max_windows_per_round`]
/// windows. Every delivered window has advanced the cursor before the next
/// one is read; an error leaves the cursor on the last delivered window.
pub async fn round(&mut self) -> Result<RoundOutcome> {
let mut out = RoundOutcome::default();
let t0 = Instant::now();
let table = self.locator.load().await?;
out.read += t0.elapsed();
let from = self.cursor();
out.at_snapshot = from;
let Some(current) = table.metadata().current_snapshot().map(|s| s.snapshot_id()) else {
return Ok(out); // no snapshot yet: nothing to deliver
};
if from == Some(current) {
return Ok(out);
}
// First run: no cursor. The initial load is the whole current snapshot
// as inserts — never a replay of history, which may hold overwrites,
// deletes and compactions the destination never saw.
let Some(from_id) = from else {
tracing::warn!(
target: "bifrost::delivery",
event = "initial_full_snapshot",
table = %self.key,
to = current,
"no cursor: the initial load is the whole current snapshot, held in memory in full",
);
self.deliver(&table, None, current, &mut out).await?;
out.behind = 0;
return Ok(out);
};
// The cursor's snapshot must still be in the table. Snapshot expiry can
// remove it, and then the changes since it are unreadable — say so,
// never resync silently.
if table.metadata().snapshot_by_id(from_id).is_none() {
return Err(BifrostError::FullReloadRequired(format!(
"{}: the cursor's snapshot {from_id} has been expired from the table, so the \
changes since it can no longer be read. Empty the destination table, delete \
this table's cursor, and restart: the next run does the initial full load.",
self.key
)));
}
let chain = table
.metadata()
.snapshots_added_between(from, current)
.map_err(|e| BifrostError::Source(format!("{}: snapshot lineage: {e}", self.key)))?;
let mut prev = from;
for (i, &snap) in chain.iter().enumerate() {
if i == self.opts.max_windows_per_round {
out.behind = chain.len() - i;
break;
}
self.deliver(&table, prev, snap, &mut out).await?;
prev = Some(snap);
}
Ok(out)
}
/// Read one window, probe the sink once, deliver, account.
///
/// `from = None` is the initial load: the current snapshot's live rows.
/// A `replace` snapshot (compaction) changes no row: it is delivered as an
/// empty window, so the cursor moves past it without a read or a load.
async fn deliver(&mut self, table: &IceTable, from: Option<i64>, to: i64, out: &mut RoundOutcome) -> Result<()> {
let src = |e: skade::SkadeError| BifrostError::Source(format!("{e}"));
let t = Instant::now();
if !self.probed {
let schema = skade::arrow_schema_of(table).map_err(src)?;
self.sink.probe(&schema).await?;
self.probed = true;
}
let snapshot = table.metadata().snapshot_by_id(to);
let replace = from.is_some() && snapshot.is_some_and(|s| matches!(s.summary().operation, Operation::Replace));
let sequence = snapshot.map(|s| s.sequence_number());
let (rows, keys, needs_full_reload) = if replace {
(crate::transform::ChangeRows::default(), Vec::new(), false)
} else {
let cl = match from {
None => skade::read_snapshot_changelog(table, to).await.map_err(src)?,
Some(_) => skade::read_changelog(table, from, to).await.map_err(src)?,
};
// An unreadable commit (a delete shape skade's planner does not
// know): `deliver_window` refuses the window and holds the cursor.
let needs_full_reload = cl.plan.needs_full_reload && from.is_some();
let keys = if self.opts.key_columns.is_empty() { cl.key_columns().to_vec() } else { self.opts.key_columns.clone() };
(partition_change_rows(&cl.rows)?, keys, needs_full_reload)
};
out.read += t.elapsed();
let window = TableWindow {
table: &self.key,
snapshots: SnapshotWindow::new(from, to),
upserts: &rows.upserts,
deletes: &rows.deletes,
needs_full_reload,
sequence,
key_columns: &keys,
};
let t = Instant::now();
let ack = deliver_window(&mut self.sink, &mut self.state, &window).await?;
out.load += t.elapsed();
out.windows += 1;
out.rows_loaded += ack.rows_loaded;
out.rows_deleted += ack.rows_deleted;
out.rows_duplicate += ack.rows_rejected_duplicate;
out.at_snapshot = Some(ack.to_snapshot());
Ok(())
}
/// Run rounds until `stop` resolves: `poll` between rounds while healthy,
/// doubling up to `max_backoff` while rounds fail. A round in flight when
/// `stop` fires is finished first — its cursor commit included — so a
/// SIGTERM never leaves a loaded-but-uncommitted window behind.
pub async fn run(
&mut self,
poll: Duration,
max_backoff: Duration,
stop: impl std::future::Future<Output = ()>,
) -> Result<()> {
let mut backoff = crate::metrics::Backoff::new(poll, max_backoff);
tokio::pin!(stop);
loop {
let result = self.round().await;
let failed = match &result {
Ok(o) => {
if o.windows > 0 {
tracing::info!(
target: "bifrost::delivery",
event = "round",
table = %self.key,
windows = o.windows,
loaded = o.rows_loaded,
deleted = o.rows_deleted,
duplicate = o.rows_duplicate,
at = ?o.at_snapshot,
behind = o.behind,
read_ms = o.read.as_millis() as u64,
load_ms = o.load.as_millis() as u64,
"delivered",
);
}
false
}
Err(e) => {
tracing::error!(target: "bifrost::delivery", event = "round_failed", table = %self.key, error = %e, "round failed, cursor held");
true
}
};
let behind = matches!(&result, Ok(o) if o.behind > 0);
let sleep = if behind { Duration::ZERO } else { backoff.next_sleep(failed) };
tokio::select! {
_ = &mut stop => return Ok(()),
_ = tokio::time::sleep(sleep) => {}
}
}
}
}
#[cfg(test)]
mod tests {
use std::sync::{Arc, Mutex};
use arrow_array::{Int64Array, StringArray};
use arrow_schema::{DataType, Field, Schema};
use async_trait::async_trait;
use super::*;
use arrow_array::RecordBatch;
use crate::sink::table::{AckToken, LoadAck};
/// What the sink saw of one window.
#[derive(Clone, Debug, PartialEq, Eq)]
struct Seen {
from: Option<i64>,
to: i64,
upserts: usize,
deletes: usize,
sequence: Option<i64>,
keys: Vec<String>,
}
/// A sink that records every window it is handed.
#[derive(Clone, Default)]
struct Recording {
windows: Arc<Mutex<Vec<Seen>>>,
probes: Arc<Mutex<usize>>,
fail_at: Option<i64>,
}
impl Recording {
fn seen(&self) -> Vec<Seen> {
self.windows.lock().unwrap().clone()
}
}
#[async_trait]
impl TableSink for Recording {
async fn probe(&mut self, _: &Schema) -> Result<()> {
*self.probes.lock().unwrap() += 1;
Ok(())
}
async fn apply(&mut self, w: &TableWindow<'_>, token: AckToken) -> Result<LoadAck> {
if self.fail_at == Some(w.snapshots.to) {
return Err(BifrostError::Other("sink down".into()));
}
self.windows.lock().unwrap().push(Seen {
from: w.snapshots.from,
to: w.snapshots.to,
upserts: w.upsert_rows(),
deletes: w.delete_rows(),
sequence: w.sequence,
keys: w.key_columns.to_vec(),
});
Ok(token.ack(w.snapshots.to, vec![]).with_counts(w.upsert_rows(), w.delete_rows(), 0))
}
}
fn people(ids: &[i64]) -> skade::arrow_array::RecordBatch {
let schema = Arc::new(Schema::new(vec![
Field::new("id", DataType::Int64, false),
Field::new("name", DataType::Utf8, true),
]));
let names: Vec<String> = ids.iter().map(|i| format!("p{i}")).collect();
RecordBatch::try_new(
schema,
vec![Arc::new(Int64Array::from(ids.to_vec())), Arc::new(StringArray::from(names))],
)
.unwrap()
}
/// The embedded warehouse's catalog (redb) admits one open handle at a
/// time; the delivery opens its own every round. Retry while it is held,
/// and keep every handle short-lived.
async fn table(dir: &std::path::Path) -> skade::Table {
let deadline = Instant::now() + Duration::from_secs(10);
loop {
match skade::Warehouse::open(dir).await {
Ok(wh) => return wh.table("people").await.unwrap(),
Err(e) if Instant::now() < deadline => {
let _ = e;
tokio::time::sleep(Duration::from_millis(5)).await;
}
Err(e) => panic!("warehouse stayed locked: {e}"),
}
}
}
async fn warehouse_with(commits: &[&[i64]]) -> (tempfile::TempDir, Vec<i64>) {
let dir = tempfile::tempdir().unwrap();
{
let wh = skade::Warehouse::open(dir.path()).await.unwrap();
wh.create_table("people", &people(&[]).schema()).await.unwrap();
}
let snaps = append(dir.path(), commits).await;
(dir, snaps)
}
async fn append(dir: &std::path::Path, commits: &[&[i64]]) -> Vec<i64> {
let mut t = table(dir).await;
let mut snaps = Vec::new();
for ids in commits {
t.append(&[people(ids)]).await.unwrap();
snaps.push(t.current_snapshot_id().unwrap());
}
snaps
}
async fn seq_of(dir: &std::path::Path, snap: i64) -> i64 {
let t = table(dir).await;
t.inner().metadata().snapshot_by_id(snap).unwrap().sequence_number()
}
fn delivery(dir: &std::path::Path, sink: Recording, opts: DeliveryOptions) -> IcebergDelivery<Recording> {
let state = SyncState::open(dir.join("cursor.json")).unwrap();
let loc = IcebergLocator::Warehouse { dir: dir.to_path_buf(), table: "people".into() };
IcebergDelivery::new(loc, "test:db.people", sink, state, opts)
}
fn keyed() -> DeliveryOptions {
DeliveryOptions { key_columns: vec!["id".into()], ..DeliveryOptions::default() }
}
#[tokio::test]
async fn the_first_run_loads_the_current_snapshot_then_one_window_per_snapshot() {
let (dir, snaps) = warehouse_with(&[&[1, 2], &[3], &[4, 5, 6]]).await;
let sink = Recording::default();
let mut d = delivery(dir.path(), sink.clone(), keyed());
let o = d.round().await.unwrap();
assert_eq!((o.windows, o.rows_loaded), (1, 6), "the initial load is one window: the live rows");
assert_eq!(d.cursor(), Some(snaps[2]));
let s0 = seq_of(dir.path(), snaps[2]).await;
assert_eq!(
sink.seen(),
vec![Seen { from: None, to: snaps[2], upserts: 6, deletes: 0, sequence: Some(s0), keys: vec!["id".into()] }]
);
assert_eq!(*sink.probes.lock().unwrap(), 1, "probe runs once, before the first window");
// Then strictly one window per new snapshot, oldest first, each with its
// own sequence number.
let more = append(dir.path(), &[&[7], &[8, 9]]).await;
let o = d.round().await.unwrap();
assert_eq!(o.windows, 2);
let w = sink.seen();
assert_eq!((w[1].from, w[1].to, w[1].upserts), (Some(snaps[2]), more[0], 1));
assert_eq!((w[2].from, w[2].to, w[2].upserts), (Some(more[0]), more[1], 2));
assert!(w[1].sequence < w[2].sequence && w[0].sequence < w[1].sequence, "{w:?}");
assert_eq!(d.cursor(), Some(more[1]));
// Caught up: nothing more, cursor unchanged.
assert_eq!(d.round().await.unwrap().windows, 0);
}
#[tokio::test]
async fn a_round_is_bounded_and_says_how_far_behind_it_is() {
let (dir, _) = warehouse_with(&[&[0]]).await;
let mut d = delivery(dir.path(), Recording::default(), DeliveryOptions { max_windows_per_round: 2, ..DeliveryOptions::default() });
d.round().await.unwrap();
let snaps = append(dir.path(), &[&[1], &[2], &[3]]).await;
let o = d.round().await.unwrap();
assert_eq!((o.windows, o.behind), (2, 1));
assert_eq!(d.cursor(), Some(snaps[1]));
let o = d.round().await.unwrap();
assert_eq!((o.windows, o.behind), (1, 0));
assert_eq!(d.cursor(), Some(snaps[2]));
}
#[tokio::test]
async fn a_failed_window_holds_the_cursor_on_the_last_delivered_one() {
let (dir, first) = warehouse_with(&[&[0]]).await;
delivery(dir.path(), Recording::default(), DeliveryOptions::default()).round().await.unwrap();
let snaps = append(dir.path(), &[&[1], &[2], &[3]]).await;
let sink = Recording { fail_at: Some(snaps[1]), ..Recording::default() };
let mut d = delivery(dir.path(), sink, DeliveryOptions::default());
assert_eq!(d.cursor(), Some(first[0]));
assert!(d.round().await.is_err());
assert_eq!(d.cursor(), Some(snaps[0]), "window 1 landed, window 2 failed: cursor on 1");
// The cursor survives a restart (it is on disk), and delivery resumes at 2.
let sink = Recording::default();
let mut d = delivery(dir.path(), sink.clone(), DeliveryOptions::default());
assert_eq!(d.cursor(), Some(snaps[0]));
d.round().await.unwrap();
let w = sink.seen();
assert_eq!(w.first().map(|w| (w.from, w.to)), Some((Some(snaps[0]), snaps[1])));
assert_eq!(d.cursor(), Some(snaps[2]));
}
/// A v3 merge-on-read delete (a deletion vector — what Spark 4.1 + Iceberg
/// 1.12 writes for MERGE INTO) is delivered as delete rows, keyed.
#[tokio::test]
async fn a_deletion_vector_is_delivered_as_keyed_deletes() {
let (dir, _) = warehouse_with(&[&[1, 2, 3]]).await;
let sink = Recording::default();
let mut d = delivery(dir.path(), sink.clone(), keyed());
d.round().await.unwrap();
{
let mut t = table(dir.path()).await;
let f = t.live_data_files().await.unwrap().remove(0);
t.delete_positions(&f, &[1]).await.unwrap(); // id 2
t.append(&[people(&[2])]).await.unwrap(); // …re-inserted: an update
}
let o = d.round().await.unwrap();
assert_eq!(o.windows, 2);
let w = sink.seen();
assert_eq!((w[1].upserts, w[1].deletes, w[1].keys.clone()), (0, 1, vec!["id".to_string()]));
assert_eq!((w[2].upserts, w[2].deletes), (1, 0));
}
/// A compaction (`replace`) changes no row: an empty window, the cursor
/// moves past it, nothing is read or loaded.
#[tokio::test]
async fn a_replace_snapshot_is_skipped_as_an_empty_window() {
let (dir, _) = warehouse_with(&[&[1], &[2]]).await;
let sink = Recording::default();
let mut d = delivery(dir.path(), sink.clone(), DeliveryOptions::default());
d.round().await.unwrap();
let replaced = {
let mut t = table(dir.path()).await;
let files = t.live_data_files().await.unwrap();
t.rewrite_files(&files, vec![vec![people(&[1, 2])]], None).await.unwrap();
t.current_snapshot_id().unwrap()
};
let o = d.round().await.unwrap();
assert_eq!(o.windows, 1);
let w = sink.seen();
assert_eq!((w[1].to, w[1].upserts, w[1].deletes), (replaced, 0, 0));
assert_eq!(d.cursor(), Some(replaced));
}
/// The cursor's snapshot expired from the table: an explicit error, the
/// cursor untouched, nothing delivered.
#[tokio::test]
async fn an_expired_cursor_requires_a_full_reload() {
let (dir, _) = warehouse_with(&[&[1]]).await;
let sink = Recording::default();
let mut d = delivery(dir.path(), sink.clone(), DeliveryOptions::default());
let mut c = d.state.get(&d.key);
c.snapshot_id = Some(4242); // a snapshot this table does not have (any more)
d.state.advance_staged(&d.key.clone(), c).await.unwrap();
let err = d.round().await.unwrap_err();
assert!(matches!(err, BifrostError::FullReloadRequired(_)), "{err}");
assert_eq!(d.cursor(), Some(4242));
assert!(sink.seen().is_empty());
}
/// A `Catalog` locator — any `dyn CatalogConnector` — delivers exactly what
/// the warehouse locator does; here the same warehouse opened as `SkadeLocal`.
#[tokio::test]
async fn a_catalog_locator_delivers_what_the_warehouse_locator_does() {
let (dir, _) = warehouse_with(&[&[1, 2], &[3]]).await;
let (sa, sb) = (Recording::default(), Recording::default());
let state = |n: &str| SyncState::open(dir.path().join(n)).unwrap();
let wh = IcebergLocator::Warehouse { dir: dir.path().to_path_buf(), table: "people".into() };
let cat = IcebergLocator::catalog(skade::connect::SkadeLocal::new(dir.path()), "main.people");
let mut a = IcebergDelivery::new(wh, "test:db.people", sa.clone(), state("a.json"), keyed());
let mut b = IcebergDelivery::new(cat, "test:db.people", sb.clone(), state("b.json"), keyed());
assert_eq!(b.cursor_key(), format!("skade:{}#main.people -> test:db.people", dir.path().display()));
a.round().await.unwrap();
b.round().await.unwrap();
append(dir.path(), &[&[4], &[5, 6]]).await;
assert_eq!(a.round().await.unwrap().windows, 2);
assert_eq!(b.round().await.unwrap().windows, 2);
assert_eq!(sb.seen(), sa.seen());
assert_eq!(b.cursor(), a.cursor());
}
/// The cursor key of a REST table is the form it had before the locator
/// took any connector, so a running delivery keeps its cursor.
#[cfg(feature = "skade-rest")]
#[test]
fn a_rest_table_keeps_its_cursor_key() {
let rest = skade::connect::IcebergRest {
uri: "https://minio:9000/_iceberg".into(),
warehouse: Some("red".into()),
prefix_or_ref: None,
token: None,
tls: None,
s3: skade::S3Config::default(),
read_only: true,
sigv4: None,
..Default::default()
};
assert_eq!(IcebergLocator::catalog(rest, "bench.people").source_id(), "rest:https://minio:9000/_iceberg#red#bench.people");
}
/// A table behind a skade-katalog gRPC front is keyed by the front.
#[cfg(feature = "skade-grpc")]
#[test]
fn a_grpc_table_is_keyed_by_its_front() {
let grpc = skade::connect::SkadeGrpc::local_files("katalog:8182");
assert_eq!(IcebergLocator::catalog(grpc, "db.people").source_id(), "grpc:http://katalog:8182#db.people");
}
/// Near-live: the poll loop picks up a snapshot committed while it runs,
/// and stops cleanly between snapshots.
#[tokio::test]
async fn the_poll_loop_picks_up_a_new_snapshot() {
let (dir, _) = warehouse_with(&[&[1]]).await;
let sink = Recording::default();
let mut d = delivery(dir.path(), sink.clone(), DeliveryOptions::default());
let (stop_tx, stop_rx) = tokio::sync::oneshot::channel::<()>();
let path = dir.path().to_path_buf();
let watcher = sink.clone();
let writer = async move {
// Wait for the initial load, commit a new snapshot, wait for it.
while watcher.seen().is_empty() {
tokio::time::sleep(Duration::from_millis(10)).await;
}
let snap = append(&path, &[&[2, 3]]).await[0];
let deadline = Instant::now() + Duration::from_secs(10);
while !watcher.seen().iter().any(|w| w.to == snap) {
assert!(Instant::now() < deadline, "the poll loop never delivered the new snapshot");
tokio::time::sleep(Duration::from_millis(10)).await;
}
stop_tx.send(()).unwrap();
snap
};
let (run, snap) = tokio::join!(
d.run(Duration::from_millis(20), Duration::from_millis(200), async {
let _ = stop_rx.await;
}),
writer
);
run.unwrap();
assert_eq!(d.cursor(), Some(snap));
assert_eq!(sink.seen().last().map(|w| w.upserts), Some(2));
}
}