skade-sql/tests/merge_cow_vs_mor.rs
//! Copy-on-write versus merge-on-read, measured on a real table.
//!
//! Two ways to change a row in an Iceberg table, both of which this tree can
//! now do:
//!
//! - **merge-on-read** (`skade::Table::delete_equality` + `append`): write an
//! equality-delete file naming the keys, then append the new rows. Cheap to
//! write. Every later read must load and apply every delete file.
//! - **copy-on-write** (`MERGE INTO` → `Transaction::overwrite_files`): rewrite
//! the data files that hold the matched keys, and commit the swap. Expensive
//! to write in proportion to the SIZE OF THOSE FILES, not to the number of
//! changed rows. Reads stay at full speed.
//!
//! Run it — it is `#[ignore]`d because it writes hundreds of MB and takes
//! minutes:
//!
//! ```text
//! cargo test --release --test merge_cow_vs_mor -- --ignored --nocapture
//! ```
//!
//! Every number this prints is wall-clock on the machine that ran it, on a
//! local filesystem. Date and host belong in whatever writes them down.
use std::sync::Arc;
use std::time::Instant;
use anyhow::Result;
use arrow_array::{Int64Array, RecordBatch};
use arrow_schema::{DataType, Field, Schema};
use skade::Warehouse;
use skade_sql::WarehouseSql;
const ROWS: i64 = 200_000;
const FILES: i64 = 10;
const ROUNDS: usize = 8;
const KEYS_PER_ROUND: i64 = 100;
fn schema() -> Schema {
Schema::new(vec![
Field::new("id", DataType::Int64, false),
Field::new("v", DataType::Int64, true),
])
}
fn batch(ids: &[i64], v: i64) -> Result<RecordBatch> {
Ok(RecordBatch::try_new(Arc::new(schema()), vec![
Arc::new(Int64Array::from(ids.to_vec())),
Arc::new(Int64Array::from(vec![v; ids.len()])),
])?)
}
/// `ROWS` rows in `FILES` equal appends → `FILES` data files.
async fn seed(dir: &std::path::Path) -> Result<Warehouse> {
let wh = skade::open(dir.join("lake")).await?;
let mut t = wh.create_table("t", &schema()).await?;
let per = ROWS / FILES;
for f in 0..FILES {
let ids: Vec<i64> = (f * per..(f + 1) * per).collect();
t.append(&[batch(&ids, 0)?]).await?;
}
Ok(wh)
}
/// Bytes under a directory tree.
fn du(p: &std::path::Path) -> u64 {
let mut total = 0;
let Ok(rd) = std::fs::read_dir(p) else {
return 0;
};
for e in rd.flatten() {
let Ok(md) = e.metadata() else { continue };
total += if md.is_dir() {
du(&e.path())
} else {
md.len()
};
}
total
}
/// The keys touched in round `r`: `KEYS_PER_ROUND` ids SPREAD across every data
/// file (the worst case for copy-on-write) or CONCENTRATED in one (the best).
fn keys(round: usize, spread: bool) -> Vec<i64> {
let per = ROWS / FILES;
if spread {
(0..KEYS_PER_ROUND)
.map(|k| (k % FILES) * per + (round as i64 * 37 + k) % per)
.collect()
} else {
(0..KEYS_PER_ROUND)
.map(|k| (round as i64 * 37 + k) % per)
.collect()
}
}
/// Full scan, timed.
async fn read_ms(wh: &Warehouse) -> Result<(u128, usize)> {
let t0 = Instant::now();
let batches = wh.sql("SELECT id, v FROM t").await?;
let n: usize = batches.iter().map(RecordBatch::num_rows).sum();
Ok((t0.elapsed().as_millis(), n))
}
#[tokio::test]
#[ignore = "writes hundreds of MB; run with --release --ignored --nocapture"]
async fn copy_on_write_versus_merge_on_read() -> Result<()> {
for spread in [true, false] {
let label = if spread { "SPREAD over all 10 files" } else { "CONCENTRATED in 1 file" };
println!(
"\n=== {ROWS} rows in {FILES} files · {ROUNDS} rounds × {KEYS_PER_ROUND} keys, {label} ==="
);
println!(
"{:<6} {:>10} {:>12} {:>12} {:>10} {:>12} {:>12}",
"round", "cow ms", "cow MB", "cow read ms", "mor ms", "mor MB", "mor read ms"
);
let cow_dir = tempfile::tempdir()?;
let mor_dir = tempfile::tempdir()?;
let cow = seed(cow_dir.path()).await?;
let mor = seed(mor_dir.path()).await?;
let cow_root = cow_dir.path().join("lake");
let mor_root = mor_dir.path().join("lake");
let mut cow_prev = du(&cow_root);
let mut mor_prev = du(&mor_root);
for round in 0..ROUNDS {
let ks = keys(round, spread);
// ── copy-on-write: one MERGE INTO ─────────────────────────────
let ctx = cow.session().await?;
ctx.register_batch("src", batch(&ks, round as i64 + 1)?)?;
let t0 = Instant::now();
let n = ctx
.sql("MERGE INTO t USING src ON t.id = src.id WHEN MATCHED THEN UPDATE SET v = src.v")
.await?
.collect()
.await?;
let cow_ms = t0.elapsed().as_millis();
let affected = n
.iter()
.find(|b| b.num_rows() > 0)
.and_then(|b| b.column_by_name("count").cloned())
.and_then(|c| {
c.as_any()
.downcast_ref::<arrow_array::UInt64Array>()
.map(|a| a.value(0))
})
.unwrap_or(0);
assert_eq!(affected as i64, KEYS_PER_ROUND, "every key must match");
let cow_now = du(&cow_root);
let cow_mb = (cow_now - cow_prev) as f64 / 1_048_576.0;
cow_prev = cow_now;
let (cow_read, cow_rows) = read_ms(&cow).await?;
assert_eq!(cow_rows as i64, ROWS);
// ── merge-on-read: equality delete + append ───────────────────
let t0 = Instant::now();
{
let mut t = mor.table("t").await?;
let key_batch = RecordBatch::try_new(
Arc::new(Schema::new(vec![Field::new("id", DataType::Int64, false)])),
vec![Arc::new(Int64Array::from(ks.clone()))],
)?;
t.delete_equality(&key_batch, &["id"]).await?;
t.append(&[batch(&ks, round as i64 + 1)?]).await?;
}
let mor_ms = t0.elapsed().as_millis();
let mor_now = du(&mor_root);
let mor_mb = (mor_now - mor_prev) as f64 / 1_048_576.0;
mor_prev = mor_now;
let (mor_read, mor_rows) = read_ms(&mor).await?;
assert_eq!(mor_rows as i64, ROWS, "merge-on-read must read back the same row count");
println!(
"{:<6} {:>10} {:>12.2} {:>12} {:>10} {:>12.2} {:>12}",
round + 1,
cow_ms,
cow_mb,
cow_read,
mor_ms,
mor_mb,
mor_read
);
}
println!(
"total on disk after {ROUNDS} rounds: cow {:.1} MB · mor {:.1} MB",
du(&cow_root) as f64 / 1_048_576.0,
du(&mor_root) as f64 / 1_048_576.0
);
}
Ok(())
}