src/warehouse/maintenance.rs
//! Warehouse **maintenance** — small-file COMPACTION + snapshot EXPIRY.
//!
//! Every warehouse write `fast_append`s a tiny Parquet data file and a new
//! snapshot. Left alone they accrete forever: a scan reads thousands of Parquet
//! footers (slow) and disk grows unbounded. iceberg-rust 0.9 ships no
//! `rewrite_data_files` / `expire_snapshots`, so both maintenance ops are
//! implemented here at the metadata layer through the same `RedbCatalog`
//! `commit_table` primitive skade's equality-delete writer uses.
//!
//! ## COMPACTION — [`compact_table`] (the `nornir warehouse compact` op)
//!
//! Per (table, partition): read the partition's rows, **sort them by the table's
//! declared [`SortOrder`]** (the data-skipping key — tight, non-overlapping
//! per-file bounds), rewrite them into a few **large** Parquet files with the
//! warehouse zstd codec, and commit a **true REPLACE snapshot**: a fresh `Data`
//! manifest holding only the new large files, carried by a manifest list that
//! **drops every old data manifest** (non-data manifests — equality deletes —
//! are carried forward). The head snapshot then references only the new files;
//! the old small data files are unreferenced by `main` and reclaimed by a
//! later [`expire_snapshots`] once the pre-compaction history snapshots age out.
//!
//! This is a TRUE replace at the table head (not append-then-supersede): the new
//! `main` snapshot does not reference any old data file. iceberg-rust 0.9's
//! `Transaction` only does `fast_append` (it rejects a manifest list that drops
//! files), so the replace is built at the metadata layer and committed via
//! `RedbCatalog::commit_table` with an optimistic `RefSnapshotIdMatch`
//! requirement — the same wire pieces an Iceberg-REST `commit_table` carries.
//!
//! ## EXPIRY — [`expire_snapshots`] (the `nornir warehouse expire` op)
//!
//! Retain the last `keep_last` snapshots (by timestamp) **and** every snapshot
//! within `keep_days` days, always keeping the current head. Remove the rest
//! from the catalog (`TableUpdate::RemoveSnapshots`), then compute the **live
//! file set** = every data file + manifest + manifest-list referenced by any
//! RETAINED snapshot, and delete every on-disk data/metadata file NOT in that
//! set. The live-set is computed from the retained snapshots' manifest lists, so
//! a file is deleted **only** when no retained snapshot references it — the
//! never-delete-a-referenced-file safety invariant.
use std::collections::{HashMap, HashSet};
use std::path::{Path, PathBuf};
use anyhow::{Context, Result, anyhow};
use arrow::array::{Array, RecordBatch, StringArray};
use arrow::compute::concat_batches;
use iceberg::spec::{
DataContentType, DataFile, DataFileFormat, ManifestFile, ManifestListWriter,
ManifestWriterBuilder, Operation, Snapshot, SnapshotReference, SnapshotRetention, Struct,
Summary,
};
use iceberg::table::Table as IceTable;
use iceberg::writer::base_writer::data_file_writer::DataFileWriterBuilder;
use iceberg::writer::file_writer::ParquetWriterBuilder;
use iceberg::writer::file_writer::location_generator::{
DefaultFileNameGenerator, DefaultLocationGenerator,
};
use iceberg::writer::file_writer::rolling_writer::RollingFileWriterBuilder;
use iceberg::writer::{IcebergWriter, IcebergWriterBuilder};
use iceberg::{Catalog, TableIdent, TableRequirement, TableUpdate};
use skade::parquet::file::properties::WriterProperties;
use skade_katalog::RedbCatalog;
use super::iceberg::{IcebergWarehouse, warehouse_compression, warehouse_sort_cols};
// ---------------------------------------------------------------------------
// Reports — what the ops return (and what the JobRecord detail folds from).
// ---------------------------------------------------------------------------
/// Per-table outcome of a [`compact_table`] run.
#[derive(Debug, Clone, Default, serde::Serialize)]
pub struct CompactReport {
pub table: String,
/// Data files referenced by `main` BEFORE compaction.
pub files_before: usize,
/// Data files referenced by `main` AFTER compaction.
pub files_after: usize,
/// Rows rewritten (== rows in the table; row count is conserved).
pub rows: u64,
/// Partitions compacted (1 for an unpartitioned table).
pub partitions: usize,
/// `true` when nothing needed doing (no SortOrder, ≤1 file, or empty).
pub skipped: bool,
}
/// Per-table outcome of an [`expire_snapshots`] run.
#[derive(Debug, Clone, Default, serde::Serialize)]
pub struct ExpireReport {
pub table: String,
pub snapshots_before: usize,
pub snapshots_after: usize,
/// Data + manifest + manifest-list files deleted from disk.
pub files_deleted: usize,
/// Bytes reclaimed from disk.
pub bytes_reclaimed: u64,
}
/// Outcome of a [`reset`] run — what was on disk, and whether it was removed.
#[derive(Debug, Clone, Default, serde::Serialize)]
pub struct ResetReport {
/// Did `<root>/catalog.redb` exist?
pub had_catalog: bool,
/// Did `<root>/warehouse/` exist?
pub had_data: bool,
pub catalog_bytes: u64,
pub data_bytes: u64,
pub data_files: usize,
/// `false` when this was a dry run (`apply == false`) or there was nothing
/// to remove — the caller must NOT print a success line in that case.
pub removed: bool,
}
// ---------------------------------------------------------------------------
// RESET
// ---------------------------------------------------------------------------
/// Total size + file count of a directory tree (best-effort; unreadable entries
/// are skipped rather than failing the walk).
fn dir_size_and_files(root: &Path) -> (u64, usize) {
let mut bytes = 0u64;
let mut files = 0usize;
for e in walkdir::WalkDir::new(root).into_iter().flatten() {
if e.file_type().is_file() {
files += 1;
if let Ok(m) = e.metadata() {
bytes += m.len();
}
}
}
(bytes, files)
}
/// **EMPTY a warehouse** — remove BOTH `<root>/catalog.redb` and
/// `<root>/warehouse/`, so the next `IcebergWarehouse::open(root)` starts from a
/// clean, zero-table warehouse.
///
/// Both paths, always. The catalog is the index and `warehouse/` is the data,
/// and the two are only meaningful together: delete only `warehouse/` and the
/// surviving catalog still names data files that no longer exist (`open()` then
/// fails on a dangling reference); delete only `catalog.redb` and the Parquet is
/// orphaned on disk with nothing that can ever reference it again. Getting this
/// pair wrong by hand is what made "empty the warehouse and re-measure" — step 1
/// of the ingest loop — unrepeatable.
///
/// `apply == false` is a DRY RUN: the report describes what is there, nothing is
/// touched, and `removed` stays `false`. Nothing on disk ⇒ `removed` is `false`
/// too, so a caller can never print a success line over a no-op.
///
/// This is not [`expire_snapshots`]-style reclaim. It is total, intentional data
/// loss. Callers are expected to have taken the catalog lock check themselves
/// (see `nornir warehouse reset`) — a live single-writer must be stopped first.
pub fn reset(root: &Path, apply: bool) -> Result<ResetReport> {
let catalog = root.join("catalog.redb");
let data = root.join("warehouse");
let mut rep = ResetReport {
had_catalog: catalog.exists(),
had_data: data.exists(),
catalog_bytes: std::fs::metadata(&catalog).map(|m| m.len()).unwrap_or(0),
..Default::default()
};
if data.is_dir() {
let (b, f) = dir_size_and_files(&data);
rep.data_bytes = b;
rep.data_files = f;
}
if !apply || (!rep.had_catalog && !rep.had_data) {
return Ok(rep);
}
if rep.had_data {
std::fs::remove_dir_all(&data)
.with_context(|| format!("remove warehouse data dir {}", data.display()))?;
}
if rep.had_catalog {
std::fs::remove_file(&catalog)
.with_context(|| format!("remove catalog {}", catalog.display()))?;
}
rep.removed = true;
Ok(rep)
}
// ---------------------------------------------------------------------------
// COMPACTION
// ---------------------------------------------------------------------------
/// Compact every table in the warehouse that declares a [`SortOrder`] (the
/// data-skipping key), or just `only_table` when `Some`. Returns one
/// [`CompactReport`] per table touched (including skipped ones).
pub fn compact_all(
wh: &IcebergWarehouse,
only_table: Option<&str>,
only_repo: Option<&str>,
) -> Result<Vec<CompactReport>> {
let tables: Vec<String> = match only_table {
Some(t) => vec![t.to_string()],
None => wh
.table_names()?
.into_iter()
// Only the sorted, point-lookup tables carry a SortOrder worth
// compacting; the rest write unsorted event logs (no skip win).
.filter(|t| !warehouse_sort_cols(t).is_empty())
.collect(),
};
let mut out = Vec::with_capacity(tables.len());
for t in tables {
out.push(compact_table(wh, &t, only_repo)?);
}
Ok(out)
}
/// Compact one table: rewrite its small per-partition data files into a few
/// large SORTED Parquet files and commit a true REPLACE snapshot. `only_repo`
/// restricts to a single `repo` partition (no-op for the others).
pub fn compact_table(
wh: &IcebergWarehouse,
table: &str,
only_repo: Option<&str>,
) -> Result<CompactReport> {
let catalog = wh.catalog().clone();
let ident = wh.table_ident(table);
let sort_cols = warehouse_sort_cols(table);
wh.block_on(async move {
compact_table_async(catalog.as_ref(), &ident, table, sort_cols, only_repo).await
})
}
async fn compact_table_async(
catalog: &RedbCatalog,
ident: &TableIdent,
table: &str,
sort_cols: &[&str],
only_repo: Option<&str>,
) -> Result<CompactReport> {
let mut report = CompactReport {
table: table.to_string(),
skipped: true,
..Default::default()
};
if sort_cols.is_empty() {
// No declared sort key — nothing to gain from rewriting.
return Ok(report);
}
let tbl = catalog.load_table(ident).await?;
let metadata = tbl.metadata();
let Some(current) = metadata.current_snapshot() else {
return Ok(report); // empty table, no snapshot.
};
// ── enumerate the head snapshot's manifests, splitting data vs non-data ──
let manifest_list = tbl.manifest_list_reader(current).load().await?;
let mut data_manifests: Vec<ManifestFile> = Vec::new();
let mut nondata_manifests: Vec<ManifestFile> = Vec::new();
let mut live_data_files: Vec<DataFile> = Vec::new();
for mf in manifest_list.entries() {
let manifest = tbl.manifest_reader().read(mf).await?;
let mut has_data = false;
for entry in manifest.entries() {
if entry.content_type() == DataContentType::Data {
has_data = true;
live_data_files.push(entry.data_file().clone());
}
}
if has_data {
data_manifests.push(mf.clone());
} else {
nondata_manifests.push(mf.clone());
}
}
report.files_before = live_data_files.len();
// ── group the live rows by partition value (the `repo` identity col) ─────
let part_col = partition_column(metadata);
let batches = skade::read_all(&tbl).await?;
let total_rows: u64 = batches.iter().map(|b| b.num_rows() as u64).sum();
report.rows = total_rows;
if live_data_files.len() <= 1 || total_rows == 0 {
// Nothing to merge.
return Ok(report);
}
let groups = group_by_partition(&batches, part_col.as_deref(), only_repo)?;
if groups.is_empty() {
return Ok(report);
}
// ── write one large SORTED Parquet file per partition group ──────────────
let mut new_files: Vec<DataFile> = Vec::new();
let mut compacted_partitions: HashSet<Option<String>> = HashSet::new();
for (part_value, batch) in &groups {
let sorted = super::iceberg::sort_batch_by_pub(batch, sort_cols)?;
let files = write_one_file(&tbl, &sorted, part_value.as_deref()).await?;
new_files.extend(files);
compacted_partitions.insert(part_value.clone());
report.partitions += 1;
}
// When restricted to one repo, the un-touched partitions' OLD data files
// must be carried forward into the new head (they are still live). Re-add
// every old data file whose partition we did NOT compact.
let mut carried_old: Vec<DataFile> = Vec::new();
if only_repo.is_some() {
for df in &live_data_files {
let pv = data_file_partition_value(df, part_col.as_deref());
if !compacted_partitions.contains(&pv) {
carried_old.push(df.clone());
}
}
}
if new_files.is_empty() {
return Ok(report);
}
report.files_after = new_files.len() + carried_old.len();
report.skipped = false;
// ── build the REPLACE snapshot (new Data manifest, drop old data manifests) ──
commit_replace_snapshot(catalog, &tbl, new_files, carried_old, nondata_manifests).await?;
Ok(report)
}
/// Write `batch` to one (or, if it overflows the rolling target size, a few)
/// Parquet data file(s) via the iceberg-rust `DataFileWriter` — the SAME writer
/// path `skade::append_with` uses, so the resulting [`DataFile`]s carry the
/// per-column lower/upper bounds the manifest needs for file-skipping. Returns
/// the closed `DataFile`s (tagged with `part_value`'s partition key).
async fn write_one_file(
tbl: &IceTable,
batch: &RecordBatch,
part_value: Option<&str>,
) -> Result<Vec<DataFile>> {
let schema = tbl.metadata().current_schema().clone();
let data_location = format!("{}/data", tbl.metadata().location());
let location_gen = DefaultLocationGenerator::with_data_location(data_location);
let file_name_gen = DefaultFileNameGenerator::new(
unique_prefix(),
Some("compact".into()),
DataFileFormat::Parquet,
);
let props = WriterProperties::builder()
.set_compression(warehouse_compression())
.build();
let pw = ParquetWriterBuilder::new(props, schema.clone());
let rolling = RollingFileWriterBuilder::new_with_default_file_size(
pw,
tbl.file_io().clone(),
location_gen,
file_name_gen,
);
// Derive the partition key for this group (identity partition over `repo`).
let partition = match part_value {
Some(v) => Some(partition_key_for_value(tbl, v)?),
None => None,
};
let mut writer = DataFileWriterBuilder::new(rolling).build(partition).await?;
writer.write(batch.clone()).await?;
let files = writer.close().await?;
Ok(files)
}
/// Commit a `Replace` snapshot whose manifest list = a fresh Data manifest of
/// `new_files` (+ `carried_old`) + every carried-forward `nondata_manifests`.
/// The old data manifests are intentionally NOT carried, so the new head
/// references only the compacted files. Optimistic: requires `main` still points
/// at the snapshot we read.
async fn commit_replace_snapshot(
catalog: &RedbCatalog,
tbl: &IceTable,
new_files: Vec<DataFile>,
carried_old: Vec<DataFile>,
nondata_manifests: Vec<ManifestFile>,
) -> Result<()> {
let metadata = tbl.metadata();
let schema = metadata.current_schema().clone();
let parent = metadata.current_snapshot();
let parent_id = parent.map(|s| s.snapshot_id());
let new_seq = metadata.last_sequence_number() + 1;
let snapshot_id = fresh_snapshot_id(parent_id);
let location = metadata.location().to_string();
let file_io = tbl.file_io().clone();
// 1. a new Data manifest with the compacted (+ carried-old) files, Added.
let manifest_path = format!("{location}/metadata/compact-m-{snapshot_id}.avro");
let mut mw = ManifestWriterBuilder::new(
file_io.new_output(&manifest_path)?,
Some(snapshot_id),
schema.clone(),
metadata.default_partition_spec().as_ref().clone(),
)
.build_v2_data();
let mut added_rows: u64 = 0;
for df in new_files.into_iter().chain(carried_old.into_iter()) {
added_rows += df.record_count();
mw.add_file(df, new_seq)?;
}
let data_manifest: ManifestFile = mw.write_manifest_file().await?;
// 2. manifest list = the new data manifest + carried-forward non-data ones.
// (Old DATA manifests are dropped — that is the replace.)
let list_path = format!("{location}/metadata/snap-{snapshot_id}.avro");
let mut lw = ManifestListWriter::v2(
file_io.new_output(&list_path)?.writer().await?,
snapshot_id,
parent_id,
new_seq,
);
lw.add_manifests(std::iter::once(data_manifest))?;
lw.add_manifests(nondata_manifests.into_iter())?;
lw.close().await?;
// 3. the Replace snapshot + commit through the catalog (optimistic on main).
let snapshot = Snapshot::builder()
.with_snapshot_id(snapshot_id)
.with_parent_snapshot_id(parent_id)
.with_sequence_number(new_seq)
.with_timestamp_ms(now_ms())
.with_manifest_list(list_path)
.with_row_range(metadata.next_row_id(), added_rows)
.with_schema_id(schema.schema_id())
.with_summary(Summary {
operation: Operation::Replace,
additional_properties: HashMap::from([(
"nornir.compacted".to_string(),
"true".to_string(),
)]),
})
.build();
let requirements = vec![TableRequirement::RefSnapshotIdMatch {
r#ref: "main".into(),
snapshot_id: parent_id,
}];
let updates = vec![
TableUpdate::AddSnapshot { snapshot },
TableUpdate::SetSnapshotRef {
ref_name: "main".into(),
reference: SnapshotReference {
snapshot_id,
retention: SnapshotRetention::Branch {
min_snapshots_to_keep: None,
max_snapshot_age_ms: None,
max_ref_age_ms: None,
},
},
},
];
catalog
.commit_table(tbl.identifier().clone(), requirements, updates)
.await
.context("commit compaction replace snapshot")?;
Ok(())
}
// ---------------------------------------------------------------------------
// EXPIRY
// ---------------------------------------------------------------------------
/// Default number of most-recent snapshots to retain when `--keep-last` is unset.
pub const DEFAULT_KEEP_LAST: usize = 10;
/// Expire old snapshots across every table (or just `only_table`), retaining the
/// last `keep_last` snapshots and every snapshot within `keep_days` days, then
/// deleting the now-unreferenced data/manifest files from disk.
pub fn expire_all(
wh: &IcebergWarehouse,
only_table: Option<&str>,
keep_last: usize,
keep_days: Option<u64>,
) -> Result<Vec<ExpireReport>> {
let tables: Vec<String> = match only_table {
Some(t) => vec![t.to_string()],
None => wh.table_names()?,
};
let mut out = Vec::with_capacity(tables.len());
for t in tables {
out.push(expire_snapshots(wh, &t, keep_last, keep_days)?);
}
Ok(out)
}
/// Expire one table's old snapshots + GC its now-orphaned files.
pub fn expire_snapshots(
wh: &IcebergWarehouse,
table: &str,
keep_last: usize,
keep_days: Option<u64>,
) -> Result<ExpireReport> {
let catalog = wh.catalog().clone();
let ident = wh.table_ident(table);
let root = wh.root().to_path_buf();
let table = table.to_string();
wh.block_on(async move {
expire_snapshots_async(
catalog.as_ref(),
&ident,
&table,
&root,
keep_last,
keep_days,
)
.await
})
}
async fn expire_snapshots_async(
catalog: &RedbCatalog,
ident: &TableIdent,
table: &str,
root: &Path,
keep_last: usize,
keep_days: Option<u64>,
) -> Result<ExpireReport> {
let mut report = ExpireReport {
table: table.to_string(),
..Default::default()
};
let tbl = catalog.load_table(ident).await?;
let metadata = tbl.metadata();
// Snapshots newest-first.
// `SnapshotRef`s (not `&Snapshot`): the manifest-list reader takes one.
let mut snaps: Vec<&iceberg::spec::SnapshotRef> = metadata.snapshots().collect();
snaps.sort_by_key(|s| std::cmp::Reverse(s.timestamp_ms()));
report.snapshots_before = snaps.len();
if snaps.is_empty() {
return Ok(report);
}
let keep_last = keep_last.max(1); // always keep at least the current head.
let cutoff_ms = keep_days.map(|d| now_ms() - (d as i64) * 86_400_000);
let current_id = metadata.current_snapshot_id();
// RETAIN: the newest `keep_last`, anything newer than the day-cutoff, and the
// current head (never expire the snapshot `main` points at).
let mut retained_ids: HashSet<i64> = HashSet::new();
for (i, s) in snaps.iter().enumerate() {
let within_n = i < keep_last;
let within_days = cutoff_ms.map(|c| s.timestamp_ms() >= c).unwrap_or(false);
let is_current = Some(s.snapshot_id()) == current_id;
if within_n || within_days || is_current {
retained_ids.insert(s.snapshot_id());
}
}
let remove_ids: Vec<i64> = snaps
.iter()
.map(|s| s.snapshot_id())
.filter(|id| !retained_ids.contains(id))
.collect();
report.snapshots_after = retained_ids.len();
if remove_ids.is_empty() {
return Ok(report); // nothing aged out.
}
// ── compute the LIVE file set = files referenced by any RETAINED snapshot ──
// (data files, manifests, AND manifest lists). Anything else under
// data/ + metadata/ that we wrote is an orphan once the removal commits.
let mut live: HashSet<String> = HashSet::new();
for s in snaps
.iter()
.filter(|s| retained_ids.contains(&s.snapshot_id()))
{
live.insert(canon(s.manifest_list()));
let ml = tbl.manifest_list_reader(s).load().await?;
for mf in ml.entries() {
live.insert(canon(&mf.manifest_path));
let manifest = tbl.manifest_reader().read(mf).await?;
for entry in manifest.entries() {
live.insert(canon(entry.data_file().file_path()));
}
}
}
// Keep every metadata.json version pointer too (never GC the catalog's own
// table-metadata blobs — only data files, manifests, manifest lists).
// We scope deletion to data/ + the snap/manifest .avro files in metadata/.
// ── commit the snapshot removal through the catalog FIRST (atomic) ──
catalog
.commit_table(
ident.clone(),
Vec::new(),
vec![TableUpdate::RemoveSnapshots {
snapshot_ids: remove_ids.clone(),
}],
)
.await
.context("commit snapshot removal")?;
// ── only AFTER the catalog no longer references them, delete orphan files ──
let (deleted, bytes) = gc_orphans(root, table, &live)?;
report.files_deleted = deleted;
report.bytes_reclaimed = bytes;
Ok(report)
}
/// Delete every data file + `*.avro` manifest/manifest-list under the table's
/// `data/` and `metadata/` dirs that is NOT in `live`. `metadata/*.json`
/// (table-metadata version blobs) are never touched — only data + avro orphans.
fn gc_orphans(root: &Path, table: &str, live: &HashSet<String>) -> Result<(usize, u64)> {
let root = root.canonicalize().unwrap_or_else(|_| root.to_path_buf());
let table_dir = root.join("warehouse").join("nornir").join(table);
let mut deleted = 0usize;
let mut bytes = 0u64;
// data/ — every Parquet data file.
let mut candidates: Vec<PathBuf> = Vec::new();
collect_files(&table_dir.join("data"), &mut candidates, &["parquet"]);
// metadata/ — manifest + manifest-list avro only (NOT *.json metadata blobs).
collect_files(&table_dir.join("metadata"), &mut candidates, &["avro"]);
for path in candidates {
let key = canon(&path.to_string_lossy());
if live.contains(&key) {
continue; // still referenced by a retained snapshot — keep.
}
let size = std::fs::metadata(&path).map(|m| m.len()).unwrap_or(0);
std::fs::remove_file(&path).with_context(|| format!("delete orphan {}", path.display()))?;
deleted += 1;
bytes += size;
}
Ok((deleted, bytes))
}
// ---------------------------------------------------------------------------
// helpers
// ---------------------------------------------------------------------------
/// The single identity-partition column name (e.g. `repo`), or `None` for an
/// unpartitioned table. The warehouse only uses single-column identity specs.
fn partition_column(metadata: &iceberg::spec::TableMetadata) -> Option<String> {
let spec = metadata.default_partition_spec();
let f = spec.fields().first()?;
let schema = metadata.current_schema();
schema.field_by_id(f.source_id).map(|nf| nf.name.clone())
}
/// Group `batches` into one merged `RecordBatch` per distinct value of the
/// partition column. `None` partition col ⇒ one group, key `None`. `only_repo`
/// restricts to rows whose partition value matches (the others are dropped from
/// the rewrite and carried forward as old files by the caller).
fn group_by_partition(
batches: &[RecordBatch],
part_col: Option<&str>,
only_repo: Option<&str>,
) -> Result<Vec<(Option<String>, RecordBatch)>> {
if batches.is_empty() {
return Ok(Vec::new());
}
let schema = batches[0].schema();
let Some(col) = part_col else {
// Unpartitioned: one big group.
let merged = concat_batches(&schema, batches)?;
if merged.num_rows() == 0 {
return Ok(Vec::new());
}
return Ok(vec![(None, merged)]);
};
// Bucket row indices per partition value, then take() each bucket.
use arrow::compute::take;
let mut order: Vec<String> = Vec::new();
let mut buckets: HashMap<String, Vec<RecordBatch>> = HashMap::new();
for b in batches {
let arr = b
.column_by_name(col)
.ok_or_else(|| anyhow!("partition column `{col}` missing from batch"))?;
let sa = arr
.as_any()
.downcast_ref::<StringArray>()
.ok_or_else(|| anyhow!("partition column `{col}` is not Utf8"))?;
// Per-value row index lists.
let mut per_value: HashMap<String, Vec<u32>> = HashMap::new();
for i in 0..b.num_rows() {
let v = if sa.is_null(i) {
String::new()
} else {
sa.value(i).to_string()
};
if let Some(want) = only_repo {
if v != want {
continue;
}
}
per_value.entry(v).or_default().push(i as u32);
}
for (v, idxs) in per_value {
let idx = arrow::array::UInt32Array::from(idxs);
let cols: Vec<_> = b
.columns()
.iter()
.map(|c| take(c.as_ref(), &idx, None))
.collect::<std::result::Result<_, _>>()?;
let part_batch = RecordBatch::try_new(b.schema(), cols)?;
if !buckets.contains_key(&v) {
order.push(v.clone());
}
buckets.entry(v).or_default().push(part_batch);
}
}
let mut out = Vec::with_capacity(order.len());
for v in order {
let parts = buckets.remove(&v).unwrap_or_default();
if parts.is_empty() {
continue;
}
let merged = concat_batches(&schema, &parts)?;
if merged.num_rows() > 0 {
out.push((Some(v), merged));
}
}
Ok(out)
}
/// The string partition value of an existing `DataFile` for `part_col`, read from
/// its partition `Struct` (identity transform ⇒ the literal column value).
fn data_file_partition_value(df: &DataFile, part_col: Option<&str>) -> Option<String> {
part_col?;
let p: &Struct = df.partition();
// Single-column identity spec ⇒ the first (only) literal is the value.
p.iter().next().flatten().map(|lit| literal_to_string(lit))
}
fn literal_to_string(lit: &iceberg::spec::Literal) -> String {
use iceberg::spec::{Literal, PrimitiveLiteral};
match lit {
Literal::Primitive(PrimitiveLiteral::String(s)) => s.clone(),
other => format!("{other:?}"),
}
}
/// Build a single-column identity [`PartitionKey`] for `value` against `tbl`.
fn partition_key_for_value(tbl: &IceTable, value: &str) -> Result<iceberg::spec::PartitionKey> {
use iceberg::spec::{Literal, PartitionKey, Struct};
let spec = tbl.metadata().default_partition_spec().as_ref().clone();
let schema = tbl.metadata().current_schema().clone();
let lit = Some(Literal::string(value));
Ok(PartitionKey::new(
spec,
schema,
Struct::from_iter(std::iter::once(lit)),
))
}
/// A fresh, monotonic-ish snapshot id distinct from `parent` (wall-clock ns).
fn fresh_snapshot_id(parent: Option<i64>) -> i64 {
let ns = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_nanos() as i64)
.unwrap_or(1)
.max(1);
if Some(ns) == parent { ns + 1 } else { ns }
}
fn now_ms() -> i64 {
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_millis() as i64)
.unwrap_or(0)
}
/// A unique file-name prefix (matches skade's `unique_prefix` convention).
fn unique_prefix() -> String {
format!("{}", uuid::Uuid::new_v4().simple())
}
/// Normalize a path/URI to a comparable key — strips a leading `file://`, then
/// canonicalizes when it resolves on disk (so `file://`, absolute, and symlinked
/// forms of the same data file collapse to one key).
fn canon(p: &str) -> String {
let stripped = p.strip_prefix("file://").unwrap_or(p);
match std::fs::canonicalize(stripped) {
Ok(c) => c.to_string_lossy().to_string(),
Err(_) => stripped.to_string(),
}
}
/// Recursively collect files under `dir` whose extension is in `exts`.
fn collect_files(dir: &Path, out: &mut Vec<PathBuf>, exts: &[&str]) {
let Ok(rd) = std::fs::read_dir(dir) else {
return;
};
for entry in rd.flatten() {
let p = entry.path();
if p.is_dir() {
collect_files(&p, out, exts);
} else if let Some(ext) = p.extension().and_then(|e| e.to_str()) {
if exts.contains(&ext) {
out.push(p);
}
}
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::warehouse::Warehouse;
use crate::warehouse::feature_dep_edges::{DepKind, EnableKind, FeatureDepEdgeRow};
use crate::warehouse::iceberg::{IcebergWarehouse, TABLE_FEATURE_DEP_EDGES};
// ── fixture helpers ──────────────────────────────────────────────────────
/// One `feature_dep_edges` append for `repo`: one edge per name (the name is
/// what it `enables`), in the given, deliberately-UNSORTED order and with the
/// declaring crate alternating `zeta`/`alpha`, so a missing sort is
/// observable. Each call writes its own tiny Parquet data file + snapshot.
fn append_edges(wh: &IcebergWarehouse, repo: &str, names: &[&str]) {
let rows: Vec<FeatureDepEdgeRow> = names
.iter()
.enumerate()
.map(|(i, name)| FeatureDepEdgeRow {
repo: repo.into(),
crate_name: if i % 2 == 0 { "zeta" } else { "alpha" }.into(),
feature: "f".into(),
enables: (*name).into(),
kind: EnableKind::Feature,
optional: false,
target_cfg: None,
dep_kind: DepKind::Normal,
})
.collect();
wh.append_feature_dep_edges(&uuid::Uuid::new_v4().to_string(), &rows)
.unwrap();
}
/// Count Parquet data files physically present under the table's `data/` dir.
fn count_data_files_on_disk(root: &Path, table: &str) -> usize {
let root = root.canonicalize().unwrap_or_else(|_| root.to_path_buf());
let dir = root
.join("warehouse")
.join("nornir")
.join(table)
.join("data");
let mut v = Vec::new();
collect_files(&dir, &mut v, &["parquet"]);
v.len()
}
/// Count `*.avro` manifest/manifest-list files under the table's metadata dir.
fn count_avro_files_on_disk(root: &Path, table: &str) -> usize {
let root = root.canonicalize().unwrap_or_else(|_| root.to_path_buf());
let dir = root
.join("warehouse")
.join("nornir")
.join(table)
.join("metadata");
let mut v = Vec::new();
collect_files(&dir, &mut v, &["avro"]);
v.len()
}
/// The canonicalized set of all files (data + manifests + manifest-lists) the
/// `keep` newest snapshots reference — the live set the GC must preserve.
fn retained_file_set(wh: &IcebergWarehouse, table: &str, keep: usize) -> HashSet<String> {
let ident = wh.table_ident(table);
wh.block_on(async {
let tbl = wh.catalog().load_table(&ident).await.unwrap();
let md = tbl.metadata();
let mut snaps: Vec<_> = md.snapshots().collect();
snaps.sort_by_key(|s| std::cmp::Reverse(s.timestamp_ms()));
let mut out = HashSet::new();
for s in snaps.iter().take(keep) {
out.insert(canon(s.manifest_list()));
let ml = tbl.manifest_list_reader(s).load().await.unwrap();
for mf in ml.entries() {
out.insert(canon(&mf.manifest_path));
let m = tbl.manifest_reader().read(mf).await.unwrap();
for e in m.entries() {
out.insert(canon(e.data_file().file_path()));
}
}
}
out
})
}
/// The set of `(repo, crate_name, enables)` rows the table currently reads back.
/// Takes `&dyn Warehouse` — it only needs the trait's `scan_arrow`, so it
/// reads any backend behind the seam.
fn read_rows(wh: &dyn Warehouse, table: &str) -> Vec<(String, String, String)> {
let batches = wh.scan_arrow(table).unwrap();
let mut rows = Vec::new();
for b in &batches {
let repo = b
.column_by_name("repo")
.unwrap()
.as_any()
.downcast_ref::<StringArray>()
.unwrap();
let file = b
.column_by_name("crate_name")
.unwrap()
.as_any()
.downcast_ref::<StringArray>()
.unwrap();
let name = b
.column_by_name("enables")
.unwrap()
.as_any()
.downcast_ref::<StringArray>()
.unwrap();
for i in 0..b.num_rows() {
rows.push((
repo.value(i).to_string(),
file.value(i).to_string(),
name.value(i).to_string(),
));
}
}
rows
}
/// Data files referenced by the table's CURRENT (head) snapshot manifest.
fn head_data_files(wh: &IcebergWarehouse, table: &str) -> Vec<String> {
let ident = wh.table_ident(table);
wh.block_on(async {
let tbl = wh.catalog().load_table(&ident).await.unwrap();
let md = tbl.metadata();
let cur = md.current_snapshot().unwrap();
let ml = tbl.manifest_list_reader(cur).load().await.unwrap();
let mut out = Vec::new();
for mf in ml.entries() {
let m = tbl.manifest_reader().read(mf).await.unwrap();
for e in m.entries() {
if e.content_type() == DataContentType::Data {
out.push(e.data_file().file_path().to_string());
}
}
}
out
})
}
fn snapshot_count(wh: &IcebergWarehouse, table: &str) -> usize {
let ident = wh.table_ident(table);
wh.block_on(async {
wh.catalog()
.load_table(&ident)
.await
.unwrap()
.metadata()
.snapshots()
.len()
})
}
// ── RESET ────────────────────────────────────────────────────────────────
/// RED-WHEN-REVERTED, in three separate ways.
///
/// 1. **Both paths, or the warehouse is bricked.** The half-nuke everyone
/// reaches for — `rm -rf warehouse/` and leave `catalog.redb` — is
/// asserted here to leave a warehouse whose tables still claim rows that
/// are no longer on disk. Drop the `remove_file(&catalog)` from
/// [`reset`] and this test fails on the re-open row count.
/// 2. **A dry run must not delete.** `apply == false` leaves both paths.
/// 3. **No success over a no-op.** Reset of a never-opened root reports
/// `removed == false`, so a caller cannot print a green line.
#[test]
fn reset_empties_both_catalog_and_data_and_never_greens_a_no_op() {
let dir = tempfile::tempdir().unwrap();
let root = dir.path().join("wh");
// (3) NOTHING THERE — must not claim to have removed anything.
let nothing = reset(&root, true).unwrap();
assert!(
!nothing.had_catalog && !nothing.had_data,
"nothing on disk yet"
);
assert!(
!nothing.removed,
"a reset with nothing to remove is NOT a removal"
);
// Populate a real warehouse: rows on disk, catalog + data both present.
{
let wh = IcebergWarehouse::open(&root).unwrap();
append_edges(&wh, "facett", &["mango", "apple"]);
append_edges(&wh, "nordisk", &["sierra"]);
assert_eq!(
read_rows(&wh, TABLE_FEATURE_DEP_EDGES).len(),
3,
"fixture has rows"
);
}
assert!(root.join("catalog.redb").is_file(), "catalog exists");
assert!(root.join("warehouse").is_dir(), "data dir exists");
// (2) DRY RUN — reports what is there, deletes nothing.
let dry = reset(&root, false).unwrap();
assert!(dry.had_catalog && dry.had_data, "dry run sees both paths");
assert!(
dry.data_files > 0 && dry.data_bytes > 0,
"dry run sizes the data ({dry:?})"
);
assert!(!dry.removed, "a dry run removes NOTHING");
assert!(
root.join("catalog.redb").is_file(),
"dry run left the catalog"
);
assert!(root.join("warehouse").is_dir(), "dry run left the data");
// (1a) THE HALF-NUKE IS BROKEN — data gone, catalog kept. Re-opening
// yields a warehouse that still *claims* the rows whose files are gone:
// the catalog is a liar, which is exactly why `reset` must remove both.
std::fs::remove_dir_all(root.join("warehouse")).unwrap();
{
let wh = IcebergWarehouse::open(&root).unwrap();
let rows = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
read_rows(&wh, TABLE_FEATURE_DEP_EDGES).len()
}));
assert!(
!matches!(rows, Ok(0)),
"half-nuke (data removed, catalog kept) must NOT read as a clean empty \
warehouse — it either still claims rows or fails; got {rows:?}"
);
}
// (1b) THE FULL RESET — both paths gone, and the next open is a genuinely
// EMPTY warehouse that accepts writes again.
let rep = reset(&root, true).unwrap();
assert!(rep.removed, "the real reset removed something");
assert!(rep.had_catalog, "the leftover catalog was there to remove");
assert!(!root.join("catalog.redb").exists(), "catalog.redb is GONE");
assert!(!root.join("warehouse").exists(), "warehouse/ is GONE");
{
let wh = IcebergWarehouse::open(&root).unwrap();
assert_eq!(
read_rows(&wh, TABLE_FEATURE_DEP_EDGES).len(),
0,
"a reset warehouse re-opens EMPTY"
);
append_edges(&wh, "facett", &["fresh"]);
assert_eq!(
read_rows(&wh, TABLE_FEATURE_DEP_EDGES).len(),
1,
"and it takes writes again"
);
}
}
// ── COMPACTION ───────────────────────────────────────────────────────────
#[test]
fn compact_merges_small_files_preserving_rows_sortedness_and_tightens_bounds() {
use skade::parquet::arrow::arrow_reader::ParquetRecordBatchReaderBuilder;
use skade::parquet::file::reader::FileReader;
use skade::parquet::file::serialized_reader::SerializedFileReader;
let dir = tempfile::tempdir().unwrap();
let root = dir.path().to_path_buf();
let wh = IcebergWarehouse::open(&root).unwrap();
// N tiny appends across two repos → many small files + many snapshots.
append_edges(&wh, "facett", &["mango", "apple", "zulu"]);
append_edges(&wh, "facett", &["delta", "banana"]);
append_edges(&wh, "nordisk", &["sierra", "alpha"]);
append_edges(&wh, "nordisk", &["tango", "bravo", "kilo"]);
append_edges(&wh, "facett", &["echo"]);
let before_head = head_data_files(&wh, TABLE_FEATURE_DEP_EDGES).len();
let before_rows = read_rows(&wh, TABLE_FEATURE_DEP_EDGES);
assert!(
before_head >= 5,
"many small head data files before compaction (got {before_head})"
);
assert!(!before_rows.is_empty(), "RAGNARÖK: fixture has rows");
// Compact.
let reports = compact_all(&wh, Some(TABLE_FEATURE_DEP_EDGES), None).unwrap();
let rep = &reports[0];
assert!(!rep.skipped, "compaction ran (not skipped)");
// (a) FILE COUNT DROPPED — the head now references far fewer data files.
let after_head = head_data_files(&wh, TABLE_FEATURE_DEP_EDGES);
assert!(
after_head.len() < before_head,
"head data file count dropped: {before_head} -> {}",
after_head.len(),
);
// Two repos = two identity partitions ⇒ at most 2 compacted files.
assert!(
after_head.len() <= 2,
"compacted to ≤2 files (one per repo partition), got {}",
after_head.len()
);
assert_eq!(rep.files_before, before_head);
assert_eq!(rep.files_after, after_head.len());
// (b) ROW COUNT + CONTENT IDENTICAL — read back, compare as multisets.
let after_rows = read_rows(&wh, TABLE_FEATURE_DEP_EDGES);
assert_eq!(after_rows.len(), before_rows.len(), "row count conserved");
let mut b = before_rows.clone();
let mut a = after_rows.clone();
b.sort();
a.sort();
assert_eq!(a, b, "content identical after compaction (multiset equal)");
assert_eq!(rep.rows, before_rows.len() as u64);
// (c) DATA STILL SORTED by the SortOrder within each compacted file, AND
// (d) TIGHTER BOUNDS — assert the new files have non-overlapping repo
// ranges (each compacted file is a single repo partition).
let mut file_repo_ranges: Vec<(String, String)> = Vec::new();
for path in &after_head {
let p = path.strip_prefix("file://").unwrap_or(path);
// sortedness within the file
let mut prev: Option<(String, String)> = None;
let f = std::fs::File::open(p).unwrap();
let mut rb = ParquetRecordBatchReaderBuilder::try_new(f)
.unwrap()
.build()
.unwrap();
while let Some(batch) = rb.next() {
let batch = batch.unwrap();
let file_col = batch
.column_by_name("enables")
.unwrap()
.as_any()
.downcast_ref::<StringArray>()
.unwrap();
let name_col = batch
.column_by_name("crate_name")
.unwrap()
.as_any()
.downcast_ref::<StringArray>()
.unwrap();
for i in 0..batch.num_rows() {
let key = (file_col.value(i).to_string(), name_col.value(i).to_string());
if let Some(pv) = &prev {
assert!(
pv <= &key,
"compacted file rows sorted by (enables,crate_name): {pv:?} !<= {key:?}"
);
}
prev = Some(key);
}
}
// per-file repo min/max bounds (the manifest skip bounds)
let f2 = std::fs::File::open(p).unwrap();
let reader = SerializedFileReader::new(f2).unwrap();
let meta = reader.metadata();
let (mut lo, mut hi): (Option<String>, Option<String>) = (None, None);
for rg in 0..meta.num_row_groups() {
for col in meta.row_group(rg).columns() {
if col.column_path().string() == "repo" {
let s = col
.statistics()
.expect("repo stats present (manifest bounds)");
let min = String::from_utf8_lossy(s.min_bytes_opt().unwrap()).to_string();
let max = String::from_utf8_lossy(s.max_bytes_opt().unwrap()).to_string();
lo = Some(lo.map_or(min.clone(), |x| x.min(min)));
hi = Some(hi.map_or(max.clone(), |x| x.max(max)));
}
}
}
let (lo, hi) = (lo.unwrap(), hi.unwrap());
// tight: a single-repo file has repo min == max (degenerate-tight range).
assert_eq!(
lo, hi,
"compacted file is a single repo partition ⇒ tight repo bound [{lo},{hi}]"
);
file_repo_ranges.push((lo, hi));
}
// (d) NON-OVERLAPPING across files: the two repos land in distinct files.
let distinct: HashSet<_> = file_repo_ranges.iter().map(|(lo, _)| lo.clone()).collect();
assert_eq!(
distinct.len(),
after_head.len(),
"each compacted file covers a distinct repo (non-overlapping bounds)"
);
eprintln!(
"[compact] feature_dep_edges: {before_head} → {} head data file(s), {} row(s) preserved, sorted, tight per-repo bounds {file_repo_ranges:?}",
after_head.len(),
after_rows.len(),
);
// Sensitivity note: dropping the manifest-list "drop old data manifests"
// step would leave files_after >= files_before and fail assertion (a).
let _ = count_data_files_on_disk; // (used by the expiry test)
}
// ── EXPIRY ───────────────────────────────────────────────────────────────
#[test]
fn expire_keeps_last_n_and_gcs_orphans_without_touching_retained_files() {
let dir = tempfile::tempdir().unwrap();
let root = dir.path().to_path_buf();
let wh = IcebergWarehouse::open(&root).unwrap();
// K appends ⇒ K snapshots (single repo so every file is the same partition).
for batch in [
&["a1", "a2"][..],
&["b1"][..],
&["c1", "c2", "c3"][..],
&["d1"][..],
&["e1", "e2"][..],
] {
append_edges(&wh, "facett", batch);
}
let snaps_before = snapshot_count(&wh, TABLE_FEATURE_DEP_EDGES);
assert!(
snaps_before >= 5,
"K snapshots created (got {snaps_before})"
);
// Files referenced by the snapshots we will RETAIN (the last 2). Capturing
// them lets us prove a retained file is NEVER deleted.
// The complete set of files (data + manifests + manifest-lists) the 2
// newest snapshots reference — these MUST survive the GC untouched.
// NOTE: an append-only table's HEAD references EVERY accumulated data
// file (fast_append carries prior manifests forward), so expiry here
// reclaims the stale per-snapshot manifest-LISTs + orphaned manifest
// avros, NOT data parquet — data reclaim needs a prior compaction (see
// `compact_then_expire_reclaims_superseded_data_files`). This is the
// correct, safe Iceberg semantics.
let live_before = retained_file_set(&wh, TABLE_FEATURE_DEP_EDGES, 2);
let avro_before = count_avro_files_on_disk(&root, TABLE_FEATURE_DEP_EDGES);
// Expire: keep the last 2 snapshots.
let reports = expire_all(&wh, Some(TABLE_FEATURE_DEP_EDGES), 2, None).unwrap();
let rep = reports
.iter()
.find(|r| r.table == TABLE_FEATURE_DEP_EDGES)
.unwrap();
// Only 2 snapshots remain.
let snaps_after = snapshot_count(&wh, TABLE_FEATURE_DEP_EDGES);
assert_eq!(
snaps_after, 2,
"exactly 2 snapshots retained (was {snaps_before})"
);
assert_eq!(rep.snapshots_after, 2);
assert!(rep.snapshots_before >= 5);
// The current snapshot still reads correctly (RAGNARÖK: non-empty + the
// newest rows present + EVERY appended row still there — the head carries
// all data forward).
let rows = read_rows(&wh, TABLE_FEATURE_DEP_EDGES);
assert!(
!rows.is_empty(),
"current snapshot reads non-empty after expiry"
);
assert!(
rows.iter().any(|(_, _, n)| n == "e1"),
"head snapshot's newest rows still readable"
);
assert!(
rows.iter().any(|(_, _, n)| n == "a1"),
"head still carries the oldest appended rows"
);
// Orphans were deleted: the stale manifest-list/manifest avro count dropped.
let avro_after = count_avro_files_on_disk(&root, TABLE_FEATURE_DEP_EDGES);
assert!(rep.files_deleted > 0, "deleted some orphan files");
assert!(
avro_after < avro_before,
"stale manifest/manifest-list avros GC'd: {avro_before} -> {avro_after}",
);
// SAFETY INVARIANT: every file the retained snapshots reference STILL
// exists on disk (no retained-snapshot file was deleted).
for f in &live_before {
assert!(
Path::new(f).exists(),
"retained-snapshot file was wrongly deleted: {f}"
);
}
eprintln!(
"[expire] feature_dep_edges: {snaps_before} → {snaps_after} snapshot(s); deleted {} orphan avro/file(s) ({} → {} avros on disk), {} byte(s) reclaimed; all {} retained files intact",
rep.files_deleted,
avro_before,
avro_after,
rep.bytes_reclaimed,
live_before.len(),
);
}
/// The realistic DATA-reclaim path: compact (replace snapshot drops old data
/// manifests) → expire (removes the pre-compaction snapshots) → the old small
/// data files are no longer referenced by any retained snapshot and are GC'd.
/// Proves (a) data parquet files drop on disk, (b) row content survives, and
/// (c) no retained-snapshot file was deleted.
#[test]
fn compact_then_expire_reclaims_superseded_data_files() {
let dir = tempfile::tempdir().unwrap();
let root = dir.path().to_path_buf();
let wh = IcebergWarehouse::open(&root).unwrap();
for batch in [
&["a1", "a2"][..],
&["b1"][..],
&["c1", "c2"][..],
&["d1"][..],
] {
append_edges(&wh, "facett", batch);
}
let rows_before = read_rows(&wh, TABLE_FEATURE_DEP_EDGES);
let data_before = count_data_files_on_disk(&root, TABLE_FEATURE_DEP_EDGES);
assert!(
data_before >= 4,
"many small data files before (got {data_before})"
);
// Compact ⇒ a Replace head referencing 1 file; old files now only live in
// the superseded history snapshots.
compact_all(&wh, Some(TABLE_FEATURE_DEP_EDGES), None).unwrap();
// Expire down to the current (compacted) snapshot only ⇒ the superseded
// history (with the old data manifests) is removed, orphaning old files.
let reports = expire_all(&wh, Some(TABLE_FEATURE_DEP_EDGES), 1, None).unwrap();
let rep = reports
.iter()
.find(|r| r.table == TABLE_FEATURE_DEP_EDGES)
.unwrap();
// The retained head's files (post-compaction) must survive.
let live = retained_file_set(&wh, TABLE_FEATURE_DEP_EDGES, 1);
// (a) data parquet files dropped — the small pre-compaction files are gone.
let data_after = count_data_files_on_disk(&root, TABLE_FEATURE_DEP_EDGES);
assert!(
data_after < data_before,
"data parquet files reclaimed: {data_before} -> {data_after}",
);
assert!(
rep.files_deleted > 0 && rep.bytes_reclaimed > 0,
"disk reclaimed"
);
// (b) content survives identically (multiset equal).
let rows_after = read_rows(&wh, TABLE_FEATURE_DEP_EDGES);
let (mut a, mut b) = (rows_after.clone(), rows_before.clone());
a.sort();
b.sort();
assert_eq!(a, b, "row content identical through compact+expire");
// (c) no retained file deleted.
for f in &live {
assert!(Path::new(f).exists(), "retained file wrongly deleted: {f}");
}
eprintln!(
"[compact+expire] feature_dep_edges: data files {data_before} → {data_after} on disk, {} byte(s) reclaimed, {} rows preserved, retained files intact",
rep.bytes_reclaimed,
rows_after.len(),
);
}
#[test]
fn expire_keeps_all_when_under_threshold() {
let dir = tempfile::tempdir().unwrap();
let root = dir.path().to_path_buf();
let wh = IcebergWarehouse::open(&root).unwrap();
append_edges(&wh, "facett", &["a"]);
append_edges(&wh, "facett", &["b"]);
let before = snapshot_count(&wh, TABLE_FEATURE_DEP_EDGES);
let reports = expire_all(&wh, Some(TABLE_FEATURE_DEP_EDGES), 10, None).unwrap();
let rep = reports
.iter()
.find(|r| r.table == TABLE_FEATURE_DEP_EDGES)
.unwrap();
assert_eq!(rep.files_deleted, 0, "nothing aged out under keep-last=10");
assert_eq!(
snapshot_count(&wh, TABLE_FEATURE_DEP_EDGES),
before,
"all snapshots retained"
);
}
}