crates/mimir-ingest/src/ring_split.rs
//! The one-shot migration of a legacy lake to the split shape: the ring's
//! rows (`repo = @universe`) copied from the source tables to the ring
//! tables, verified by count per owner crate, then the lake switched
//! (`mimir.ring = split`). The old rows stay in the source tables until the
//! next compaction drops them (`compact.rs`, which also partitions the source
//! tables by repo) — so until then the lake can be switched back by removing
//! the property, nothing having been deleted.
//!
//! Idempotent and crash-safe: a ring table that already holds every ring row
//! is skipped; one holding some of them (a crash mid-copy) is emptied and
//! copied again; the switch is one metadata commit at the very end.
use std::collections::BTreeMap;
use std::sync::Arc;
use std::sync::atomic::Ordering;
use std::time::Instant;
use anyhow::{Context, Result, bail};
use skade::ScanFilter;
use skade::arrow_array::RecordBatch;
use crate::delta::Kind;
use crate::log::{EventLog, InFlight, on};
use crate::schema::{COL_KRATE, COL_REPO, CODE_NODE_EVENTS, PROP_RING, RING_SPLIT, UNIVERSE_REPO, qualified};
/// What [`EventLog::split_ring`] does and how.
#[derive(Debug, Clone)]
pub struct SplitOptions {
/// Rows per append while copying (bounds memory: ~200 B a row).
pub chunk_rows: usize,
/// Test hook: die after the copy, before the verify and the switch.
#[doc(hidden)]
pub fail_after_copy: bool,
}
impl Default for SplitOptions {
fn default() -> Self {
Self { chunk_rows: 1 << 20, fail_after_copy: false }
}
}
/// What the split did. `copied == 0 && switched == false`: the lake was
/// split already.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct SplitReport {
/// Ring rows copied into the ring tables by this run.
pub copied: u64,
/// Ring rows the ring tables hold after the run (every kind).
pub verified: u64,
/// Ring tables this run emptied and copied again (a crash mid-copy before).
pub recopied: usize,
/// Did this run switch the lake?
pub switched: bool,
pub ms: u64,
}
impl std::fmt::Display for SplitReport {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(
f,
"ring split: copied {} row(s) (recopied {} table(s)), {} ring row(s) verified, {} in {} ms",
self.copied,
self.recopied,
self.verified,
if self.switched { "switched" } else { "not switched (was split already)" },
self.ms
)
}
}
impl EventLog {
/// Copy the ring out of the source tables, verify, switch (see the module).
pub async fn split_ring(&self, opts: &SplitOptions) -> Result<SplitReport> {
if self.is_pinned() {
bail!("a pinned read handle cannot split the log");
}
let log = self.clone();
let opts = opts.clone();
on(self.shared.rt(), async move { split(log, &opts).await }).await
}
}
async fn split(log: EventLog, opts: &SplitOptions) -> Result<SplitReport> {
let t0 = Instant::now();
let mut report = SplitReport::default();
if log.ring_split() {
report.ms = t0.elapsed().as_millis() as u64;
return Ok(report);
}
let shared = log.shared.clone();
// 1. copy, per kind
for kind in Kind::SOURCE {
let src_name = EventLog::table_name(kind);
let ring_name = EventLog::ring_table_name(kind).expect("a source kind has a ring table");
let src = shared.wh.table(&qualified(src_name)).await?;
let mut ring = shared.wh.table(&qualified(ring_name)).await?;
let expected = krate_counts(src.inner(), Some(UNIVERSE_REPO)).await?;
let have = krate_counts(ring.inner(), None).await?;
let (expected_rows, have_rows) = (total(&expected), total(&have));
if have_rows == expected_rows && have == expected {
continue;
}
if have_rows > 0 {
// a crash mid-copy: empty the ring table (one overwrite) and copy again
let live = ring.live_data_files().await?;
let empty = ring.write_rewrite_files(Vec::new(), None).await?;
{
let _commit = shared.commit_lock.lock().await;
shared.release_base.store(shared.wh.catalog().pointer_generation(), Ordering::Release);
let _in_flight = InFlight::begin(&shared.commits);
ring.commit_overwrite(&live, empty).await.with_context(|| format!("empty {ring_name} before copying again"))?;
}
report.recopied += 1;
}
ring.set_write_props(log.write_props().clone());
let cols = skade::arrow_schema_of(src.inner())?;
let cols: Vec<&str> = cols.fields().iter().map(|f| f.name().as_str()).collect();
let mut buf: Vec<RecordBatch> = Vec::new();
let mut buffered = 0usize;
let mut failed: Option<anyhow::Error> = None;
// the copy streams the source table and appends a chunk at a time
let chunks: Arc<std::sync::Mutex<Vec<Vec<RecordBatch>>>> = Arc::new(std::sync::Mutex::new(Vec::new()));
let sink = chunks.clone();
skade::read_filtered_each(src.inner(), Some(&ScanFilter::eq(COL_REPO, UNIVERSE_REPO)), &cols, |b| {
buffered += b.num_rows();
buf.push(b);
if buffered >= opts.chunk_rows {
sink.lock().unwrap_or_else(|e| e.into_inner()).push(std::mem::take(&mut buf));
buffered = 0;
}
true
})
.await?;
if let Some(e) = failed.take() {
return Err(e);
}
if !buf.is_empty() {
chunks.lock().unwrap_or_else(|e| e.into_inner()).push(buf);
}
let chunks = std::mem::take(&mut *chunks.lock().unwrap_or_else(|e| e.into_inner()));
for chunk in chunks {
let rows: usize = chunk.iter().map(|b| b.num_rows()).sum();
if rows == 0 {
continue;
}
let _commit = shared.commit_lock.lock().await;
shared.release_base.store(shared.wh.catalog().pointer_generation(), Ordering::Release);
let _in_flight = InFlight::begin(&shared.commits);
ring.append(&chunk).await.with_context(|| format!("copy {rows} ring rows into {ring_name}"))?;
report.copied += rows as u64;
}
}
if opts.fail_after_copy {
bail!("split_ring: fail_after_copy (test hook) — copied {} row(s), nothing switched", report.copied);
}
// 2. verify: the ring tables hold exactly the source tables' ring rows, per owner
for kind in Kind::SOURCE {
let src = shared.wh.table(&qualified(EventLog::table_name(kind))).await?;
let ring = shared.wh.table(&qualified(EventLog::ring_table_name(kind).expect("source kind"))).await?;
let expected = krate_counts(src.inner(), Some(UNIVERSE_REPO)).await?;
let have = krate_counts(ring.inner(), None).await?;
if expected != have {
bail!(
"split_ring: {} holds {} ring row(s) in {} owner(s), {} holds {} in {} — not switched",
EventLog::table_name(kind),
total(&expected),
expected.len(),
EventLog::ring_table_name(kind).expect("source kind"),
total(&have),
have.len()
);
}
report.verified += total(&have);
}
// 3. switch: one metadata commit, then every handle of this log reads split
{
let mut t = shared.wh.table(&qualified(CODE_NODE_EVENTS)).await?;
let _commit = shared.commit_lock.lock().await;
shared.release_base.store(shared.wh.catalog().pointer_generation(), Ordering::Release);
let _in_flight = InFlight::begin(&shared.commits);
t.set_properties([(PROP_RING, RING_SPLIT)]).await.context("switch the lake to split")?;
}
log.set_ring_split();
report.switched = true;
report.ms = t0.elapsed().as_millis() as u64;
Ok(report)
}
fn total(m: &BTreeMap<String, u64>) -> u64 {
m.values().sum()
}
/// Rows per owner crate of a table (of `repo` when given), streamed.
async fn krate_counts(t: &skade::iceberg::table::Table, repo: Option<&str>) -> Result<BTreeMap<String, u64>> {
use skade::arrow_array::{Array, StringArray};
let filter = repo.map(|r| ScanFilter::eq(COL_REPO, r));
let mut out: BTreeMap<String, u64> = BTreeMap::new();
let mut bad: Option<anyhow::Error> = None;
skade::read_filtered_each(t, filter.as_ref(), &[COL_REPO, COL_KRATE], |b| {
let step = (|| -> Result<()> {
let r = b.column_by_name(COL_REPO).context("repo")?;
let r = skade::arrow_cast::cast(r, &skade::arrow_schema::DataType::Utf8)?;
let r = r.as_any().downcast_ref::<StringArray>().context("repo utf8")?;
let k = b.column_by_name(COL_KRATE).context("krate")?;
let k = skade::arrow_cast::cast(k, &skade::arrow_schema::DataType::Utf8)?;
let k = k.as_any().downcast_ref::<StringArray>().context("krate utf8")?;
for i in 0..b.num_rows() {
if repo.is_some_and(|x| r.value(i) != x) {
continue;
}
*out.entry(k.value(i).to_string()).or_default() += 1;
}
Ok(())
})();
match step {
Ok(()) => true,
Err(e) => {
bad = Some(e);
false
}
}
})
.await?;
if let Some(e) = bad {
return Err(e);
}
Ok(out)
}