knut-bifrost 0.3.0

//! **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}");
    }
}