//! **How landed rows are HISTORIZED** — the business key a collector declares, the
//! system columns that let a re-read change one row instead of five hundred, and an
//! honest matrix of what merge/upsert each backend natively supports.
//!
//! # The question, and the vocabulary
//!
//! Three words get used for one idea and they are not the same thing:
//!
//! * **`MERGE INTO`** is the **SQL statement** (`MERGE INTO t USING s ON … WHEN
//! MATCHED THEN UPDATE … WHEN NOT MATCHED THEN INSERT …`). It is a piece of syntax an
//! engine parses.
//! * **upsert** is the **operation** — update the row if the key is there, insert it if
//! it is not. `MERGE INTO` is one way to spell it; `INSERT … ON CONFLICT` is another.
//! * **copy-on-write** and **merge-on-read** are the two **Iceberg mechanisms** that
//! implement it. Copy-on-write rewrites every data file that contained a changed row;
//! merge-on-read writes a small *delete file* beside the data and resolves it at scan
//! time. Both end at the same table; they trade write cost against read cost.
//!
//! # Where the upsert happens: THE BACKEND, natively
//!
//! The owner's call, 2026-09-03: *"the implementation will use the backends support for
//! this"*. So there is no portable trick in this crate. The leaf **declares the
//! business key**; korp's warehouse layer performs the merge with each backend's own
//! native support, and a backend that cannot REFUSES BY NAME rather than appending
//! twice and calling it a merge. The verb lives in `korp::warehouse::Upsert`
//! ([`UpsertKind`], `merge`), one arm per impl.
//!
//! Measured on oden 2026-09-03, not assumed:
//!
//! | backend | how IT does it |
//! |---|---|
//! | skade (embedded) | Iceberg **merge-on-read**: `skade::Table::delete_equality` (an equality-delete snapshot) then `append`. **Both already existed in skade** — no new Iceberg writer, no vendor patch. This is the backend active on this box |
//! | skade (remote) | refuses — it cannot even `land` (`korp-warehouse-25`) |
//! | iceberg-rest | refuses — it holds a RAW `iceberg::Table`, not skade's wrapper, so it does not inherit `delete_equality`. Lifting that primitive is a skade change |
//! | spark41 + Iceberg 1.11.0 | `MERGE INTO`, performed by Spark through the Iceberg extensions the running 4.1.2 image registers in its own `spark-defaults.conf` |
//! | spark42 | refuses — the running 4.2.0 image carries **no Iceberg jar at all** |
//!
//! Two things that are easy to assume and are **wrong**:
//!
//! * **Spark 4.2 has no `AUTO CDC` SQL.** Every jar in the running `apache/spark:4.2.0`
//! image was scanned: `AUTO CDC`, `APPLY CHANGES` and `SEQUENCE BY` appear nowhere as
//! keywords. What exists is `AUTO_CDC_FLOW_DETAILS` in `spark-connect_2.13-4.2.0.jar`
//! — a `PipelineCommand.DefineFlow` oneof branch. AUTO CDC in 4.2 is a
//! **Declarative-Pipelines Connect verb**, reachable only through `define_graph`, not
//! a statement you can send through `adhoc_sql`.
//! * **`iceberg` 0.10.1 does not help.** Checked against crates.io: 0.10.1 adds
//! `expire_snapshots` and `update_schema` to `src/transaction/` and STILL has no
//! overwrite / rewrite / delete action. `equality_delete_writer.rs` is still there and
//! still uncommittable through `Transaction`. A version bump buys nothing here.
//!
//! # What THIS crate contributes: the key, and the history columns
//!
//! [`BusinessKey`] is the declaration the merge consumes — a collector says what
//! identifies a row, defaulted from the [`join_keys`](crate::CollectorMeta::join_keys)
//! it already publishes, and a collector with none says `append_only` rather than
//! having a key guessed for it.
//!
//! The system columns below are the **companion**, not a rival mechanism:
//!
//! | column | what it holds |
//! |---|---|
//! | [`KEY_COL`] | the business key's value, the declared columns joined |
//! | [`HASH_COL`] | a fingerprint of the row's DATA columns — what makes "did this row change?" a comparison and not a guess |
//! | [`VALID_FROM_COL`] | when this version of the row was read (RFC-3339 UTC) |
//! | [`RUN_COL`] | the freshness mark of the run that landed it, so a row traces to the slice it came from |
//!
//! [`HASH_COL`] earns its place under a native merge too: [`suppress_unchanged`] drops
//! the rows a re-read brought back **byte-identical**, so a merge that would have
//! rewritten 500 rows to change one rewrites one. On the live polisen feed that is
//! measured at 499 rows suppressed out of 500.
//!
//! [`latest_per_key_sql`] remains for the backends whose answer is `AppendOnly` — it is
//! how their tables are READ, not a second way of writing them.
//!
//! # The case that actually needs it
//!
//! With a checkpoint alone (see [`crate::freshness`]) "only new rows" appends cleanly:
//! new keys, new rows, no duplicates. But the owner's second case — *same file, the date
//! or a flag changed, so re-read it* — brings **the same key back changed**. A plain
//! append then holds that key twice with no way to tell which is current. That is the
//! case [`HASH_COL`] and [`latest_per_key_sql`] answer, and it is the red test:
//! re-read a changed source and the CURRENT view must hold one row per key, not two,
//! with the old value still reachable.
//!
//! [`suppress_unchanged`] closes the other half: a re-read that brings back a row whose
//! fingerprint has not changed appends **nothing**, so a source re-read ten times does
//! not grow the table ten-fold. That is the same defect as `polisen_events` going
//! 500 → 1 504 by re-landing, seen from the landing side instead of the fetching side.
//!
//! # The key is DECLARED, and derived from what is already published
//!
//! Same discipline as [`crate::requires`] and [`crate::freshness`]:
//! [`Collector::business_key`](crate::Collector::business_key) defaults to
//! [`BusinessKey::derived`] over
//! [`CollectorMeta::join_keys`](crate::CollectorMeta::join_keys) — the field every
//! collector already publishes as "the entity-resolution spine". A collector with no
//! join keys derives [`BusinessKey::append_only`], which says plainly that its rows
//! cannot be matched across runs, instead of inventing a key that would silently
//! collapse distinct rows into one.
use std::collections::BTreeMap;
use std::sync::Arc;
use anyhow::Result;
use arrow_array::{Array, RecordBatch, StringArray};
use arrow_schema::{DataType, Field, Schema as ArrowSchema};
use serde::{Deserialize, Serialize};
// ─────────────────────────────────────────────────────────────────────────────
// The system columns
// ─────────────────────────────────────────────────────────────────────────────
/// The business key's value for a row — the declared columns joined by [`KEY_SEP`].
pub const KEY_COL: &str = "_korp_key";
/// A fingerprint of the row's DATA columns (every column that is not a system column).
/// Two rows with the same [`KEY_COL`] and the same `HASH_COL` are the same version.
pub const HASH_COL: &str = "_korp_hash";
/// When this version of the row was read — RFC-3339 UTC. The history axis.
pub const VALID_FROM_COL: &str = "_korp_valid_from";
/// The freshness mark of the run that landed this row, so a row traces to its slice.
pub const RUN_COL: &str = "_korp_run";
/// The separator between the parts of a composite key. `US` (unit separator, `0x1F`)
/// because it cannot occur in the text of an open-data field, so `("ab","c")` and
/// `("a","bc")` can never collide into one key — which a `-` or a `|` would allow.
pub const KEY_SEP: char = '\u{1f}';
/// Every system column, in the order they are appended.
pub const SYSTEM_COLS: &[&str] = &[KEY_COL, HASH_COL, VALID_FROM_COL, RUN_COL];
/// Is `name` one of the system columns? Data columns are everything else, and the
/// fingerprint is taken over data columns only — hashing the timestamp would make every
/// row differ from itself.
pub fn is_system_col(name: &str) -> bool {
SYSTEM_COLS.contains(&name)
}
// ─────────────────────────────────────────────────────────────────────────────
// The declaration
// ─────────────────────────────────────────────────────────────────────────────
/// **What identifies a row across runs.** Declared by the collector, defaulted from
/// its published join keys, and never guessed.
#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct BusinessKey {
/// The columns that identify a row. **Empty = this collector is append-only**: its
/// rows cannot be matched across runs, and that is a declaration, not a gap.
pub columns: Vec<String>,
/// One line an operator can read: what the key is, or why there is none.
pub why: String,
}
impl BusinessKey {
/// A declared key over these columns.
pub fn of(columns: &[&str], why: impl Into<String>) -> Self {
BusinessKey {
columns: columns.iter().map(|s| s.to_string()).collect(),
why: why.into(),
}
}
/// **This collector's rows cannot be matched across runs** — the honest answer, and
/// what a source of unkeyed observations (a raw sensor tick, a positional sample)
/// really is. A UI shows "append-only"; nothing pretends there is a current state.
pub fn append_only(why: impl Into<String>) -> Self {
BusinessKey {
columns: Vec::new(),
why: why.into(),
}
}
/// **The default derived from what every collector already publishes** — its
/// [`join_keys`](crate::CollectorMeta::join_keys), the entity-resolution spine.
///
/// A collector that overrides nothing still answers, and answers from its own
/// words. No join keys derives [`append_only`](BusinessKey::append_only) rather
/// than picking a column: a wrong key silently collapses distinct rows, which is a
/// data loss nobody sees, while append-only is merely verbose.
pub fn derived(join_keys: &[String]) -> Self {
if join_keys.is_empty() {
return BusinessKey::append_only(
"declares no join keys, so its rows cannot be matched across runs — \
every run appends",
);
}
BusinessKey {
columns: join_keys.to_vec(),
why: format!(
"derived from the declared join keys ({}) — override `business_key()` \
if the real identity is narrower",
join_keys.join(", ")
),
}
}
/// **[`derived`](BusinessKey::derived), checked against the columns the collector
/// actually emits.**
///
/// Deriving a key from `join_keys` conflated two different things and it was wrong
/// for **29 of 53** collectors when this was measured (2026-09-04). `join_keys` is
/// an entity-resolution SPINE — the axes a table joins to others on, written in the
/// vocabulary of the domain (`coordinate`, `elområde`, `REF_AREA`) — while a
/// business key must name columns that exist in the landed rows (`lat`/`lon`,
/// `in_domain`, `geo`). `trafikverket-roads` derived `["train_id", "coordinate",
/// "timestamp"]` over a schema of road elements.
///
/// **A partly-present key becomes append-only, never the surviving subset.** That
/// is the whole safety property: narrowing `["train_id", "coordinate", "timestamp"]`
/// to the one column that exists gives `["timestamp"]`, which collapses every road
/// element sharing an instant into one row — a silent data loss, and strictly worse
/// than the honest append. A declaration that does not survive contact with the
/// schema was a declaration about something else.
///
/// `emitted = None` means the schema could not be resolved without config; that is
/// also append-only, and the `why` says so rather than pretending.
pub fn derived_over(join_keys: &[String], emitted: Option<&[String]>) -> Self {
if join_keys.is_empty() {
return BusinessKey::derived(join_keys);
}
let Some(emitted) = emitted else {
return BusinessKey::append_only(format!(
"declares join keys ({}) but its schema needs config to resolve, so none \
of them is known to be a column it emits — and a key that is not \
emitted matches nothing",
join_keys.join(", ")
));
};
let missing: Vec<&str> = join_keys
.iter()
.filter(|k| !emitted.iter().any(|e| e == *k))
.map(String::as_str)
.collect();
if missing.is_empty() {
return BusinessKey::derived(join_keys);
}
BusinessKey::append_only(format!(
"its join keys ({}) are a JOIN SPINE, not a promise of columns: {} {} absent \
from the rows it lands ({}). A partly-present key is not an identity — it \
would collapse distinct rows — so this stays append-only until the \
collector declares a real `business_key()`",
join_keys.join(", "),
missing.join(", "),
if missing.len() == 1 { "is" } else { "are" },
emitted.join(", "),
))
}
/// Does this collector have a key at all?
pub fn is_keyed(&self) -> bool {
!self.columns.is_empty()
}
/// The one-line label a pane shows.
pub fn label(&self) -> String {
if self.is_keyed() {
format!("keyed on {}", self.columns.join(" + "))
} else {
format!("append-only — {}", self.why)
}
}
/// **Which declared key columns are missing from `schema`?** A key naming a column
/// the collector does not emit cannot identify anything, and the failure would only
/// show up as rows that never match. Checked before the first append, never after.
pub fn missing_from(&self, schema: &ArrowSchema) -> Vec<String> {
self.columns
.iter()
.filter(|c| schema.field_with_name(c).is_err())
.cloned()
.collect()
}
}
// ─────────────────────────────────────────────────────────────────────────────
// The fingerprint
// ─────────────────────────────────────────────────────────────────────────────
/// The text form of one cell, for keying and fingerprinting.
///
/// A null renders as the empty string and a null MARKER is folded into the hash
/// separately (see [`row_hash`]), so a genuinely-empty string and a null do not
/// fingerprint the same. Types this crate never emits render through their `Debug`,
/// which is stable within a build and is honest about being a fallback.
fn cell_text(col: &dyn Array, row: usize) -> Option<String> {
if col.is_null(row) {
return None;
}
use arrow_array::cast::AsArray;
use arrow_array::types::{Float64Type, Int32Type, Int64Type};
Some(match col.data_type() {
DataType::Utf8 => col.as_string::<i32>().value(row).to_string(),
DataType::LargeUtf8 => col.as_string::<i64>().value(row).to_string(),
DataType::Int64 => col.as_primitive::<Int64Type>().value(row).to_string(),
DataType::Int32 | DataType::Date32 => {
col.as_primitive::<Int32Type>().value(row).to_string()
}
DataType::Float64 => col.as_primitive::<Float64Type>().value(row).to_string(),
DataType::Boolean => col.as_boolean().value(row).to_string(),
DataType::Binary => {
let b = col.as_binary::<i32>().value(row);
// The bytes' own fingerprint, not their text — a WKB blob has no text form
// and rendering it would be both huge and lossy.
format!("{:016x}", fnv1a(b))
}
_ => format!("{:?}", col.slice(row, 1)),
})
}
/// FNV-1a, 64-bit.
///
/// A **content fingerprint**, not a security hash: it answers "did this row change",
/// and nothing here defends against an adversary choosing rows to collide. It is used
/// rather than blake3 because `history` is unconditional (it lives under `parsers`,
/// like [`crate::requires`]) while `blake3` rides on the `daemon` feature, and a leaf
/// does not grow a dependency to spell one `u64`. If this ever guards something an
/// attacker can write, it must become blake3 and that is a deliberate change, not a
/// silent upgrade.
fn fnv1a(bytes: &[u8]) -> u64 {
let mut h: u64 = 0xcbf2_9ce4_8422_2325;
for b in bytes {
h ^= *b as u64;
h = h.wrapping_mul(0x0000_0100_0000_01b3);
}
h
}
/// **One cell's text as the key/fingerprint sees it** — exported so a HOST computing
/// the same identity (korp's skade merge, comparing incoming keys to held ones) goes
/// through this function and not a second rendering of its own. Two renderings of
/// "what is this cell's identity" that disagree would split one key into two.
pub fn cell_key(col: &dyn Array, row: usize) -> String {
cell_text(col, row).unwrap_or_default()
}
/// The **business key value** of one row — the declared columns' text, joined by
/// [`KEY_SEP`], in the declared order. A row missing a key column's value contributes
/// an empty part, which is a real key value (many open-data rows have a blank field)
/// and not an error.
pub fn row_key(batch: &RecordBatch, row: usize, key: &BusinessKey) -> String {
key.columns
.iter()
.map(|c| {
batch
.column_by_name(c)
.and_then(|a| cell_text(a.as_ref(), row))
.unwrap_or_default()
})
.collect::<Vec<_>>()
.join(&KEY_SEP.to_string())
}
/// The **fingerprint of one row's DATA columns** — every column that is not a system
/// column, in schema order, with the column name folded in so moving a value between
/// two columns changes the hash.
pub fn row_hash(batch: &RecordBatch, row: usize) -> String {
let mut buf: Vec<u8> = Vec::with_capacity(256);
for (i, f) in batch.schema().fields().iter().enumerate() {
if is_system_col(f.name()) {
continue;
}
buf.extend_from_slice(f.name().as_bytes());
buf.push(b'=');
match cell_text(batch.column(i).as_ref(), row) {
// A distinct marker, so a NULL and an empty string do not fingerprint the
// same — they mean different things in every source this crate reads.
None => buf.extend_from_slice(b"\x00NULL"),
Some(t) => buf.extend_from_slice(t.as_bytes()),
}
buf.push(KEY_SEP as u8);
}
format!("{:016x}", fnv1a(&buf))
}
// ─────────────────────────────────────────────────────────────────────────────
// Historizing a batch
// ─────────────────────────────────────────────────────────────────────────────
/// The schema with the four system columns appended — what
/// [`Collector::schema`](crate::Collector::schema) becomes on the landing side.
/// Appending an already-historized schema is a no-op, so the function is idempotent
/// and a double call cannot produce `_korp_key_1`.
pub fn historize_schema(schema: &ArrowSchema) -> ArrowSchema {
if schema.field_with_name(KEY_COL).is_ok() {
return schema.clone();
}
let mut fields: Vec<Arc<Field>> = schema.fields().iter().cloned().collect();
for name in SYSTEM_COLS {
fields.push(Arc::new(Field::new(*name, DataType::Utf8, false)));
}
ArrowSchema::new(fields)
}
/// **Add the system columns to a landed batch.**
///
/// `valid_from` is the run's RFC-3339 stamp and `run` its freshness mark — both the
/// SAME value for every row of one run, so a whole slice is one point on the history
/// axis and can be filtered as one.
///
/// An unkeyed ([`BusinessKey::append_only`]) collector still gets the columns, with an
/// empty [`KEY_COL`]: the history axis and the fingerprint are useful on their own, and
/// a schema that changed shape depending on the declaration would make the warehouse's
/// tables inconsistent with each other.
pub fn historize(
batch: &RecordBatch,
key: &BusinessKey,
valid_from: &str,
run: &str,
) -> Result<RecordBatch> {
if batch.schema().field_with_name(KEY_COL).is_ok() {
return Ok(batch.clone());
}
let n = batch.num_rows();
let keys: Vec<String> = (0..n).map(|r| row_key(batch, r, key)).collect();
let hashes: Vec<String> = (0..n).map(|r| row_hash(batch, r)).collect();
let schema = historize_schema(batch.schema().as_ref());
let mut cols: Vec<Arc<dyn Array>> = batch.columns().to_vec();
cols.push(Arc::new(StringArray::from(keys)));
cols.push(Arc::new(StringArray::from(hashes)));
cols.push(Arc::new(StringArray::from(vec![valid_from.to_string(); n])));
cols.push(Arc::new(StringArray::from(vec![run.to_string(); n])));
Ok(RecordBatch::try_new(Arc::new(schema), cols)?)
}
/// **What the current state of a table is, by key** — `key → fingerprint`. The host
/// reads this from the warehouse (or from the last land it made) and hands it to
/// [`suppress_unchanged`].
pub type KeyDigest = BTreeMap<String, String>;
/// The digest of an already-historized batch — the `(key, hash)` pairs it carries.
/// This is how a host builds a [`KeyDigest`] from what it just landed without going
/// back to the warehouse.
pub fn digest_of(batch: &RecordBatch) -> KeyDigest {
let (Some(k), Some(h)) = (
batch.column_by_name(KEY_COL),
batch.column_by_name(HASH_COL),
) else {
return KeyDigest::new();
};
let (Some(k), Some(h)) = (
k.as_any().downcast_ref::<StringArray>(),
h.as_any().downcast_ref::<StringArray>(),
) else {
return KeyDigest::new();
};
(0..batch.num_rows())
.map(|r| (k.value(r).to_string(), h.value(r).to_string()))
.collect()
}
/// What [`suppress_unchanged`] did — the honest accounting of a re-read.
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub struct Suppressed {
/// Rows that came back byte-identical to the version already held. Not appended.
pub unchanged: u64,
/// Rows whose key is held but whose fingerprint differs — a NEW VERSION of an
/// existing row. This is the owner's "the date or a flag changed" case.
pub changed: u64,
/// Rows whose key has never been seen.
pub new: u64,
}
impl Suppressed {
/// Rows that will actually be appended.
pub fn appended(&self) -> u64 {
self.changed + self.new
}
/// The one-line report a run logs.
pub fn line(&self) -> String {
format!(
"{} new + {} changed appended, {} unchanged suppressed",
self.new, self.changed, self.unchanged
)
}
}
/// **Drop the rows a re-read brought back unchanged.**
///
/// The half of idempotence a checkpoint cannot give: a checkpoint stops a run
/// re-FETCHING a slice, and this stops a run re-LANDING rows it already holds when the
/// slice has to be re-fetched anyway (a source that only offers a whole document, a
/// re-read forced by a changed flag, an overlapping window).
///
/// An **unkeyed** collector is passed through untouched and `Suppressed::new` counts
/// every row: with no key there is nothing to match against, and dropping rows on a
/// fingerprint alone would silently delete legitimately-identical observations — two
/// sensor ticks with the same value are two facts, not one.
pub fn suppress_unchanged(
batch: &RecordBatch,
key: &BusinessKey,
held: &KeyDigest,
) -> Result<(RecordBatch, Suppressed)> {
if !key.is_keyed() {
return Ok((
batch.clone(),
Suppressed {
new: batch.num_rows() as u64,
..Default::default()
},
));
}
let keys = batch
.column_by_name(KEY_COL)
.and_then(|a| a.as_any().downcast_ref::<StringArray>().cloned());
let hashes = batch
.column_by_name(HASH_COL)
.and_then(|a| a.as_any().downcast_ref::<StringArray>().cloned());
let (Some(keys), Some(hashes)) = (keys, hashes) else {
anyhow::bail!(
"suppress_unchanged needs a historized batch — call `historize()` first \
(no `{KEY_COL}`/`{HASH_COL}` column present)"
);
};
let mut acc = Suppressed::default();
let mut take: Vec<bool> = Vec::with_capacity(batch.num_rows());
for r in 0..batch.num_rows() {
match held.get(keys.value(r)) {
None => {
acc.new += 1;
take.push(true);
}
Some(h) if h == hashes.value(r) => {
acc.unchanged += 1;
take.push(false);
}
Some(_) => {
acc.changed += 1;
take.push(true);
}
}
}
if acc.unchanged == 0 {
return Ok((batch.clone(), acc));
}
let mask = arrow_array::BooleanArray::from(take);
let filtered = arrow_select::filter::filter_record_batch(batch, &mask)?;
Ok((filtered, acc))
}
/// **The current state of a historized table: one row per key, the newest version.**
///
/// The ONE writer of this semantic. Every reader — korp's Tables view, a Spark job, a
/// `korp collect --show` — goes through this string, so no two of them can disagree
/// about what "current" means. Rows are ordered by [`VALID_FROM_COL`] then
/// [`RUN_COL`], so two versions landed in the same second still order deterministically
/// by the slice they came from.
///
/// Portable ANSI SQL (a window function and a subquery) — no Iceberg extension, no
/// engine-specific syntax, so it runs on every backend including Spark 4.2 without
/// Iceberg.
pub fn latest_per_key_sql(table: &str) -> String {
format!(
"SELECT * EXCEPT (_korp_rn) FROM (\n \
SELECT *, ROW_NUMBER() OVER (\n \
PARTITION BY {KEY_COL} ORDER BY {VALID_FROM_COL} DESC, {RUN_COL} DESC\n \
) AS _korp_rn\n \
FROM {table}\n\
) WHERE _korp_rn = 1"
)
}
/// **Every version of one key, newest first** — the history read. The other half of
/// [`latest_per_key_sql`], and the query that proves an older value is still reachable.
pub fn history_of_key_sql(table: &str) -> String {
format!(
"SELECT * FROM {table} WHERE {KEY_COL} = ? \
ORDER BY {VALID_FROM_COL} DESC, {RUN_COL} DESC"
)
}
/// **Latest-per-key over an in-memory batch** — the same semantic as
/// [`latest_per_key_sql`], for a caller with the rows in hand (a test, the leaf's own
/// proof, a host with no SQL engine). Same ordering rule, and it is checked against the
/// SQL's wording by the module's tests so the two cannot drift.
pub fn latest_per_key(batch: &RecordBatch) -> Result<RecordBatch> {
let (Some(k), Some(v), Some(run)) = (
batch.column_by_name(KEY_COL),
batch.column_by_name(VALID_FROM_COL),
batch.column_by_name(RUN_COL),
) else {
anyhow::bail!("latest_per_key needs a historized batch — call `historize()` first");
};
let (Some(k), Some(v), Some(run)) = (
k.as_any().downcast_ref::<StringArray>(),
v.as_any().downcast_ref::<StringArray>(),
run.as_any().downcast_ref::<StringArray>(),
) else {
anyhow::bail!("the system columns must be Utf8");
};
// key → (valid_from, run, row) of the newest seen so far.
let mut best: BTreeMap<&str, (usize, &str, &str)> = BTreeMap::new();
for r in 0..batch.num_rows() {
let (vf, rn) = (v.value(r), run.value(r));
best.entry(k.value(r))
.and_modify(|cur| {
if (vf, rn) > (cur.1, cur.2) {
*cur = (r, vf, rn);
}
})
.or_insert((r, vf, rn));
}
let mut keep: Vec<usize> = best.values().map(|(r, _, _)| *r).collect();
keep.sort_unstable();
let mask = arrow_array::BooleanArray::from(
(0..batch.num_rows())
.map(|r| keep.binary_search(&r).is_ok())
.collect::<Vec<bool>>(),
);
Ok(arrow_select::filter::filter_record_batch(batch, &mask)?)
}
// ─────────────────────────────────────────────────────────────────────────────
// What each backend can actually do
// ─────────────────────────────────────────────────────────────────────────────
/// **What one warehouse backend can do about an existing row.** Measured on this tree
/// (see the module docs' table), not read off a vendor page.
#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum MergeSupport {
/// The backend can execute a real upsert (`MERGE INTO`) through korp **today**.
Merge,
/// The engine underneath CAN do it, but this arm does not surface it yet, so it
/// REFUSES BY NAME. A real, nameable gap — and NOT the same answer as "Iceberg
/// cannot do this", which is what an honest matrix must not let a reader conclude.
Reachable,
/// Append only. The current state comes from [`latest_per_key_sql`].
AppendOnly,
/// The backend refuses `land` itself — it cannot even append, so historization is
/// moot here.
CannotLand,
}
impl MergeSupport {
/// The stable machine word.
pub fn key(self) -> &'static str {
match self {
MergeSupport::Merge => "merge",
MergeSupport::Reachable => "reachable_not_wired",
MergeSupport::AppendOnly => "append_only",
MergeSupport::CannotLand => "cannot_land",
}
}
/// **Can a run land rows here at all?** The precondition every other question sits
/// on top of.
pub fn can_land(self) -> bool {
!matches!(self, MergeSupport::CannotLand)
}
/// **Does append-with-validity work here?** True wherever rows can land — which is
/// the whole argument for the design.
pub fn supports_append_validity(self) -> bool {
self.can_land()
}
}
/// One backend's row in the matrix: what it is, what it can do, and why.
#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct BackendRow {
/// The backend's name as korp's warehouse config spells it.
pub backend: &'static str,
/// What it can do about an existing row.
pub merge: MergeSupport,
/// The measured reason.
pub why: &'static str,
}
/// **The honest matrix.** A constant, so a doc, a pane and a test all read the same
/// answer, and so a claim about a backend has exactly one place to be corrected.
pub const BACKENDS: &[BackendRow] = &[
BackendRow {
backend: "skade-embedded",
merge: MergeSupport::Merge,
why: "Iceberg merge-on-read, WIRED: korp::warehouse::Upsert drives \
`skade::Table::delete_equality` (an equality-delete snapshot) then \
`append`. Both primitives already existed in skade; nothing new writes \
Iceberg. This is the backend active on oden",
},
BackendRow {
backend: "skade-remote",
merge: MergeSupport::CannotLand,
why: "the remote arm refuses `land` itself (korp-warehouse-25) — there is no \
write path to merge INTO",
},
BackendRow {
backend: "iceberg-rest",
merge: MergeSupport::Reachable,
why: "refuses by name today: it holds a RAW iceberg::Table through the REST \
catalog, not skade's wrapper, so it does not carry delete_equality. \
Lifting that primitive to a raw Table + dyn Catalog is a small skade \
refactor — reachable, not written",
},
BackendRow {
backend: "spark41-iceberg",
merge: MergeSupport::Merge,
why: "MERGE INTO, WIRED: korp::warehouse::spark41_impl::merge_sql builds one \
statement and Spark performs it through the Iceberg extensions the live \
4.1.2 image registers itself (iceberg-spark-runtime 1.11.0)",
},
BackendRow {
backend: "spark42-no-iceberg",
merge: MergeSupport::CannotLand,
why: "no Iceberg runtime: `land` already refuses (korp-warehouse-31), so there \
is no table format to merge into. Spark 4.2's AUTO CDC is a \
Declarative-Pipelines Connect verb (DefineFlow.auto_cdc_flow_details), \
measured — NOT SQL — and it would need a table format too",
},
];
/// What `backend` can do, or `None` when it is not one this matrix has measured. A
/// backend nobody has measured is NOT assumed to merge.
pub fn merge_support(backend: &str) -> Option<MergeSupport> {
BACKENDS
.iter()
.find(|r| r.backend == backend)
.map(|r| r.merge)
}
#[cfg(test)]
mod tests {
use super::*;
use arrow_array::{Int64Array, StringArray};
use arrow_schema::Field;
fn schema() -> Arc<ArrowSchema> {
Arc::new(ArrowSchema::new(vec![
Field::new("id", DataType::Int64, true),
Field::new("status", DataType::Utf8, true),
Field::new("note", DataType::Utf8, true),
]))
}
fn batch(rows: &[(i64, &str, Option<&str>)]) -> RecordBatch {
RecordBatch::try_new(
schema(),
vec![
Arc::new(Int64Array::from(rows.iter().map(|r| r.0).collect::<Vec<_>>())),
Arc::new(StringArray::from(
rows.iter().map(|r| r.1.to_string()).collect::<Vec<_>>(),
)),
Arc::new(StringArray::from(
rows.iter().map(|r| r.2.map(str::to_string)).collect::<Vec<_>>(),
)),
],
)
.unwrap()
}
fn key() -> BusinessKey {
BusinessKey::of(&["id"], "the source's own id")
}
// ── the declaration ──────────────────────────────────────────────────────
/// A collector with no join keys SAYS it is append-only rather than inventing a
/// key; one with join keys derives a real one. Both arms.
#[test]
fn the_business_key_is_derived_from_the_published_join_keys() {
let none = BusinessKey::derived(&[]);
assert!(!none.is_keyed(), "no join keys means no key — never a guess");
assert!(none.label().starts_with("append-only"), "{}", none.label());
assert!(none.why.contains("cannot be matched"));
let some = BusinessKey::derived(&["orgnummer".into(), "kommunkod".into()]);
assert!(some.is_keyed());
assert_eq!(some.columns, vec!["orgnummer", "kommunkod"]);
assert!(some.label().contains("orgnummer + kommunkod"), "{}", some.label());
assert_eq!(BusinessKey::default().is_keyed(), false, "the zero value is append-only");
}
/// A key naming a column the collector does not emit is caught at the declaration,
/// not as rows that mysteriously never match.
#[test]
fn a_key_column_absent_from_the_schema_is_named() {
let s = schema();
assert!(key().missing_from(&s).is_empty(), "`id` is emitted");
assert_eq!(
BusinessKey::of(&["id", "orgnummer"], "x").missing_from(&s),
vec!["orgnummer".to_string()]
);
}
// ── the fingerprint, red first ───────────────────────────────────────────
/// **RED:** the case the whole design exists for. The same key comes back with a
/// changed flag; a plain append holds it twice with no way to tell which is
/// current. **GREEN:** the fingerprint differs, the change is appended as a new
/// version, and latest-per-key returns ONE row — the new one — while the old value
/// is still physically there.
#[test]
fn a_changed_row_becomes_a_new_version_and_the_current_view_holds_one_row_per_key() {
let k = key();
let v1 = historize(&batch(&[(1, "open", None), (2, "open", None)]), &k, "2026-09-03T10:00:00Z", "run-1").unwrap();
// The same two ids, one of them with a changed flag.
let v2 = historize(&batch(&[(1, "open", None), (2, "closed", None)]), &k, "2026-09-03T11:00:00Z", "run-2").unwrap();
// RED — a plain append of both batches holds id=2 TWICE.
let both = arrow_select::concat::concat_batches(&v1.schema(), [&v1, &v2]).unwrap();
assert_eq!(both.num_rows(), 4, "the physical table really does hold both versions");
// GREEN — the current view is one row per key, and it is the NEW value.
let now = latest_per_key(&both).unwrap();
assert_eq!(now.num_rows(), 2, "one row per key, not two");
let status = now.column_by_name("status").unwrap().as_any().downcast_ref::<StringArray>().unwrap().iter().map(|s| s.unwrap().to_string()).collect::<Vec<_>>();
assert_eq!(status, vec!["open", "closed"], "the CURRENT value wins");
// …and the old value is still reachable through the history axis.
let old: Vec<String> = (0..both.num_rows())
.filter(|r| both.column_by_name(VALID_FROM_COL).unwrap().as_any().downcast_ref::<StringArray>().unwrap().value(*r) == "2026-09-03T10:00:00Z")
.map(|r| both.column_by_name("status").unwrap().as_any().downcast_ref::<StringArray>().unwrap().value(r).to_string())
.collect();
assert_eq!(old, vec!["open", "open"], "the previous version is still readable");
// The run that landed each version is recorded, so a row traces to its slice.
let runs = both.column_by_name(RUN_COL).unwrap().as_any().downcast_ref::<StringArray>().unwrap();
assert_eq!(runs.value(0), "run-1");
assert_eq!(runs.value(3), "run-2");
}
/// **A re-read that changed nothing must append nothing** — the landing-side half
/// of the `polisen_events` 500 → 1 504 defect.
#[test]
fn an_unchanged_re_read_appends_no_rows_at_all() {
let k = key();
let first = historize(&batch(&[(1, "open", None), (2, "open", None)]), &k, "t1", "run-1").unwrap();
let held = digest_of(&first);
assert_eq!(held.len(), 2);
// The identical source, read again.
let again = historize(&batch(&[(1, "open", None), (2, "open", None)]), &k, "t2", "run-2").unwrap();
let (out, acc) = suppress_unchanged(&again, &k, &held).unwrap();
assert_eq!(out.num_rows(), 0, "nothing is appended");
assert_eq!(acc.unchanged, 2);
assert_eq!(acc.appended(), 0);
assert!(acc.line().contains("2 unchanged suppressed"), "{}", acc.line());
// RED arm: WITHOUT the digest every row looks new — the guard is doing work.
let (all, none_held) = suppress_unchanged(&again, &k, &KeyDigest::new()).unwrap();
assert_eq!(all.num_rows(), 2, "with nothing held, everything is new");
assert_eq!(none_held.new, 2);
// …and a genuinely changed row is NOT suppressed.
let changed = historize(&batch(&[(1, "open", None), (2, "closed", None)]), &k, "t3", "run-3").unwrap();
let (out, acc) = suppress_unchanged(&changed, &k, &held).unwrap();
assert_eq!(out.num_rows(), 1, "only the changed row is appended");
assert_eq!((acc.unchanged, acc.changed, acc.new), (1, 1, 0));
let id = out.column_by_name("id").unwrap().as_any().downcast_ref::<Int64Array>().unwrap();
assert_eq!(id.value(0), 2, "and it is the row that changed");
}
/// A null and an empty string are different facts and must not fingerprint the
/// same — a source that starts sending `""` where it sent nothing HAS changed.
#[test]
fn a_null_and_an_empty_string_do_not_fingerprint_alike() {
let with_null = batch(&[(1, "open", None)]);
let with_empty = batch(&[(1, "open", Some(""))]);
assert_ne!(row_hash(&with_null, 0), row_hash(&with_empty, 0));
// …and two genuinely identical rows DO — the hash is not just always different.
assert_eq!(row_hash(&with_null, 0), row_hash(&batch(&[(1, "open", None)]), 0));
// A value moved between columns changes the hash (the name is folded in).
assert_ne!(
row_hash(&batch(&[(1, "a", Some("b"))]), 0),
row_hash(&batch(&[(1, "b", Some("a"))]), 0)
);
}
/// A composite key cannot be forged by moving a character across the separator.
#[test]
fn a_composite_key_cannot_collide_across_its_parts() {
let s = Arc::new(ArrowSchema::new(vec![
Field::new("a", DataType::Utf8, true),
Field::new("b", DataType::Utf8, true),
]));
let mk = |a: &str, b: &str| {
RecordBatch::try_new(
s.clone(),
vec![
Arc::new(StringArray::from(vec![a.to_string()])) as Arc<dyn Array>,
Arc::new(StringArray::from(vec![b.to_string()])),
],
)
.unwrap()
};
let k = BusinessKey::of(&["a", "b"], "composite");
assert_ne!(row_key(&mk("ab", "c"), 0, &k), row_key(&mk("a", "bc"), 0, &k));
}
/// An unkeyed collector is passed through: dropping identical rows on a
/// fingerprint alone would delete legitimately-repeated observations.
#[test]
fn an_unkeyed_collector_is_never_deduplicated() {
let k = BusinessKey::append_only("raw sensor ticks have no identity");
let b = historize(&batch(&[(1, "x", None), (1, "x", None)]), &k, "t", "r").unwrap();
// Both rows carry an EMPTY key — the schema is the same shape either way.
let keys = b.column_by_name(KEY_COL).unwrap().as_any().downcast_ref::<StringArray>().unwrap();
assert_eq!(keys.value(0), "");
// A digest that would have matched them changes nothing.
let mut held = KeyDigest::new();
held.insert(String::new(), row_hash(&batch(&[(1, "x", None)]), 0));
let (out, acc) = suppress_unchanged(&b, &k, &held).unwrap();
assert_eq!(out.num_rows(), 2, "two identical ticks are two facts");
assert_eq!(acc.new, 2);
}
/// Historizing is idempotent and never renames a column.
#[test]
fn historizing_twice_changes_nothing() {
let k = key();
let once = historize(&batch(&[(1, "open", None)]), &k, "t", "r").unwrap();
let twice = historize(&once, &k, "OTHER", "OTHER").unwrap();
assert_eq!(once.schema(), twice.schema());
assert_eq!(twice.num_rows(), 1);
assert_eq!(
twice.column_by_name(VALID_FROM_COL).unwrap().as_any().downcast_ref::<StringArray>().unwrap().value(0),
"t",
"the first stamp is not overwritten"
);
assert_eq!(once.schema().fields().len(), schema().fields().len() + 4);
for c in SYSTEM_COLS {
assert!(once.schema().field_with_name(c).is_ok(), "{c} is present");
assert!(is_system_col(c));
}
assert!(!is_system_col("id"));
}
/// A batch that was never historized is refused rather than silently passed —
/// suppressing on absent columns would append everything and look like it worked.
#[test]
fn suppress_refuses_a_batch_that_was_never_historized() {
let e = suppress_unchanged(&batch(&[(1, "x", None)]), &key(), &KeyDigest::new())
.expect_err("a raw batch must be refused");
assert!(e.to_string().contains("historize"), "{e}");
assert!(latest_per_key(&batch(&[(1, "x", None)])).is_err());
}
/// The in-memory latest-per-key and the SQL agree about the ordering rule — two
/// readers of "current" that disagreed would be the exact bug this avoids.
#[test]
fn the_sql_and_the_in_memory_reader_state_the_same_ordering() {
let sql = latest_per_key_sql("polisen_events");
assert!(sql.contains("PARTITION BY _korp_key"), "{sql}");
assert!(
sql.contains("ORDER BY _korp_valid_from DESC, _korp_run DESC"),
"the tie-break is the run: {sql}"
);
assert!(sql.contains("_korp_rn = 1"), "{sql}");
assert!(sql.contains("polisen_events"));
// The in-memory reader breaks the same tie the same way: same stamp, later run
// wins.
let k = key();
let a = historize(&batch(&[(1, "first", None)]), &k, "t", "run-1").unwrap();
let b = historize(&batch(&[(1, "second", None)]), &k, "t", "run-2").unwrap();
let both = arrow_select::concat::concat_batches(&a.schema(), [&a, &b]).unwrap();
let now = latest_per_key(&both).unwrap();
assert_eq!(now.num_rows(), 1);
assert_eq!(
now.column_by_name("status").unwrap().as_any().downcast_ref::<StringArray>().unwrap().value(0),
"second",
"the later RUN wins when the stamps tie"
);
let hist = history_of_key_sql("polisen_events");
assert!(hist.contains("_korp_key = ?") && hist.contains("ORDER BY _korp_valid_from DESC"), "{hist}");
}
/// The backend matrix is honest in both directions: some backends CAN merge and
/// some cannot, and an unmeasured backend is not assumed to.
#[test]
fn the_backend_matrix_discriminates_and_defaults_to_no() {
assert_eq!(merge_support("spark41-iceberg"), Some(MergeSupport::Merge));
assert_eq!(merge_support("spark42-no-iceberg"), Some(MergeSupport::CannotLand));
// skade merges natively — equality delete + append, both already in skade.
assert_eq!(merge_support("skade-embedded"), Some(MergeSupport::Merge));
// iceberg-rest COULD, and refuses by name until the primitive is lifted.
assert_eq!(merge_support("iceberg-rest"), Some(MergeSupport::Reachable));
assert_eq!(merge_support("something-new"), None, "unmeasured is not 'yes'");
for b in BACKENDS {
assert!(!b.why.is_empty(), "{} states its reason", b.backend);
}
// **The design's justification, as a count.** Exactly ONE of the five real
// backends can perform a MERGE INTO through korp today; two cannot even land.
let merging = BACKENDS.iter().filter(|b| b.merge == MergeSupport::Merge).count();
assert_eq!(merging, 2, "skade-embedded and spark41 merge natively today");
let landing: Vec<&str> = BACKENDS
.iter()
.filter(|b| b.merge.can_land())
.map(|b| b.backend)
.collect();
assert_eq!(landing.len(), 3, "three backends can land at all: {landing:?}");
// Every backend that can land can at least be READ latest-per-key — the
// fallback for the arms that refuse to merge.
for b in BACKENDS {
assert_eq!(
b.merge.supports_append_validity(),
b.merge.can_land(),
"{} — the read-side fallback needs nothing but a land",
b.backend
);
}
// And the matrix does not collapse into one answer.
let distinct: std::collections::BTreeSet<&str> =
BACKENDS.iter().map(|b| b.merge.key()).collect();
assert!(distinct.len() >= 3, "the matrix really discriminates: {distinct:?}");
}
}