crates/knut-bifrost/src/sink/table.rs
//! **Table** sinks — the `TableSink` family, sibling of [`super::graph`].
//!
//! A table sink delivers one CDC *window* (a snapshot pair plus the rows that
//! changed between them) into a row-and-column store. It shares nothing with
//! [`GraphSink`](super::graph::GraphSink) but the crate's error type, and that
//! is deliberate — see `.nornir/doris-sink-design.md` f2.
//!
//! # The one invariant this module exists to enforce
//!
//! **The delivery cursor advances strictly after the sink acknowledges.**
//!
//! * Load, then commit: a crash in between replays the window. The label a sink
//! derives from the snapshot pair is byte-identical on the replay, so the
//! store rejects it and the replay is a no-op.
//! * Commit, then load: a crash in between **silently loses the window**. No
//! error, no retry, no lag metric — the rows are simply not there, and nobody
//! finds out until an answer is wrong.
//!
//! `listener.rs` gets this right for the graph path, but by comment
//! (`listener.rs:301`). A comment is not a mechanism, so the table path spells
//! it out in the type system:
//!
//! * [`LoadAck`] has no public constructor. The only way to make one is
//! [`AckToken::ack`], which **consumes** the token.
//! * The only [`AckToken`] in existence is the one [`deliver_window`] mints and
//! hands to [`TableSink::apply`].
//! * [`deliver_window`] is the only function in this crate that advances a
//! table-delivery cursor, and it does so from [`LoadAck::to_snapshot`] — never
//! from the window it was given.
//!
//! So an edit that tries to advance the cursor early has to first invent a
//! `LoadAck`, and it cannot: the token it would need is not reachable from
//! anywhere except inside a live `apply` call. That is the whole point.
use std::fmt::Write as _;
use arrow_array::RecordBatch;
use arrow_schema::Schema;
use async_trait::async_trait;
use crate::{
error::{BifrostError, Result},
state::SyncState,
};
// ── the window ──────────────────────────────────────────────────────────────
/// The half-open snapshot range a window covers: changes strictly after `from`,
/// up to and including `to`. `from = None` means "from the beginning".
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub struct SnapshotWindow {
pub from: Option<i64>,
pub to: i64,
}
impl SnapshotWindow {
pub const fn new(from: Option<i64>, to: i64) -> Self {
Self { from, to }
}
}
/// One CDC window, ready to deliver.
///
/// `upserts` / `deletes` are the output of
/// [`partition_change_rows`](crate::transform::partition_change_rows), which is
/// reused rather than reimplemented — that is what makes the four-tag mapping
/// correct here for free (`INSERT` / `UPDATE_AFTER` / folded `UPDATE` upsert,
/// `DELETE` deletes, `UPDATE_BEFORE` is dropped because its paired
/// `UPDATE_AFTER` fully overwrites the row, and an unknown tag is a hard error).
/// The `_change_type` sentinel column is already stripped.
#[derive(Debug, Clone, Copy)]
pub struct TableWindow<'a> {
/// The **source** table this window came from (the bifröst/skade table key).
/// Also the key the cursor is stored under. Not the destination table.
pub table: &'a str,
pub snapshots: SnapshotWindow,
pub upserts: &'a [RecordBatch],
pub deletes: &'a [RecordBatch],
/// [`skade::DeltaPlan::needs_full_reload`] for this window: the additive
/// delta cannot express what happened (an overwrite/replace commit, or a
/// position-delete file), so the only correct answer is "re-read
/// everything".
///
/// [`deliver_window`] **refuses** such a window. It never loads it and never
/// moves the cursor. On a 2 B-row table a silent fallback turns a streaming
/// job into an 81 GB rewrite at whatever hour the writer happened to run.
///
/// This is a backstop, not the primary gate — by the time a window exists
/// the expensive read has already happened, because
/// `skade::Table::read_changelog` handles a full-reload window by re-reading
/// the whole `to` snapshot itself. Call [`refuse_full_reload`] against
/// `read_delta`'s plan *before* calling `read_changelog`.
pub needs_full_reload: bool,
/// The Iceberg **data sequence number** of `snapshots.to` (a one-snapshot
/// window), when known. A Unique Key destination with a sequence column
/// versions every row of the window by it, so a replayed or late window can
/// never overwrite newer data (Doris: `function_column.sequence_col`).
/// `None` = unversioned (tests, sources without snapshots).
pub sequence: Option<i64>,
/// The row identity (primary-key column names) when known — from the
/// Iceberg table's identifier fields / equality-delete ids, or configured.
/// A Unique Key sink uses it to drop a delete whose key is upserted in the
/// same window, and to ship identity-only deletes. Empty = unknown.
pub key_columns: &'a [String],
}
impl TableWindow<'_> {
pub fn upsert_rows(&self) -> usize {
self.upserts.iter().map(RecordBatch::num_rows).sum()
}
pub fn delete_rows(&self) -> usize {
self.deletes.iter().map(RecordBatch::num_rows).sum()
}
pub fn is_empty(&self) -> bool {
self.upsert_rows() == 0 && self.delete_rows() == 0
}
/// The schema of the rows in this window, if it has any.
pub fn schema(&self) -> Option<arrow_schema::SchemaRef> {
self.upserts
.first()
.or_else(|| self.deletes.first())
.map(RecordBatch::schema)
}
}
// ── the unforgeable acknowledgement ─────────────────────────────────────────
/// A capability handed to [`TableSink::apply`] and consumed to build the
/// [`LoadAck`] that proves the load happened.
///
/// Zero-sized, not `Clone`, no public constructor. [`deliver_window`] mints
/// exactly one per call.
#[derive(Debug)]
pub struct AckToken(());
impl AckToken {
/// Turn the token into the proof. Consuming `self` is the point: an `apply`
/// implementation gets one token per call and can therefore produce at most
/// one `LoadAck` per call.
#[must_use]
pub fn ack(self, to_snapshot: i64, labels: Vec<String>) -> LoadAck {
LoadAck {
to_snapshot,
labels,
rows_loaded: 0,
rows_deleted: 0,
rows_rejected_duplicate: 0,
}
}
}
/// Proof that a window was durably accepted by a sink.
///
/// There is no public constructor: the only way to obtain one is
/// [`AckToken::ack`], and the only source of an [`AckToken`] is
/// [`deliver_window`]. Advancing a delivery cursor requires one of these, so the
/// "cursor after ack" ordering cannot be inverted by a later edit without
/// deleting this type.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct LoadAck {
/// The snapshot the cursor may now advance to. The cursor reads this, never
/// the window it was handed.
to_snapshot: i64,
/// Every store-side label this window consumed, in chunk order.
pub labels: Vec<String>,
/// Rows the store accepted.
pub rows_loaded: usize,
/// Delete-marked rows the store accepted.
pub rows_deleted: usize,
/// Rows the store refused because their label had already been seen — a
/// replay working exactly as designed, not an error.
pub rows_rejected_duplicate: usize,
}
impl LoadAck {
/// The snapshot the cursor may advance to.
pub const fn to_snapshot(&self) -> i64 {
self.to_snapshot
}
#[must_use]
pub fn with_counts(mut self, loaded: usize, deleted: usize, duplicate: usize) -> Self {
self.rows_loaded = loaded;
self.rows_deleted = deleted;
self.rows_rejected_duplicate = duplicate;
self
}
}
// ── the trait ───────────────────────────────────────────────────────────────
/// A place table-shaped rows go.
#[async_trait]
pub trait TableSink: Send {
/// Fail fast on a schema the sink cannot accept.
///
/// Called once before the first window. Discovering a column-order or type
/// difference mid-stream, after a partial load, is far worse than refusing
/// to boot — so a mismatch here is an error, never a warning.
async fn probe(&mut self, schema: &Schema) -> Result<()>;
/// Apply exactly one window, then consume `token` to acknowledge it.
///
/// **Must be idempotent**: the same window may arrive twice after a crash
/// between the load and the cursor commit. The intended mechanism is a label
/// derived from the snapshot pair (see [`stream_load_label`]) which the
/// store rejects on replay — not a dedup table on this side.
async fn apply(&mut self, window: &TableWindow<'_>, token: AckToken) -> Result<LoadAck>;
}
// ── the choke point ─────────────────────────────────────────────────────────
/// Deliver one window and advance the cursor — **in that order**.
///
/// This is the only function in the crate that advances a table-delivery
/// cursor. It cannot advance it without a [`LoadAck`], and it cannot obtain one
/// without a successful [`TableSink::apply`].
///
/// A failing `apply` leaves the cursor exactly where it was, so the next round
/// recomputes the same window, derives the same labels, and the store rejects
/// whatever already landed.
pub async fn deliver_window<S: TableSink + ?Sized>(
sink: &mut S,
state: &mut SyncState,
window: &TableWindow<'_>,
) -> Result<LoadAck> {
// 0. Refuse before anything else. A full-reload window must never be loaded
// by accident, and must never advance the cursor.
refuse_full_reload(window.table, window.needs_full_reload)?;
// 1. The load. Nothing below this line runs if it fails.
let ack = sink.apply(window, AckToken(())).await?;
// A sink that acks a different snapshot than the one it was asked to load
// would silently skip or re-do a window. Refuse rather than trust it.
if ack.to_snapshot != window.snapshots.to {
return Err(BifrostError::Other(format!(
"{}: sink acked snapshot {} but the window ended at {} — refusing to \
move the cursor",
window.table, ack.to_snapshot, window.snapshots.to,
)));
}
// 2. Only now the cursor, and only from the ack.
let mut next = state.get(window.table);
next.snapshot_id = Some(ack.to_snapshot);
state.advance_staged(window.table, next).await?;
state.flush()?;
tracing::info!(
target: "bifrost::sink",
event = "table_window_delivered",
table = %window.table,
from = ?window.snapshots.from,
to = ack.to_snapshot,
labels = ack.labels.len(),
rows = ack.rows_loaded,
deleted = ack.rows_deleted,
duplicate = ack.rows_rejected_duplicate,
"window loaded, then cursor advanced",
);
crate::testmatrix::emit(
crate::testmatrix::COMPONENT,
"table_sink::cursor_after_ack",
true,
&format!(
"{} advanced to {} after {} label(s)",
window.table,
ack.to_snapshot,
ack.labels.len()
),
);
Ok(ack)
}
// ── the full-reload refusal ─────────────────────────────────────────────────
/// **The pre-check.** Call this with [`skade::DeltaPlan::needs_full_reload`],
/// obtained from `read_delta`, *before* calling `read_changelog`.
///
/// Why the order matters, and it is not obvious: `read_changelog` handles a
/// full-reload window by re-reading the whole `to` snapshot as a pure `INSERT`
/// stream (skade `read.rs:3045-3062`). Checking the flag on the batch it returns
/// means the expensive read has already been paid for. `read_delta` returns
/// early with no rows and the flag set (skade `read.rs:2793`), so this check
/// costs a snapshot-lineage walk and nothing else.
///
/// It also catches a quieter hazard: a driver that used `read_delta` alone on
/// such a window would apply **zero rows** and believe it had succeeded. The
/// refusal is what turns that silence into an error.
///
/// Resolution is an explicit, operator-initiated one-shot reload. Never
/// automatic.
pub fn refuse_full_reload(table: &str, needs_full_reload: bool) -> Result<()> {
if !needs_full_reload {
return Ok(());
}
crate::testmatrix::emit(
crate::testmatrix::COMPONENT,
"table_sink::full_reload_refused",
false,
&format!("{table}: window needs a full reload — refused, cursor held"),
);
tracing::error!(
target: "bifrost::sink",
event = "full_reload_refused",
table = %table,
"window needs a full reload — refusing (cursor not advanced)",
);
Err(BifrostError::Other(format!(
"{table}: this window needs a FULL RELOAD (an overwrite/replace commit or \
a position-delete file defeats the incremental delta). Refusing: a silent \
fallback would rewrite the whole table. Run an explicit one-shot reload, \
then resume."
)))
}
// ── the label ───────────────────────────────────────────────────────────────
/// Doris caps a load label at 128 characters. Other stores are more generous;
/// this is the tightest constraint the builder has to satisfy.
pub const MAX_LABEL_LEN: usize = 128;
/// Build the store-side load label for one chunk of one window.
///
/// ```text
/// skade-<db>:<table>-<from|init>-<to>-<chunk>
/// ```
///
/// **The character set is enforced by Doris, and it was measured, not guessed.**
/// Doris validates a load label against `^[\-_A-Za-z0-9:\p{L}]{1,128}$` and
/// rejects anything else with `[ANALYSIS_ERROR] ... Label format error`. A `.`
/// is **not** allowed — an earlier draft of this used `<db>.<table>` and every
/// load failed. Hence the `:`, which is in the permitted set.
///
/// **Derived, never generated.** A replay of the same window produces the same
/// label with no state kept anywhere, which is what makes the crash story a
/// one-liner: the store rejects the duplicate and the replay is a no-op.
///
/// **Collision-free by construction.** `-` is the structural separator and `:`
/// joins database to table, so neither is in the character set allowed inside a
/// name — every character
/// outside `[A-Za-z0-9_]` becomes `_`, including the `.` between database and
/// table. A label therefore always splits into exactly five `-`-separated
/// fields, and two different `(db, table, from, to, chunk)` tuples cannot render
/// the same string. (Allowing `-` through would let `("x", "y-1", 2, 3)` collide
/// with `("x", "y", 1, 2)`-chunk-3.)
///
/// When the name portion would push the label past [`MAX_LABEL_LEN`] it is
/// truncated **and** suffixed with a hash of the full name, so truncation cannot
/// reintroduce a collision. The snapshot and chunk fields are never truncated.
pub fn stream_load_label(
db: &str,
table: &str,
from: Option<i64>,
to: i64,
chunk: usize,
) -> String {
let mut tail = String::new();
match from {
Some(f) => write!(tail, "-{f}-{to}-{chunk}").expect("string write"),
None => write!(tail, "-init-{to}-{chunk}").expect("string write"),
}
let full_name = format!("{}:{}", label_component(db), label_component(table));
// "skade-" + name + tail
let budget = MAX_LABEL_LEN.saturating_sub("skade-".len() + tail.len());
let name = if full_name.len() <= budget {
full_name
} else {
let digest = format!("_{:016x}", fnv1a64(full_name.as_bytes()));
let keep = budget.saturating_sub(digest.len());
let mut short: String = full_name.chars().take(keep).collect();
short.push_str(&digest);
short
};
format!("skade-{name}{tail}")
}
/// One name (a database or a table) rendered into the label's character set,
/// **injectively**.
///
/// Everything outside `[A-Za-z0-9_]` becomes `_`. `-` and `:` are excluded on
/// purpose: `-` is the label's field separator and `:` joins database to table
/// (see [`stream_load_label`]). Replacement
/// alone is *not* injective: `personer.2g` and `personer_2g` are different
/// tables that would render the same string, and a shared label between two
/// tables is a silently skipped load. So whenever a character actually had to be
/// replaced, a hash of the original is appended, which restores injectivity
/// while leaving well-behaved names (the overwhelming majority) untouched and
/// readable.
///
/// RED-when-broken: drop the suffix and `tests::labels_do_not_collide` fails on
/// exactly that pair — it did, which is why the suffix exists.
fn label_component(s: &str) -> String {
if s.is_empty() {
return "_e".into();
}
let mut changed = false;
let sanitized: String = s
.chars()
.map(|c| {
if c.is_ascii_alphanumeric() || c == '_' {
c
} else {
changed = true;
'_'
}
})
.collect();
if changed {
format!("{sanitized}_{:08x}", fnv1a64(s.as_bytes()) as u32)
} else {
sanitized
}
}
/// Doris validates every load label against this regex, and rejects anything
/// else with `[ANALYSIS_ERROR] ... Label format error`:
///
/// ```text
/// ^[\-_A-Za-z0-9:\p{L}]{1,128}$
/// ```
///
/// Checked here by hand rather than with a regex dependency. `\p{L}` is any
/// Unicode letter; [`stream_load_label`] only ever emits ASCII, so the check is
/// strictly wider than what the builder produces — which is the point: it
/// verifies the *builder*, not the regex.
pub fn is_valid_doris_label(label: &str) -> bool {
!label.is_empty()
&& label.chars().count() <= MAX_LABEL_LEN
&& label.chars().all(|c| {
c == '-' || c == '_' || c == ':' || c.is_ascii_alphanumeric() || c.is_alphabetic()
})
}
/// FNV-1a, 64-bit. Used only to keep a truncated label collision-free; no
/// dependency, deterministic across runs and platforms.
fn fnv1a64(bytes: &[u8]) -> u64 {
let mut h: u64 = 0xcbf2_9ce4_8422_2325;
for b in bytes {
h ^= u64::from(*b);
h = h.wrapping_mul(0x0000_0100_0000_01b3);
}
h
}
#[cfg(test)]
mod tests {
use super::*;
use std::collections::HashSet;
use std::sync::{Arc, Mutex};
use arrow_array::{Int64Array, StringArray};
use arrow_schema::{DataType, Field};
use std::sync::Arc as StdArc;
fn batch(ids: Vec<i64>) -> RecordBatch {
let n = ids.len();
RecordBatch::try_new(
StdArc::new(Schema::new(vec![
Field::new("id", DataType::Int64, false),
Field::new("name", DataType::Utf8, false),
])),
vec![
StdArc::new(Int64Array::from(ids)),
StdArc::new(StringArray::from(
(0..n).map(|i| format!("n{i}")).collect::<Vec<_>>(),
)),
],
)
.unwrap()
}
/// What a fake sink did, in order, so a test can assert *sequence* rather
/// than trust a comment.
#[derive(Debug, Clone, PartialEq, Eq)]
enum Event {
/// `apply` ran; carries the cursor value it observed **on disk** at that
/// moment. `None` = never advanced.
Applied(Option<i64>),
}
struct SpySink {
log: Arc<Mutex<Vec<Event>>>,
/// Path of the file-backed ledger, read from inside `apply`.
state_path: std::path::PathBuf,
fail: bool,
/// Snapshot to ack; `None` = ack the window's own `to` (the honest case).
ack_snapshot: Option<i64>,
}
#[async_trait]
impl TableSink for SpySink {
async fn probe(&mut self, _schema: &Schema) -> Result<()> {
Ok(())
}
async fn apply(&mut self, window: &TableWindow<'_>, token: AckToken) -> Result<LoadAck> {
// Read the DURABLE cursor from inside the load. If the cursor were
// committed first, this would already show the new snapshot.
let seen = std::fs::read_to_string(&self.state_path)
.ok()
.and_then(|s| serde_json::from_str::<serde_json::Value>(&s).ok())
.and_then(|v| {
v.get(window.table)
.and_then(|t| t.get("snapshot_id"))
.and_then(serde_json::Value::as_i64)
});
self.log.lock().unwrap().push(Event::Applied(seen));
if self.fail {
return Err(BifrostError::Other("sink refused".into()));
}
let to = self.ack_snapshot.unwrap_or(window.snapshots.to);
Ok(token.ack(to, vec!["l0".into()]).with_counts(
window.upsert_rows(),
window.delete_rows(),
0,
))
}
}
fn spy(dir: &std::path::Path, fail: bool) -> (SpySink, Arc<Mutex<Vec<Event>>>) {
let log = Arc::new(Mutex::new(Vec::new()));
(
SpySink {
log: Arc::clone(&log),
state_path: dir.join("state.json"),
fail,
ack_snapshot: None,
},
log,
)
}
/// **The data-loss bug, asserted directly.** The sink observes the durable
/// cursor from *inside* `apply`; it must still be at the OLD value. Then,
/// after `deliver_window` returns, it must be at the new one.
///
/// RED-when-broken: swap the two statements in `deliver_window` and the
/// in-apply observation becomes `Some(1043)` — which is exactly the state a
/// crash would make permanent while the rows never landed.
#[tokio::test]
async fn cursor_is_committed_strictly_after_the_load() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("state.json");
let mut state = SyncState::open(&path).unwrap();
let (mut sink, log) = spy(dir.path(), false);
// Round 1: nothing delivered yet.
let ups = [batch(vec![1, 2, 3])];
let w = TableWindow {
table: "bench.doris2g",
snapshots: SnapshotWindow::new(Some(1042), 1043),
upserts: &ups,
deletes: &[],
needs_full_reload: false,
sequence: None,
key_columns: &[],
};
let ack = deliver_window(&mut sink, &mut state, &w).await.unwrap();
assert_eq!(ack.to_snapshot(), 1043);
assert_eq!(ack.rows_loaded, 3);
assert_eq!(
log.lock().unwrap().as_slice(),
&[Event::Applied(None)],
"the load must run while the cursor is still unset"
);
assert_eq!(state.get("bench.doris2g").snapshot_id, Some(1043));
// Round 2: the load must observe 1043, not 1044.
let w2 = TableWindow {
table: "bench.doris2g",
snapshots: SnapshotWindow::new(Some(1043), 1044),
upserts: &ups,
deletes: &[],
needs_full_reload: false,
sequence: None,
key_columns: &[],
};
deliver_window(&mut sink, &mut state, &w2).await.unwrap();
assert_eq!(
log.lock().unwrap()[1],
Event::Applied(Some(1043)),
"the load must run BEFORE the cursor moves to 1044"
);
assert_eq!(state.get("bench.doris2g").snapshot_id, Some(1044));
crate::testmatrix::emit(
crate::testmatrix::COMPONENT,
"table_sink::cursor_strictly_after_load",
state.get("bench.doris2g").snapshot_id == Some(1044),
"apply observed the old cursor both rounds",
);
}
/// A failing load must leave the cursor untouched, so the next round
/// recomputes the identical window (and the identical labels).
#[tokio::test]
async fn a_failed_load_does_not_advance_the_cursor() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("state.json");
let mut state = SyncState::open(&path).unwrap();
let (mut sink, _log) = spy(dir.path(), true);
let ups = [batch(vec![1])];
let w = TableWindow {
table: "t",
snapshots: SnapshotWindow::new(Some(7), 8),
upserts: &ups,
deletes: &[],
needs_full_reload: false,
sequence: None,
key_columns: &[],
};
assert!(deliver_window(&mut sink, &mut state, &w).await.is_err());
assert_eq!(
state.get("t").snapshot_id,
None,
"a refused load must not move the cursor"
);
crate::testmatrix::emit(
crate::testmatrix::COMPONENT,
"table_sink::failed_load_holds_cursor",
state.get("t").snapshot_id.is_none(),
"cursor unchanged after a failed apply",
);
}
/// A sink that acks a snapshot other than the one it was asked to load would
/// silently skip or re-do a window. Refuse it.
#[tokio::test]
async fn a_mismatched_ack_is_refused() {
let dir = tempfile::tempdir().unwrap();
let mut state = SyncState::open(dir.path().join("state.json")).unwrap();
let (mut sink, _log) = spy(dir.path(), false);
sink.ack_snapshot = Some(9999);
let ups = [batch(vec![1])];
let w = TableWindow {
table: "t",
snapshots: SnapshotWindow::new(Some(7), 8),
upserts: &ups,
deletes: &[],
needs_full_reload: false,
sequence: None,
key_columns: &[],
};
let err = deliver_window(&mut sink, &mut state, &w)
.await
.expect_err("mismatched ack must be refused");
assert!(
err.to_string().contains("refusing to move the cursor"),
"{err}"
);
assert_eq!(state.get("t").snapshot_id, None);
}
/// **`needs_full_reload` refuses rather than reloading.** The sink is never
/// called, and the cursor does not move — so the next round sees the same
/// window and refuses again, loudly, until a human intervenes.
///
/// RED-when-broken: make `deliver_window` fall through and an overwrite
/// commit silently rewrites the whole destination table.
#[tokio::test]
async fn a_full_reload_window_is_refused_and_never_loaded() {
let dir = tempfile::tempdir().unwrap();
let mut state = SyncState::open(dir.path().join("state.json")).unwrap();
let (mut sink, log) = spy(dir.path(), false);
let ups = [batch(vec![1, 2])];
let w = TableWindow {
table: "bench.doris2g",
snapshots: SnapshotWindow::new(Some(1042), 1043),
upserts: &ups,
deletes: &[],
needs_full_reload: true,
sequence: None,
key_columns: &[],
};
let err = deliver_window(&mut sink, &mut state, &w)
.await
.expect_err("a full-reload window must be refused");
assert!(err.to_string().contains("FULL RELOAD"), "{err}");
assert!(
log.lock().unwrap().is_empty(),
"the sink must not be called at all"
);
assert_eq!(
state.get("bench.doris2g").snapshot_id,
None,
"cursor must not move on a refused window"
);
// The standalone pre-check, which is what a read path should call.
assert!(refuse_full_reload("t", false).is_ok());
assert!(refuse_full_reload("t", true).is_err());
}
/// Same window ⇒ same label, every time, with no state kept anywhere. This
/// is the entire replay-safety mechanism.
#[test]
fn label_is_derived_and_stable() {
let a = stream_load_label("fake", "personer2g", Some(1042), 1043, 0);
let b = stream_load_label("fake", "personer2g", Some(1042), 1043, 0);
assert_eq!(a, b);
assert_eq!(a, "skade-fake:personer2g-1042-1043-0");
assert_eq!(
stream_load_label("fake", "personer2g", None, 1043, 0),
"skade-fake:personer2g-init-1043-0"
);
}
/// **Collision-free across snapshot pairs and names.** `-` is the field
/// separator and is therefore banned inside the name, which is what stops
/// `("x", "y-1", 2, 3, 0)` from rendering the same string as
/// `("x", "y", 1, 2, 3)`.
///
/// RED-when-broken: let `-` through `sanitize_name` and this fails.
#[test]
fn labels_do_not_collide() {
let mut seen = HashSet::new();
let names = [
("fake", "personer2g"),
("fake", "personer2g-1"),
("fake-1", "personer2g"),
("fake", "personer.2g"),
("fake", "personer_2g"),
("f", "a"),
];
let mut n = 0;
for (db, t) in names {
for from in [None, Some(0), Some(1), Some(2), Some(1042)] {
for to in [1_i64, 2, 3, 1043] {
for chunk in 0..4 {
let l = stream_load_label(db, t, from, to, chunk);
assert!(
l.len() <= MAX_LABEL_LEN,
"label over {MAX_LABEL_LEN} chars: {l}"
);
assert!(
seen.insert(l.clone()),
"label collision on ({db}, {t}, {from:?}, {to}, {chunk}): {l}"
);
n += 1;
}
}
}
}
crate::testmatrix::emit(
crate::testmatrix::COMPONENT,
"table_sink::labels_collision_free",
seen.len() == n,
&format!("{n} distinct windows ⇒ {} distinct labels", seen.len()),
);
}
/// **Every generated label must satisfy Doris's own regex.** Measured
/// against the live cluster: a `.` in the label is rejected outright with
/// `[ANALYSIS_ERROR] ... Label format error. regex:
/// ^[\-_A-Za-z0-9:\p{L}]{1,128}$`, which is why database and table are
/// joined with `:` and every other character is folded to `_`.
///
/// RED-when-broken: put the `.` back and this fails on the first name that
/// contains one.
#[test]
fn every_label_satisfies_the_doris_regex() {
let nasty = [
("fake", "personer2g"),
("fake", "personer.2g"),
("my db", "tab/le"),
("db", "t\u{e4}bell"),
("db", "\u{4eba}"),
("db", &*"x".repeat(400)),
("", ""),
];
let mut n = 0;
for (db, t) in nasty {
for from in [None, Some(1_042)] {
for chunk in [0usize, 17] {
let l = stream_load_label(db, t, from, 1_043, chunk);
assert!(
is_valid_doris_label(&l),
"label rejected by Doris's regex: {l:?} (from {db:?}/{t:?})"
);
assert!(!l.contains('.'), "a dot is invalid in a Doris label: {l}");
assert!(l.len() <= MAX_LABEL_LEN, "over 128 chars: {l}");
n += 1;
}
}
}
crate::testmatrix::emit(
crate::testmatrix::COMPONENT,
"table_sink::label_matches_doris_regex",
true,
&format!("{n} labels, all matching ^[-_A-Za-z0-9:\\p{{L}}]{{1,128}}$"),
);
}
/// A name long enough to need truncation must stay unique — the hash
/// suffix, not the prefix, is what guarantees that.
#[test]
fn truncated_labels_stay_unique_and_in_bounds() {
let long_a = "a".repeat(400);
let long_b = format!("{}b", "a".repeat(399));
let la = stream_load_label("db", &long_a, Some(1), 2, 0);
let lb = stream_load_label("db", &long_b, Some(1), 2, 0);
assert!(la.len() <= MAX_LABEL_LEN && lb.len() <= MAX_LABEL_LEN);
assert_ne!(la, lb, "truncation must not reintroduce a collision");
// The snapshot fields survive truncation intact.
assert!(la.ends_with("-1-2-0"), "{la}");
}
}