crates/mimir-ingest/src/compact.rs
//! The lake's compaction: every uncompressed or small data file of a table
//! rewritten as zstd Parquet and merged, committed as ONE Iceberg `replace`
//! snapshot per table (the same rows; a changelog reads it as nothing).
//!
//! Why: the lake was written uncompressed (skade's default) one file per
//! release — 17 GB on dwarf for what zstd holds in ~1.7 GB, 52 of 85 edge
//! files under 1 MB (REPORT.md, 2026-10-16). The consumer now appends zstd;
//! this job brings the files written before to the same shape and keeps the
//! tail of small per-release files merged.
//!
//! The job is crash-safe at every step, and a reader never sees half a state:
//!
//! 1. the new files are written to storage, referenced by nothing (a crash
//! here leaves orphans a later run sweeps);
//! 2. ONE catalog commit swaps them in (`replace`), under the log's commit
//! lock so it never interleaves with a release — a crash before it leaves
//! the old snapshot, after it the new one, never a mix;
//! 3. the snapshots before the replace are expired (metadata only): only the
//! replace stays, and the cache stamp ([`EventLog::stamp`]) resolves to the
//! append it replaced by the parent id the replace carries;
//! 4. storage is swept: files no kept snapshot reads, older than `grace` —
//! a reader pinned before the swap keeps streaming the old files until it
//! is done, so the grace is longer than any scan.
use std::collections::{BTreeMap, HashSet};
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::sync::atomic::Ordering;
use std::time::{Duration, Instant, SystemTime};
use anyhow::{Context, Result, bail};
use skade::arrow_array::{ArrayRef, RecordBatch, new_null_array};
use skade::arrow_schema::SchemaRef;
use skade::iceberg::spec::{ManifestContentType, ManifestStatus};
use skade::iceberg::table::Table as IceTable;
use skade::parquet::basic::Compression;
use skade::parquet::file::reader::{FileReader, SerializedFileReader};
use skade::WriteProps;
use crate::delta::Kind;
use crate::log::{EventLog, InFlight, TABLES, on, sorted_by};
use crate::schema::{COL_REPO, UNIVERSE_REPO, qualified};
/// What [`EventLog::compact`] rewrites and how.
#[derive(Debug, Clone)]
pub struct CompactOptions {
/// zstd level of the rewritten files. Measured on the largest edge file
/// (5.4 M rows, 1.10 GB uncompressed): level 3 → 95 MB in 4.1 s, level
/// 6 → 87 MB in 7.8 s, level 9 → 85 MB in 14.7 s. The rewrite is written
/// once and read for the life of the lake: 6.
pub level: i32,
/// Rows per rewritten file (one group of consecutive releases).
pub target_rows: usize,
/// A zstd file with fewer rows than this is "small": merged with its
/// neighbours (of the same partition) when there are at least two small
/// files to merge. Capped at `target_rows / 2`, or every rewritten file
/// would be rewritten again on the next run.
pub small_rows: u64,
/// How old an unreferenced file must be before the sweep deletes it: a
/// reader pinned before the swap still streams the files it planned, so
/// longer than the longest scan. Beside a live writer the sweep uses at
/// least [`MIN_GRACE_BESIDE_WRITER`] whatever this says: the set of files
/// to keep is read before the directory is listed, and a manifest the
/// writer lands in between must not look like an orphan.
pub grace: Duration,
/// Snapshots a replace keeps (itself included). A tracked fold's delta
/// mark (mimir-room's kept room / ring view) sits a consume or two behind
/// the compaction; expiring below that mark makes its lineage unprovable
/// and every advance full-reloads. The files the window reads stay until
/// it moves past them.
pub keep_snapshots: usize,
/// Test hook: die after the new files are written, before the commit.
#[doc(hidden)]
pub fail_before_commit: bool,
}
/// The default kept window: eight snapshots of metadata cost nothing next to
/// a 140-200 s refold per consume.
pub const KEEP_SNAPSHOTS: usize = 8;
impl Default for CompactOptions {
fn default() -> Self {
Self { level: 6, target_rows: 4 << 20, small_rows: 2 << 20, grace: Duration::from_secs(1800), keep_snapshots: KEEP_SNAPSHOTS, fail_before_commit: false }
}
}
/// What one table's compaction did. `files_out == 0` and `snapshot == None`:
/// nothing was rewritten (the sweep may still have run).
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct CompactReport {
pub table: String,
/// Files rewritten (removed by the replace) and their bytes.
pub files_in: usize,
pub bytes_in: u64,
/// Files written in their place and their bytes.
pub files_out: usize,
pub bytes_out: u64,
/// Rows moved (verified equal on both sides before the commit).
pub rows: u64,
/// The replace snapshot, when one was committed.
pub snapshot: Option<i64>,
/// Snapshots expired after the replace.
pub expired: usize,
/// Files the sweep deleted and their bytes.
pub swept_files: usize,
pub swept_bytes: u64,
/// Ring rows dropped from a split lake's source table (they live in the
/// ring table): the commit was an overwrite, not a replace.
pub dropped_ring: u64,
/// This run partitioned the table by repo (a split lake's source table
/// written before the split).
pub partitioned: bool,
pub ms: u64,
}
impl std::fmt::Display for CompactReport {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
if self.snapshot.is_none() {
return write!(
f,
"{}: nothing to rewrite; swept {} file(s) ({} MB) in {} ms",
self.table,
self.swept_files,
self.swept_bytes >> 20,
self.ms
);
}
write!(
f,
"{}: {} file(s) ({} MB) → {} file(s) ({} MB), {} rows{}{}, {} snapshot {}, expired {}, swept {} file(s) ({} MB) in {} ms",
self.table,
self.files_in,
self.bytes_in >> 20,
self.files_out,
self.bytes_out >> 20,
self.rows,
if self.dropped_ring > 0 { format!(", {} ring rows dropped", self.dropped_ring) } else { String::new() },
if self.partitioned { ", partitioned by repo" } else { "" },
if self.dropped_ring > 0 { "overwrite" } else { "replace" },
self.snapshot.unwrap_or(0),
self.expired,
self.swept_files,
self.swept_bytes >> 20,
self.ms
)
}
}
/// The sweep's floor on the grace while a writer holds the log's seat.
pub const MIN_GRACE_BESIDE_WRITER: Duration = Duration::from_secs(2);
impl EventLog {
/// Compact every table of the lake ([`CompactOptions`]), one at a time.
/// Any reader may hold a pin meanwhile; the consumer may release
/// meanwhile (the commits serialize on the log's lock). An error on one
/// table stops the run — the tables before it are done, the table itself
/// is whole (a replace lands or it does not).
pub async fn compact(&self, opts: &CompactOptions) -> Result<Vec<CompactReport>> {
if self.is_pinned() {
bail!("a pinned read handle cannot compact the log");
}
let mut out = Vec::with_capacity(TABLES.len());
for name in TABLES {
out.push(self.compact_table(name, opts).await?);
}
Ok(out)
}
/// [`EventLog::compact`] for one table (unqualified name).
pub async fn compact_table(&self, name: &str, opts: &CompactOptions) -> Result<CompactReport> {
let shared = self.shared.clone();
let (name, opts) = (name.to_string(), opts.clone());
on(self.shared.rt(), async move { compact_one(shared, &name, &opts).await }).await
}
}
async fn compact_one(shared: Arc<crate::log::Shared>, name: &str, opts: &CompactOptions) -> Result<CompactReport> {
let t0 = Instant::now();
let mut report = CompactReport { table: name.to_string(), ..Default::default() };
let q = qualified(name);
let mut t = shared.wh.table(&q).await.with_context(|| format!("compact: load {q}"))?;
// never write or delete under a directory that is not this lake's: a
// catalog copied from another lake keeps that lake's locations
if let Some(root) = &shared.root {
let loc = local(t.inner().metadata().location());
let loc = std::fs::canonicalize(&loc).unwrap_or(loc);
if !loc.starts_with(root) {
bail!("compact {q}: the table's location {} is outside the lake {} — nothing touched", loc.display(), root.display());
}
}
// a source table of a split lake is written per repo and holds no ring
// row: the compaction brings a table written before the split there
let split_source = shared.ring_split.load(Ordering::Acquire) && Kind::SOURCE.iter().any(|k| EventLog::table_name(*k) == name);
if split_source && !partitioned_by(t.inner(), COL_REPO) {
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.partition_by(&[COL_REPO]).await.with_context(|| format!("compact: partition {q} by repo"))?;
report.partitioned = true;
}
let ice: IceTable = t.inner().clone();
let live = live_files(&ice).await?;
let selection = select(&live, opts, split_source);
if !selection.is_empty() {
let schema = skade::arrow_schema_of(&ice)?;
let sort_col: Option<String> = sorted_by(&ice).first().map(|c| c.to_string());
let sort_order_id = sort_col.as_ref().map(|_| t.inner().metadata().default_sort_order_id() as i32);
let props = WriteProps::new(Compression::ZSTD(
skade::parquet::basic::ZstdLevel::try_new(opts.level).context("compact: zstd level")?,
))
.row_group_size(256 * 1024);
t.set_write_props(props);
let rows_in: u64 = selection.iter().map(|f| f.rows).sum();
let mut staged: Option<skade::StagedRewrite> = None;
let mut expect = rows_in;
if partitioned_by(&ice, COL_REPO) {
// per repo: the files already written for it, plus the rows of
// the files written before the partition (one filtered pass per
// repo); the ring's rows are dropped on the way
let mut by_repo: BTreeMap<String, Vec<Live>> = BTreeMap::new();
let mut mixed: Vec<Live> = Vec::new();
for f in &selection {
match &f.partition {
Some(r) => by_repo.entry(r.clone()).or_default().push(f.clone()),
None => mixed.push(f.clone()),
}
}
let mut mixed_counts: BTreeMap<String, u64> = BTreeMap::new();
for f in &mixed {
for (r, n) in repo_counts(&f.path)? {
*mixed_counts.entry(r).or_default() += n;
}
}
let dropped: u64 = mixed_counts.remove(UNIVERSE_REPO).unwrap_or(0);
expect = rows_in - dropped;
report.dropped_ring = dropped;
let repos: std::collections::BTreeSet<String> = by_repo.keys().cloned().chain(mixed_counts.keys().cloned()).collect();
for r in repos {
let mut batches = Vec::new();
for f in by_repo.get(&r).into_iter().flatten() {
batches.extend(read_conformed(&f.path, &schema, None).with_context(|| format!("compact: read {}", f.path))?);
}
if mixed_counts.contains_key(&r) {
for f in &mixed {
batches.extend(read_conformed(&f.path, &schema, Some(&r)).with_context(|| format!("compact: read {} for {r}", f.path))?);
}
}
let mut groups = Vec::new();
for chunk in chunk_rows(batches, opts.target_rows) {
let merged = merge(&schema, chunk, sort_col.as_deref())?;
if !merged.is_empty() {
groups.push(merged);
}
}
if groups.is_empty() {
continue;
}
let part = t.write_rewrite_files(groups, sort_order_id).await.with_context(|| format!("compact: write {q} for {r}"))?;
match &mut staged {
Some(s) => s.extend(part),
None => staged = Some(part),
}
}
} else {
let mut groups: Vec<Vec<RecordBatch>> = Vec::new();
for group in group_by_rows(&selection, opts.target_rows) {
let mut batches = Vec::new();
for f in &group {
batches.extend(read_conformed(&f.path, &schema, None).with_context(|| format!("compact: read {}", f.path))?);
}
let merged = merge(&schema, batches, sort_col.as_deref())?;
if !merged.is_empty() {
groups.push(merged);
}
}
if !groups.is_empty() {
staged = Some(t.write_rewrite_files(groups, sort_order_id).await.context("compact: write the rewritten files")?);
}
}
let staged = match staged {
Some(s) => s,
None => t.write_rewrite_files(Vec::new(), sort_order_id).await?,
};
if staged.rows != expect {
// the staged files are orphans; a later sweep takes them
bail!("compact {q}: expected {expect} rows from {} file(s) ({rows_in} read, {} ring rows dropped), wrote {} — nothing committed", selection.len(), report.dropped_ring, staged.rows);
}
if opts.fail_before_commit {
bail!("compact {q}: fail_before_commit (test hook) — {} file(s) staged, nothing committed", staged.paths().len());
}
report.files_in = selection.len();
report.bytes_in = selection.iter().map(|f| f.bytes).sum();
report.files_out = staged.paths().len();
report.bytes_out = staged.bytes();
report.rows = staged.rows;
let remove: Vec<String> = selection.iter().map(|f| f.path.clone()).collect();
// 2. + 3. the swap and the expiry: two catalog commits, each atomic,
// never beside a release
{
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);
if report.dropped_ring > 0 {
t.commit_overwrite(&remove, staged).await.with_context(|| format!("compact: overwrite commit on {q}"))?;
} else {
t.commit_rewrite(&remove, staged).await.with_context(|| format!("compact: replace commit on {q}"))?;
}
}
report.snapshot = t.inner().metadata().current_snapshot().map(|s| s.snapshot_id());
shared.release_base.store(shared.wh.catalog().pointer_generation(), Ordering::Release);
let _in_flight = InFlight::begin(&shared.commits);
report.expired = t.expire_snapshots_keep(opts.keep_snapshots.max(1)).await.with_context(|| format!("compact: expire snapshots of {q}"))?.len();
}
}
// 4. storage: what no kept snapshot reads, past the grace
let fresh = shared.wh.table(&q).await?;
let grace = if shared.seat.load(Ordering::Acquire) { opts.grace.max(MIN_GRACE_BESIDE_WRITER) } else { opts.grace };
let (swept_files, swept_bytes) = sweep(fresh.inner(), grace).await?;
report.swept_files = swept_files;
report.swept_bytes = swept_bytes;
report.ms = t0.elapsed().as_millis() as u64;
Ok(report)
}
/// Is `col` the table's (one) identity partition column?
fn partitioned_by(t: &IceTable, col: &str) -> bool {
let md = t.metadata();
let spec = md.default_partition_spec();
spec.fields().len() == 1 && md.current_schema().name_by_field_id(spec.fields()[0].source_id) == Some(col)
}
/// Batches split into runs of about `target_rows` rows (a file each).
fn chunk_rows(batches: Vec<RecordBatch>, target_rows: usize) -> Vec<Vec<RecordBatch>> {
let mut out: Vec<Vec<RecordBatch>> = Vec::new();
let (mut cur, mut rows) = (Vec::new(), 0usize);
for b in batches {
if !cur.is_empty() && rows + b.num_rows() > target_rows {
out.push(std::mem::take(&mut cur));
rows = 0;
}
rows += b.num_rows();
cur.push(b);
}
if !cur.is_empty() {
out.push(cur);
}
out
}
/// One live data file of a table's current snapshot.
#[derive(Debug, Clone)]
struct Live {
path: String,
rows: u64,
bytes: u64,
/// The data sequence number it was added at (release order).
seq: i64,
zstd: bool,
/// Its partition value (the repo) when it was written under a `repo`
/// partition; `None` for a file of an unpartitioned spec.
partition: Option<String>,
/// Written under a spec that is not the table's default (a file written
/// before the table was partitioned).
old_spec: bool,
/// Holds ring rows (its `repo` lower bound is `@universe`).
ring: bool,
}
/// The data files of the current snapshot: path, rows, bytes, order, codec.
async fn live_files(t: &IceTable) -> Result<Vec<Live>> {
use skade::iceberg::spec::{Literal, PrimitiveLiteral};
let meta = t.metadata();
let Some(s) = meta.current_snapshot() else { return Ok(Vec::new()) };
let default_spec = meta.default_partition_spec_id();
let repo_field = meta.current_schema().field_id_by_name(COL_REPO);
let mut out = Vec::new();
for mf in t.manifest_list_reader(s).load().await?.entries() {
if mf.content != ManifestContentType::Data {
continue;
}
for e in t.manifest_reader().read(mf).await?.entries() {
if e.status() == ManifestStatus::Deleted {
continue;
}
let df = e.data_file();
let path = e.file_path().to_string();
let zstd = is_zstd(&local(&path)).unwrap_or(true);
let partition = if mf.partition_spec_id == default_spec {
df.partition().fields().first().and_then(|f| match f {
Some(Literal::Primitive(PrimitiveLiteral::String(v))) => Some(v.clone()),
_ => None,
})
} else {
None
};
let ring = repo_field
.and_then(|id| df.lower_bounds().get(&id))
.is_some_and(|d| matches!(d.literal(), PrimitiveLiteral::String(v) if v == UNIVERSE_REPO));
out.push(Live {
path,
rows: df.record_count(),
bytes: df.file_size_in_bytes(),
seq: e.sequence_number().unwrap_or(mf.sequence_number),
zstd,
partition,
old_spec: mf.partition_spec_id != default_spec,
ring,
});
}
}
out.sort_by(|a, b| (a.seq, &a.path).cmp(&(b.seq, &b.path)));
Ok(out)
}
/// The files to rewrite: every one not zstd, and the small zstd ones when
/// there are at least two small files to merge; on a split lake's source
/// table also every file of an old spec or holding ring rows. Idempotent: a
/// run over its own output finds every file zstd, in the default spec,
/// without ring rows, and at most one small one.
fn select(live: &[Live], opts: &CompactOptions, split_source: bool) -> Vec<Live> {
let small_rows = opts.small_rows.min(opts.target_rows as u64 / 2);
// small files merge with their neighbours IN THE SAME PARTITION: per
// partition value (one bucket for every file of an unpartitioned table),
// the small zstd files and the plain ones that will be rewritten anyway
let mut company: BTreeMap<Option<&str>, usize> = BTreeMap::new();
for f in live.iter().filter(|f| f.rows > 0 && (!f.zstd || f.rows < small_rows)) {
*company.entry(f.partition.as_deref()).or_default() += 1;
}
live.iter()
.filter(|f| f.rows > 0)
.filter(|f| {
!f.zstd
|| (f.rows < small_rows && company.get(&f.partition.as_deref()).copied().unwrap_or(0) >= 2)
|| (split_source && (f.old_spec || f.ring))
})
.cloned()
.collect()
}
/// Consecutive files up to `target_rows` per group (release order kept).
fn group_by_rows(files: &[Live], target_rows: usize) -> Vec<Vec<Live>> {
let mut groups: Vec<Vec<Live>> = Vec::new();
let mut cur: Vec<Live> = Vec::new();
let mut rows = 0usize;
for f in files {
if !cur.is_empty() && rows + f.rows as usize > target_rows {
groups.push(std::mem::take(&mut cur));
rows = 0;
}
rows += f.rows as usize;
cur.push(f.clone());
}
if !cur.is_empty() {
groups.push(cur);
}
groups
}
fn local(path: &str) -> PathBuf {
PathBuf::from(path.strip_prefix("file://").unwrap_or(path))
}
/// Is the file's first column chunk zstd? (`None`: unreadable or empty.)
fn is_zstd(path: &Path) -> Option<bool> {
let r = SerializedFileReader::new(std::fs::File::open(path).ok()?).ok()?;
let md = r.metadata();
if md.num_row_groups() == 0 || md.row_group(0).num_columns() == 0 {
return None;
}
Some(matches!(md.row_group(0).column(0).compression(), Compression::ZSTD(_)))
}
/// Rows per `repo` value of a data file (the `repo` column alone is read).
fn repo_counts(path: &str) -> Result<BTreeMap<String, u64>> {
use skade::arrow_array::{Array, StringArray};
use skade::parquet::arrow::ProjectionMask;
use skade::parquet::arrow::arrow_reader::ParquetRecordBatchReaderBuilder;
let f = std::fs::File::open(local(path))?;
let builder = ParquetRecordBatchReaderBuilder::try_new(f)?;
let idx = builder.parquet_schema().columns().iter().position(|c| c.name() == COL_REPO).context("compact: no `repo` column")?;
let mask = ProjectionMask::leaves(builder.parquet_schema(), [idx]);
let reader = builder.with_projection(mask).with_batch_size(64 * 1024).build()?;
let mut out: BTreeMap<String, u64> = BTreeMap::new();
for b in reader {
let b = b?;
let c = skade::arrow_cast::cast(b.column(0), &skade::arrow_schema::DataType::Utf8)?;
let c = c.as_any().downcast_ref::<StringArray>().context("repo utf8")?;
for i in 0..c.len() {
*out.entry(c.value(i).to_string()).or_default() += 1;
}
}
Ok(out)
}
/// A data file's rows in the TABLE's column order and types: a column the
/// file predates (added in place) comes back null, a narrower string or
/// timestamp type is cast. `only_repo`: the rows of that repo alone (a
/// parquet row filter on the `repo` column: the other columns are decoded
/// for the selected rows only).
fn read_conformed(path: &str, schema: &SchemaRef, only_repo: Option<&str>) -> Result<Vec<RecordBatch>> {
use skade::arrow_array::StringArray;
use skade::parquet::arrow::ProjectionMask;
use skade::parquet::arrow::arrow_reader::{ArrowPredicateFn, ParquetRecordBatchReaderBuilder, RowFilter};
let f = std::fs::File::open(local(path))?;
let mut builder = ParquetRecordBatchReaderBuilder::try_new(f)?.with_batch_size(64 * 1024);
if let Some(repo) = only_repo {
let idx = builder.parquet_schema().columns().iter().position(|c| c.name() == COL_REPO).context("compact: no `repo` column")?;
let mask = ProjectionMask::leaves(builder.parquet_schema(), [idx]);
let want = repo.to_string();
let pred = ArrowPredicateFn::new(mask, move |b: RecordBatch| {
let c = skade::arrow_cast::cast(b.column(0), &skade::arrow_schema::DataType::Utf8)?;
arrow_ord::cmp::eq(&c, &StringArray::new_scalar(want.as_str()))
});
builder = builder.with_row_filter(RowFilter::new(vec![Box::new(pred)]));
}
let reader = builder.build()?;
let mut out = Vec::new();
for b in reader {
let b = b?;
if b.num_rows() == 0 {
continue;
}
let mut cols: Vec<ArrayRef> = Vec::with_capacity(schema.fields().len());
for field in schema.fields() {
let col = match b.column_by_name(field.name()) {
Some(c) if c.data_type() == field.data_type() => c.clone(),
Some(c) => skade::arrow_cast::cast(c, field.data_type()).with_context(|| format!("compact: column `{}` {:?} → {:?}", field.name(), c.data_type(), field.data_type()))?,
None => new_null_array(field.data_type(), b.num_rows()),
};
cols.push(col);
}
out.push(RecordBatch::try_new(schema.clone(), cols)?);
}
Ok(out)
}
/// One group's batches as one file's row groups (256 k rows each), sorted by
/// `sort_col` when the table declares one (a merge of sorted files is not
/// sorted; the data file says `sort_order_id` only when its rows are).
fn merge(schema: &SchemaRef, batches: Vec<RecordBatch>, sort_col: Option<&str>) -> Result<Vec<RecordBatch>> {
let all = arrow_select::concat::concat_batches(schema, &batches).context("compact: concat")?;
if all.num_rows() == 0 {
return Ok(Vec::new());
}
let all = match sort_col {
Some(c) => {
let col = all.column_by_name(c).with_context(|| format!("compact: sort column `{c}`"))?;
let idx = arrow_ord::sort::sort_to_indices(col, None, None)?;
arrow_select::take::take_record_batch(&all, &idx)?
}
None => all,
};
const CHUNK: usize = 256 * 1024;
Ok((0..all.num_rows()).step_by(CHUNK).map(|at| all.slice(at, CHUNK.min(all.num_rows() - at))).collect())
}
/// Delete the files under the table's `data/` and `metadata/*.avro` that no
/// kept snapshot reads (a `Deleted` manifest entry is not a read), when they
/// are older than `grace`. Returns (files, bytes).
async fn sweep(t: &IceTable, grace: Duration) -> Result<(usize, u64)> {
let meta = t.metadata();
let mut keep: HashSet<PathBuf> = HashSet::new();
for s in meta.snapshots() {
keep.insert(local(s.manifest_list()));
let list = match t.manifest_list_reader(s).load().await {
Ok(l) => l,
Err(_) => continue,
};
for mf in list.entries() {
keep.insert(local(&mf.manifest_path));
let Ok(m) = t.manifest_reader().read(mf).await else { continue };
for e in m.entries() {
if e.status() != ManifestStatus::Deleted {
keep.insert(local(e.file_path()));
}
}
}
}
let root = local(meta.location());
let now = SystemTime::now();
let mut files = 0usize;
let mut bytes = 0u64;
for (dir, ext) in [("data", "parquet"), ("metadata", "avro")] {
let Ok(rd) = std::fs::read_dir(root.join(dir)) else { continue };
let mut candidates: BTreeMap<PathBuf, u64> = BTreeMap::new();
for entry in rd.flatten() {
let p = entry.path();
if p.extension().is_none_or(|x| x != ext) || keep.contains(&p) {
continue;
}
let Ok(md) = entry.metadata() else { continue };
let old_enough = md.modified().ok().and_then(|m| now.duration_since(m).ok()).is_some_and(|age| age >= grace);
if old_enough {
candidates.insert(p, md.len());
}
}
for (p, len) in candidates {
if std::fs::remove_file(&p).is_ok() {
files += 1;
bytes += len;
}
}
}
Ok((files, bytes))
}
#[cfg(test)]
mod tests {
use super::*;
fn live(path: &str, rows: u64, bytes: u64, seq: i64, zstd: bool) -> Live {
Live { path: path.into(), rows, bytes, seq, zstd, partition: None, old_spec: false, ring: false }
}
#[test]
fn plain_files_are_always_rewritten_and_small_zstd_ones_only_in_company() {
let opts = CompactOptions { target_rows: 200, small_rows: 100, ..Default::default() };
// one small zstd file alone: nothing to merge it with
assert!(select(&[live("a", 5, 10, 1, true)], &opts, false).is_empty());
// a big zstd file stays, a plain one goes
let got = select(&[live("big", 150, 1000, 1, true), live("plain", 5, 10, 2, false)], &opts, false);
assert_eq!(got.iter().map(|f| f.path.as_str()).collect::<Vec<_>>(), vec!["plain"]);
// two small zstd files merge
let got = select(&[live("s1", 5, 10, 1, true), live("s2", 5, 10, 2, true)], &opts, false);
assert_eq!(got.len(), 2);
// an empty file is never read
assert!(select(&[live("empty", 0, 10, 1, false)], &opts, false).is_empty());
// idempotent: the output of a run (zstd files of target_rows, plus one
// tail below small_rows) is not selected again
let out = [live("o1", 200, 50, 1, true), live("o2", 200, 50, 2, true), live("tail", 40, 9, 3, true)];
assert!(select(&out, &opts, false).is_empty(), "{:?}", select(&out, &opts, false));
// small_rows is capped below half the target, whatever the caller says
let loose = CompactOptions { target_rows: 200, small_rows: 1000, ..Default::default() };
assert!(select(&out, &loose, false).is_empty());
// a split lake's source table: a file of an old spec, or one holding
// ring rows, is rewritten however big and zstd it is — and never
// selected on a table that is not a split source table
let before = [live("old", 200, 50, 1, true)];
let mut old = before.clone();
old[0].old_spec = true;
assert_eq!(select(&old, &opts, true).len(), 1);
assert!(select(&old, &opts, false).is_empty());
let mut ring = before.clone();
ring[0].ring = true;
assert_eq!(select(&ring, &opts, true).len(), 1);
assert!(select(&ring, &opts, false).is_empty());
// a partitioned table: small files merge within a partition only —
// one small file per repo is what a run leaves, and stays
let mut per_repo = [live("ra", 40, 9, 1, true), live("rb", 40, 9, 2, true), live("rc", 40, 9, 3, true)];
for f in per_repo.iter_mut() {
f.partition = Some(f.path.clone());
}
assert!(select(&per_repo, &opts, true).is_empty(), "{:?}", select(&per_repo, &opts, true));
let mut two_in_ra = per_repo.to_vec();
two_in_ra.push(Live { path: "ra2".into(), partition: Some("ra".into()), ..live("ra2", 40, 9, 4, true) });
assert_eq!(select(&two_in_ra, &opts, true).iter().map(|f| f.path.as_str()).collect::<Vec<_>>(), vec!["ra", "ra2"]);
}
#[test]
fn groups_keep_release_order_and_the_row_budget() {
let files: Vec<Live> = (1..=5).map(|i| live(&format!("f{i}"), 10, 1, i, false)).collect();
let groups = group_by_rows(&files, 25);
let names: Vec<Vec<&str>> = groups.iter().map(|g| g.iter().map(|f| f.path.as_str()).collect()).collect();
assert_eq!(names, vec![vec!["f1", "f2"], vec!["f3", "f4"], vec!["f5"]]);
// a file bigger than the budget is its own group
let groups = group_by_rows(&[live("x", 100, 1, 1, false), live("y", 1, 1, 2, false)], 25);
assert_eq!(groups.len(), 2);
}
}