From de3737ce6521431736bb5c4445302dfb3d27e5f8 Mon Sep 17 00:00:00 2001 From: Claude Date: Sun, 18 Jan 2026 04:46:55 +0000 Subject: [PATCH 1/4] feat: add snapshot cleanup command for automatic snapshot expiration This commit adds a new CLI command for managing and cleaning up old snapshots based on configurable retention policies. New features: - `icepick snapshot list `: List all snapshots with age and refs - `icepick snapshot cleanup
`: Expire old snapshots based on policy - `--older-than-days N`: Minimum age before a snapshot can expire (default: 7) - `--retain-last N`: Minimum snapshots to always retain (default: 10) - `--dry-run`: Preview what would be removed without executing - Both conditions must be met before a snapshot is expired Implementation details: - Added `snapshot_cleanup` module with planning and execution logic - Extended Catalog trait with `expire_snapshots()` method - Added `RemoveSnapshots` variant to REST API TableUpdate enum - Added `snapshot` CLI subcommand with `list` and `cleanup` operations - Snapshots referenced by branches/tags or marked as current are protected --- src/bin/icepick.rs | 8 +- src/catalog/catalog_trait.rs | 25 ++ src/catalog/rest/catalog_impl.rs | 43 +++ src/catalog/rest/catalog_trait.rs | 8 + src/catalog/rest/commit_types.rs | 7 + src/catalog/rest_catalog.rs | 8 + src/cli/commands/mod.rs | 1 + src/cli/commands/snapshot.rs | 443 +++++++++++++++++++++++++++ src/lib.rs | 7 + src/snapshot_cleanup/mod.rs | 486 ++++++++++++++++++++++++++++++ 10 files changed, 1035 insertions(+), 1 deletion(-) create mode 100644 src/cli/commands/snapshot.rs create mode 100644 src/snapshot_cleanup/mod.rs diff --git a/src/bin/icepick.rs b/src/bin/icepick.rs index 9fb25d2..e6b7dd7 100644 --- a/src/bin/icepick.rs +++ b/src/bin/icepick.rs @@ -2,7 +2,8 @@ use clap::{Parser, Subcommand}; use icepick::cli::commands::{ - catalog as catalog_cmd, compact as compact_cmd, namespace as namespace_cmd, table as table_cmd, + catalog as catalog_cmd, compact as compact_cmd, namespace as namespace_cmd, + snapshot as snapshot_cmd, table as table_cmd, }; use icepick::cli::{CatalogConfig, OutputFormat}; @@ -41,6 +42,10 @@ enum Commands { #[command(subcommand)] Table(table_cmd::TableCommand), + /// Snapshot operations (list, cleanup) + #[command(subcommand)] + Snapshot(snapshot_cmd::SnapshotCommand), + /// Compact a table Compact(compact_cmd::CompactArgs), } @@ -66,6 +71,7 @@ async fn main() { Commands::Catalog(cmd) => catalog_cmd::execute(cmd, &config, cli.output).await, Commands::Namespace(cmd) => namespace_cmd::execute(cmd, &config, cli.output).await, Commands::Table(cmd) => table_cmd::execute(cmd, &config, cli.output).await, + Commands::Snapshot(cmd) => snapshot_cmd::execute(cmd, &config, cli.output).await, Commands::Compact(args) => compact_cmd::execute(args, &config, cli.output).await, }; diff --git a/src/catalog/catalog_trait.rs b/src/catalog/catalog_trait.rs index 07e11c2..b4d9f66 100644 --- a/src/catalog/catalog_trait.rs +++ b/src/catalog/catalog_trait.rs @@ -86,4 +86,29 @@ pub trait Catalog: Send + Sync { "Schema evolution not supported for this catalog implementation", )) } + + /// Expire (remove) snapshots from a table by their IDs + /// + /// This method removes the specified snapshots from the table metadata, + /// allowing their associated data files to be garbage collected. + /// + /// # Arguments + /// * `identifier` - The table identifier + /// * `snapshot_ids` - List of snapshot IDs to expire + /// + /// # Returns + /// Ok(()) if successful, error otherwise + /// + /// # Notes + /// - The current snapshot cannot be expired + /// - Snapshots referenced by branches or tags should not be expired + async fn expire_snapshots( + &self, + _identifier: &TableIdent, + _snapshot_ids: &[i64], + ) -> Result<()> { + Err(crate::error::Error::invalid_input( + "Snapshot expiration not supported for this catalog implementation", + )) + } } diff --git a/src/catalog/rest/catalog_impl.rs b/src/catalog/rest/catalog_impl.rs index 3cc20aa..2edf6cc 100644 --- a/src/catalog/rest/catalog_impl.rs +++ b/src/catalog/rest/catalog_impl.rs @@ -468,3 +468,46 @@ fn commit_table_enabled() -> bool { Err(_) => false, } } + +impl IcebergRestCatalog { + /// Expire snapshots by their IDs using the REST catalog API + pub(super) async fn expire_snapshots_impl( + &self, + identifier: &crate::spec::TableIdent, + snapshot_ids: &[i64], + ) -> crate::error::Result<()> { + if snapshot_ids.is_empty() { + return Ok(()); + } + + // Load current table to get requirements + let table = self.load_table_impl(identifier).await?; + let metadata = table.metadata(); + let current_snapshot_id = metadata.current_snapshot_id(); + let reference = self.options.reference().to_string(); + + // Build commit request with RemoveSnapshots update + let requirements = vec![ + TableRequirement::AssertTableUuid { + uuid: metadata.table_uuid().to_string(), + }, + TableRequirement::AssertRefSnapshotId { + r#ref: reference, + snapshot_id: current_snapshot_id, + }, + ]; + + let request = CommitTableRequest { + requirements, + updates: vec![TableUpdate::RemoveSnapshots { + snapshot_ids: snapshot_ids.to_vec(), + }], + }; + + self.commit_table(identifier, request) + .await + .map_err(helpers::from_catalog_error)?; + + Ok(()) + } +} diff --git a/src/catalog/rest/catalog_trait.rs b/src/catalog/rest/catalog_trait.rs index b134072..c8b710d 100644 --- a/src/catalog/rest/catalog_trait.rs +++ b/src/catalog/rest/catalog_trait.rs @@ -72,4 +72,12 @@ impl crate::catalog::Catalog for IcebergRestCatalog { self.update_table_metadata_impl(identifier, old_metadata_location, new_metadata_location) .await } + + async fn expire_snapshots( + &self, + identifier: &crate::spec::TableIdent, + snapshot_ids: &[i64], + ) -> crate::error::Result<()> { + self.expire_snapshots_impl(identifier, snapshot_ids).await + } } diff --git a/src/catalog/rest/commit_types.rs b/src/catalog/rest/commit_types.rs index 94f3d9f..9cc57dd 100644 --- a/src/catalog/rest/commit_types.rs +++ b/src/catalog/rest/commit_types.rs @@ -75,6 +75,13 @@ pub enum TableUpdate { max_ref_age_ms: Option, }, + /// Remove snapshots by their IDs + #[serde(rename = "remove-snapshots")] + RemoveSnapshots { + #[serde(rename = "snapshot-ids")] + snapshot_ids: Vec, + }, + #[serde(rename = "upgrade-format-version")] UpgradeFormatVersion { #[serde(rename = "format-version")] diff --git a/src/catalog/rest_catalog.rs b/src/catalog/rest_catalog.rs index 6af56bc..ab87b43 100644 --- a/src/catalog/rest_catalog.rs +++ b/src/catalog/rest_catalog.rs @@ -289,6 +289,10 @@ impl Catalog for RestCatalog { .update_table_metadata(identifier, old_metadata_location, new_metadata_location) .await } + + async fn expire_snapshots(&self, identifier: &TableIdent, snapshot_ids: &[i64]) -> Result<()> { + self.inner.expire_snapshots(identifier, snapshot_ids).await + } } // Implement Catalog trait for WASM targets without Send requirement. @@ -345,6 +349,10 @@ impl Catalog for RestCatalog { .update_table_metadata(identifier, old_metadata_location, new_metadata_location) .await } + + async fn expire_snapshots(&self, identifier: &TableIdent, snapshot_ids: &[i64]) -> Result<()> { + self.inner.expire_snapshots(identifier, snapshot_ids).await + } } #[cfg_attr(not(target_family = "wasm"), async_trait)] diff --git a/src/cli/commands/mod.rs b/src/cli/commands/mod.rs index 188b38a..7d653e7 100644 --- a/src/cli/commands/mod.rs +++ b/src/cli/commands/mod.rs @@ -3,4 +3,5 @@ pub mod catalog; pub mod compact; pub mod namespace; +pub mod snapshot; pub mod table; diff --git a/src/cli/commands/snapshot.rs b/src/cli/commands/snapshot.rs new file mode 100644 index 0000000..0d992ba --- /dev/null +++ b/src/cli/commands/snapshot.rs @@ -0,0 +1,443 @@ +//! Snapshot commands for listing and cleanup + +use crate::cli::catalog::CatalogConfig; +use crate::cli::output::{format_number, print, OutputFormat, Outputable}; +use crate::cli::util::parse_table_ident; +use crate::snapshot_cleanup::{ + plan_snapshot_cleanup, CleanupOptions, CleanupPlan, RetentionReason, +}; +use chrono::{TimeZone, Utc}; +use clap::Subcommand; +use comfy_table::{Row, Table as ComfyTable}; +use serde::Serialize; + +/// Snapshot commands +#[derive(Debug, Subcommand)] +pub enum SnapshotCommand { + /// List snapshots in a table + List { + /// Table identifier (namespace.table) + table: String, + }, + + /// Cleanup old snapshots based on retention policy + Cleanup { + /// Table identifier (namespace.table) + table: String, + + /// Minimum age in days before a snapshot can be expired + #[arg(long, default_value = "7")] + older_than_days: u32, + + /// Minimum number of snapshots to always retain (most recent) + #[arg(long, default_value = "10")] + retain_last: usize, + + /// Show plan without executing + #[arg(long)] + dry_run: bool, + }, +} + +/// Snapshot list output +#[derive(Debug, Serialize)] +pub struct SnapshotList { + pub table: String, + pub snapshots: Vec, + pub total_count: usize, + pub current_snapshot_id: Option, +} + +#[derive(Debug, Serialize)] +pub struct SnapshotEntry { + pub snapshot_id: i64, + pub timestamp: String, + pub age_days: f64, + pub operation: String, + pub is_current: bool, + pub refs: Vec, +} + +impl Outputable for SnapshotList { + fn to_text(&self) -> String { + if self.snapshots.is_empty() { + return format!("No snapshots found in table '{}'.", self.table); + } + + let mut lines = vec![format!("Snapshots in '{}':", self.table), String::new()]; + + let mut table = ComfyTable::new(); + table.set_header(Row::from(vec![ + "Snapshot ID", + "Timestamp", + "Age", + "Operation", + "Current", + "Refs", + ])); + + for snapshot in &self.snapshots { + let age_str = if snapshot.age_days < 1.0 { + format!("{:.1}h", snapshot.age_days * 24.0) + } else { + format!("{:.1}d", snapshot.age_days) + }; + + table.add_row(Row::from(vec![ + snapshot.snapshot_id.to_string(), + snapshot.timestamp.clone(), + age_str, + snapshot.operation.clone(), + if snapshot.is_current { "yes" } else { "" }.to_string(), + snapshot.refs.join(", "), + ])); + } + lines.push(table.to_string()); + + lines.push(String::new()); + lines.push(format!("Total: {} snapshots", self.total_count)); + + lines.join("\n") + } +} + +/// Cleanup plan output +#[derive(Debug, Serialize)] +pub struct CleanupPlanOutput { + pub table: String, + pub older_than_days: u32, + pub retain_last: usize, + pub total_snapshots: usize, + pub snapshots_to_remove: Vec, + pub snapshots_to_retain: Vec, + pub dry_run: bool, +} + +#[derive(Debug, Serialize)] +pub struct SnapshotToRemove { + pub snapshot_id: i64, + pub timestamp: String, + pub age_days: f64, + pub operation: String, +} + +#[derive(Debug, Serialize)] +pub struct SnapshotToRetain { + pub snapshot_id: i64, + pub timestamp: String, + pub age_days: f64, + pub reason: String, +} + +impl Outputable for CleanupPlanOutput { + fn to_text(&self) -> String { + let mut lines = vec![ + format!("Snapshot Cleanup Plan for {}", self.table), + String::new(), + format!("Policy:"), + format!(" Older than: {} days", self.older_than_days), + format!(" Retain last: {} snapshots", self.retain_last), + String::new(), + ]; + + if self.snapshots_to_remove.is_empty() { + lines.push("No snapshots eligible for removal.".to_string()); + } else { + lines.push(format!( + "Snapshots to remove ({}):", + self.snapshots_to_remove.len() + )); + + let mut table = ComfyTable::new(); + table.set_header(Row::from(vec![ + "Snapshot ID", + "Timestamp", + "Age", + "Operation", + ])); + + for snapshot in &self.snapshots_to_remove { + table.add_row(Row::from(vec![ + snapshot.snapshot_id.to_string(), + snapshot.timestamp.clone(), + format!("{:.1}d", snapshot.age_days), + snapshot.operation.clone(), + ])); + } + lines.push(table.to_string()); + } + + lines.push(String::new()); + lines.push(format!( + "Snapshots to retain ({}):", + self.snapshots_to_retain.len() + )); + + if !self.snapshots_to_retain.is_empty() { + let mut retain_table = ComfyTable::new(); + retain_table.set_header(Row::from(vec![ + "Snapshot ID", + "Timestamp", + "Age", + "Reason", + ])); + + for snapshot in &self.snapshots_to_retain { + let age_str = if snapshot.age_days < 1.0 { + format!("{:.1}h", snapshot.age_days * 24.0) + } else { + format!("{:.1}d", snapshot.age_days) + }; + + retain_table.add_row(Row::from(vec![ + snapshot.snapshot_id.to_string(), + snapshot.timestamp.clone(), + age_str, + snapshot.reason.clone(), + ])); + } + lines.push(retain_table.to_string()); + } + + lines.push(String::new()); + lines.push("Summary".to_string()); + lines.push(format!( + " Total: {} snapshots", + format_number(self.total_snapshots as u64) + )); + lines.push(format!( + " Remove: {} snapshots", + format_number(self.snapshots_to_remove.len() as u64) + )); + lines.push(format!( + " Retain: {} snapshots", + format_number(self.snapshots_to_retain.len() as u64) + )); + + if self.dry_run && !self.snapshots_to_remove.is_empty() { + lines.push(String::new()); + lines.push("Dry run complete. Remove --dry-run to execute.".to_string()); + } + + lines.join("\n") + } +} + +/// Cleanup result output +#[derive(Debug, Serialize)] +pub struct CleanupResultOutput { + pub table: String, + pub snapshots_removed: usize, + pub snapshots_retained: usize, + pub removed_snapshot_ids: Vec, +} + +impl Outputable for CleanupResultOutput { + fn to_text(&self) -> String { + let mut lines = vec![format!("Snapshot Cleanup Complete for {}", self.table)]; + + lines.push(String::new()); + lines.push(format!( + "Removed: {} snapshots", + format_number(self.snapshots_removed as u64) + )); + lines.push(format!( + "Retained: {} snapshots", + format_number(self.snapshots_retained as u64) + )); + + if !self.removed_snapshot_ids.is_empty() && self.removed_snapshot_ids.len() <= 10 { + lines.push(String::new()); + lines.push("Removed snapshot IDs:".to_string()); + for id in &self.removed_snapshot_ids { + lines.push(format!(" {}", id)); + } + } + + lines.join("\n") + } +} + +fn format_timestamp(timestamp_ms: i64) -> String { + Utc.timestamp_millis_opt(timestamp_ms) + .single() + .map(|dt| dt.format("%Y-%m-%d %H:%M:%S UTC").to_string()) + .unwrap_or_else(|| "Invalid timestamp".to_string()) +} + +fn format_retention_reason(reason: &RetentionReason) -> String { + match reason { + RetentionReason::CurrentSnapshot => "Current snapshot".to_string(), + RetentionReason::WithinRetainCount => "Within retain-last count".to_string(), + RetentionReason::NotOldEnough => "Not old enough".to_string(), + RetentionReason::ReferencedByRef(refs) => format!("Referenced by: {}", refs), + } +} + +fn build_cleanup_plan_output(table: &str, plan: &CleanupPlan, dry_run: bool) -> CleanupPlanOutput { + let snapshots_to_remove: Vec = plan + .snapshots_to_remove + .iter() + .map(|s| SnapshotToRemove { + snapshot_id: s.snapshot_id, + timestamp: format_timestamp(s.timestamp_ms), + age_days: s.age_days, + operation: s.operation.clone(), + }) + .collect(); + + let snapshots_to_retain: Vec = plan + .snapshots_to_retain + .iter() + .map(|s| SnapshotToRetain { + snapshot_id: s.info.snapshot_id, + timestamp: format_timestamp(s.info.timestamp_ms), + age_days: s.info.age_days, + reason: format_retention_reason(&s.reason), + }) + .collect(); + + CleanupPlanOutput { + table: table.to_string(), + older_than_days: plan.older_than_days, + retain_last: plan.retain_last, + total_snapshots: plan.total_snapshots, + snapshots_to_remove, + snapshots_to_retain, + dry_run, + } +} + +/// Execute a snapshot command +pub async fn execute( + command: SnapshotCommand, + config: &CatalogConfig, + format: OutputFormat, +) -> Result<(), String> { + let catalog = config.create_catalog().await?; + + match command { + SnapshotCommand::List { table: table_str } => { + let table_ident = parse_table_ident(&table_str)?; + let table = catalog + .load_table(&table_ident) + .await + .map_err(|e| format!("Failed to load table: {}", e))?; + + let metadata = table.metadata(); + let current_snapshot_id = metadata.current_snapshot_id(); + + // Build map of snapshot_id -> ref names + let mut snapshot_refs: std::collections::HashMap> = + std::collections::HashMap::new(); + for (ref_name, snapshot_ref) in metadata.refs() { + snapshot_refs + .entry(snapshot_ref.snapshot_id()) + .or_default() + .push(ref_name.clone()); + } + + let now_ms = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map_err(|e| format!("Failed to get current time: {}", e))? + .as_millis() as i64; + + let mut snapshots: Vec = metadata + .snapshots() + .iter() + .map(|s| { + let age_ms = now_ms - s.timestamp_ms(); + let age_days = age_ms as f64 / (24.0 * 60.0 * 60.0 * 1000.0); + + SnapshotEntry { + snapshot_id: s.snapshot_id(), + timestamp: format_timestamp(s.timestamp_ms()), + age_days, + operation: s.summary().operation().to_string(), + is_current: current_snapshot_id == Some(s.snapshot_id()), + refs: snapshot_refs + .get(&s.snapshot_id()) + .cloned() + .unwrap_or_default(), + } + }) + .collect(); + + // Sort by timestamp descending (newest first) + snapshots.sort_by(|a, b| b.snapshot_id.cmp(&a.snapshot_id)); + + let result = SnapshotList { + table: table_str, + total_count: snapshots.len(), + current_snapshot_id, + snapshots, + }; + + print(&result, format); + Ok(()) + } + + SnapshotCommand::Cleanup { + table: table_str, + older_than_days, + retain_last, + dry_run, + } => { + let table_ident = parse_table_ident(&table_str)?; + let table = catalog + .load_table(&table_ident) + .await + .map_err(|e| format!("Failed to load table: {}", e))?; + + // Build cleanup options + let options = CleanupOptions::new() + .with_older_than_days(older_than_days) + .with_retain_last(retain_last) + .with_dry_run(dry_run); + + // Create cleanup plan + let plan = plan_snapshot_cleanup(&table, &options) + .map_err(|e| format!("Failed to create cleanup plan: {}", e))?; + + if plan.is_empty() { + println!("No snapshots eligible for cleanup."); + return Ok(()); + } + + if dry_run { + // Output plan + let plan_output = build_cleanup_plan_output(&table_str, &plan, true); + print(&plan_output, format); + return Ok(()); + } + + // Show plan first + let plan_output = build_cleanup_plan_output(&table_str, &plan, false); + print(&plan_output, format); + + println!("\nExecuting cleanup..."); + + // Execute cleanup + let snapshot_ids: Vec = plan + .snapshots_to_remove + .iter() + .map(|s| s.snapshot_id) + .collect(); + + catalog + .expire_snapshots(&table_ident, &snapshot_ids) + .await + .map_err(|e| format!("Cleanup failed: {}", e))?; + + let result = CleanupResultOutput { + table: table_str, + snapshots_removed: plan.snapshots_to_remove.len(), + snapshots_retained: plan.snapshots_to_retain.len(), + removed_snapshot_ids: snapshot_ids, + }; + + print(&result, format); + Ok(()) + } + } +} diff --git a/src/lib.rs b/src/lib.rs index c770f89..2bbf1c2 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -59,6 +59,7 @@ pub mod io; pub mod manifest; pub mod reader; pub mod scan; +pub mod snapshot_cleanup; pub mod spec; pub mod table; pub mod transaction; @@ -100,3 +101,9 @@ pub use compact::{ // Re-export expression types pub use expr::{parse_filter, ColumnRef, ComparisonOp, Datum, Predicate}; + +// Re-export snapshot cleanup types +pub use snapshot_cleanup::{ + execute_snapshot_cleanup, plan_snapshot_cleanup, CleanupOptions, CleanupPlan, CleanupResult, + RetainedSnapshot, RetentionReason, SnapshotInfo, +}; diff --git a/src/snapshot_cleanup/mod.rs b/src/snapshot_cleanup/mod.rs new file mode 100644 index 0000000..df88687 --- /dev/null +++ b/src/snapshot_cleanup/mod.rs @@ -0,0 +1,486 @@ +//! Snapshot cleanup and expiration for Iceberg tables +//! +//! This module provides functionality to expire old table snapshots based on +//! configurable retention policies. Snapshot expiration helps reduce metadata +//! overhead, improve table operations, and decrease storage costs. +//! +//! # Retention Policy +//! +//! Snapshot expiration uses two parameters: +//! - `older_than_days`: Age threshold in days +//! - `retain_last`: Minimum snapshot count to always retain +//! +//! Both conditions must be met before a snapshot is expired, ensuring you +//! always retain recent snapshots even if they exceed the age threshold. +//! +//! # Example +//! +//! ```no_run +//! use icepick::snapshot_cleanup::{plan_snapshot_cleanup, CleanupOptions}; +//! use icepick::catalog::Catalog; +//! +//! # async fn example() -> Result<(), Box> { +//! # let catalog: icepick::R2Catalog = todo!(); +//! # let table_ident: icepick::TableIdent = todo!(); +//! let table = catalog.load_table(&table_ident).await?; +//! +//! // Plan cleanup: expire snapshots older than 7 days, keep at least 10 +//! let options = CleanupOptions::new() +//! .with_older_than_days(7) +//! .with_retain_last(10); +//! +//! let plan = plan_snapshot_cleanup(&table, &options)?; +//! +//! // Preview what would be removed +//! println!("Would remove {} snapshots", plan.snapshots_to_remove.len()); +//! +//! // Execute the cleanup +//! // let result = execute_snapshot_cleanup(&table, &catalog, plan).await?; +//! # Ok(()) +//! # } +//! ``` + +use crate::catalog::Catalog; +use crate::error::{Error, Result}; +use crate::spec::{Snapshot, TableMetadata}; +use crate::table::Table; +use serde::{Deserialize, Serialize}; +use std::collections::HashSet; + +/// Options for snapshot cleanup operations +#[derive(Debug, Clone)] +pub struct CleanupOptions { + /// Minimum age in days before a snapshot can be expired + older_than_days: u32, + /// Minimum number of snapshots to always retain (most recent) + retain_last: usize, + /// Whether this is a dry run (no actual changes) + dry_run: bool, +} + +impl Default for CleanupOptions { + fn default() -> Self { + Self { + older_than_days: 7, + retain_last: 10, + dry_run: false, + } + } +} + +impl CleanupOptions { + /// Create new cleanup options with default values + pub fn new() -> Self { + Self::default() + } + + /// Set the minimum age in days before a snapshot can be expired + pub fn with_older_than_days(mut self, days: u32) -> Self { + self.older_than_days = days; + self + } + + /// Set the minimum number of snapshots to always retain + pub fn with_retain_last(mut self, count: usize) -> Self { + self.retain_last = count; + self + } + + /// Set whether this is a dry run + pub fn with_dry_run(mut self, dry_run: bool) -> Self { + self.dry_run = dry_run; + self + } + + /// Get the older than days threshold + pub fn older_than_days(&self) -> u32 { + self.older_than_days + } + + /// Get the retain last count + pub fn retain_last(&self) -> usize { + self.retain_last + } + + /// Check if this is a dry run + pub fn dry_run(&self) -> bool { + self.dry_run + } +} + +/// Information about a snapshot for cleanup planning +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct SnapshotInfo { + /// Snapshot ID + pub snapshot_id: i64, + /// Timestamp when snapshot was created (milliseconds since epoch) + pub timestamp_ms: i64, + /// Parent snapshot ID if any + pub parent_snapshot_id: Option, + /// Age in days + pub age_days: f64, + /// Operation that created this snapshot + pub operation: String, + /// Whether this is the current snapshot + pub is_current: bool, + /// Branch or tag names referencing this snapshot + pub refs: Vec, +} + +impl SnapshotInfo { + fn from_snapshot( + snapshot: &Snapshot, + current_snapshot_id: Option, + refs: Vec, + now_ms: i64, + ) -> Self { + let age_ms = now_ms - snapshot.timestamp_ms(); + let age_days = age_ms as f64 / (24.0 * 60.0 * 60.0 * 1000.0); + + Self { + snapshot_id: snapshot.snapshot_id(), + timestamp_ms: snapshot.timestamp_ms(), + parent_snapshot_id: snapshot.parent_snapshot_id(), + age_days, + operation: snapshot.summary().operation().to_string(), + is_current: current_snapshot_id == Some(snapshot.snapshot_id()), + refs, + } + } +} + +/// Reason why a snapshot is retained +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +pub enum RetentionReason { + /// Snapshot is the current table state + CurrentSnapshot, + /// Snapshot is within the retain_last count + WithinRetainCount, + /// Snapshot is not old enough to expire + NotOldEnough, + /// Snapshot is referenced by a branch or tag + ReferencedByRef(String), +} + +/// A snapshot that will be retained with reason +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct RetainedSnapshot { + /// Snapshot information + pub info: SnapshotInfo, + /// Reason for retention + pub reason: RetentionReason, +} + +/// Plan for snapshot cleanup +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct CleanupPlan { + /// Snapshots that will be removed + pub snapshots_to_remove: Vec, + /// Snapshots that will be retained with reasons + pub snapshots_to_retain: Vec, + /// Total snapshot count before cleanup + pub total_snapshots: usize, + /// Options used for this plan + pub older_than_days: u32, + pub retain_last: usize, +} + +impl CleanupPlan { + /// Check if the plan has any snapshots to remove + pub fn is_empty(&self) -> bool { + self.snapshots_to_remove.is_empty() + } + + /// Get the number of snapshots that will be removed + pub fn removal_count(&self) -> usize { + self.snapshots_to_remove.len() + } + + /// Get the number of snapshots that will be retained + pub fn retention_count(&self) -> usize { + self.snapshots_to_retain.len() + } +} + +/// Result of snapshot cleanup execution +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct CleanupResult { + /// Number of snapshots removed + pub snapshots_removed: usize, + /// Number of snapshots retained + pub snapshots_retained: usize, + /// IDs of removed snapshots + pub removed_snapshot_ids: Vec, + /// Manifest list files that can be garbage collected + pub orphaned_manifest_lists: Vec, +} + +/// Plan which snapshots to cleanup based on the retention policy +/// +/// This function determines which snapshots can be safely expired without +/// affecting the current table state or any referenced branches/tags. +pub fn plan_snapshot_cleanup(table: &Table, options: &CleanupOptions) -> Result { + let metadata = table.metadata(); + let snapshots = metadata.snapshots(); + + if snapshots.is_empty() { + return Ok(CleanupPlan { + snapshots_to_remove: vec![], + snapshots_to_retain: vec![], + total_snapshots: 0, + older_than_days: options.older_than_days, + retain_last: options.retain_last, + }); + } + + let now_ms = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map_err(|e| Error::unexpected(format!("Failed to get current time: {}", e)))? + .as_millis() as i64; + + let current_snapshot_id = metadata.current_snapshot_id(); + let age_threshold_ms = (options.older_than_days as i64) * 24 * 60 * 60 * 1000; + + // Build a map of snapshot_id -> ref names + let mut snapshot_refs: std::collections::HashMap> = + std::collections::HashMap::new(); + for (ref_name, snapshot_ref) in metadata.refs() { + snapshot_refs + .entry(snapshot_ref.snapshot_id()) + .or_default() + .push(ref_name.clone()); + } + + // Convert snapshots to SnapshotInfo and sort by timestamp (newest first) + let mut snapshot_infos: Vec = snapshots + .iter() + .map(|s| { + SnapshotInfo::from_snapshot( + s, + current_snapshot_id, + snapshot_refs.get(&s.snapshot_id()).cloned().unwrap_or_default(), + now_ms, + ) + }) + .collect(); + + // Sort by timestamp descending (newest first) + snapshot_infos.sort_by(|a, b| b.timestamp_ms.cmp(&a.timestamp_ms)); + + let mut snapshots_to_remove = Vec::new(); + let mut snapshots_to_retain = Vec::new(); + + // Track which snapshots are in the "retain last N" window + let retain_last_ids: HashSet = snapshot_infos + .iter() + .take(options.retain_last) + .map(|s| s.snapshot_id) + .collect(); + + for info in snapshot_infos { + // Determine if this snapshot should be retained and why + let retention_reason = if info.is_current { + Some(RetentionReason::CurrentSnapshot) + } else if !info.refs.is_empty() { + Some(RetentionReason::ReferencedByRef(info.refs.join(", "))) + } else if retain_last_ids.contains(&info.snapshot_id) { + Some(RetentionReason::WithinRetainCount) + } else if (now_ms - info.timestamp_ms) < age_threshold_ms { + Some(RetentionReason::NotOldEnough) + } else { + None + }; + + if let Some(reason) = retention_reason { + snapshots_to_retain.push(RetainedSnapshot { info, reason }); + } else { + snapshots_to_remove.push(info); + } + } + + Ok(CleanupPlan { + total_snapshots: snapshots.len(), + snapshots_to_remove, + snapshots_to_retain, + older_than_days: options.older_than_days, + retain_last: options.retain_last, + }) +} + +/// Execute the snapshot cleanup plan +/// +/// This removes the specified snapshots from the table metadata and commits +/// the changes to the catalog. +pub async fn execute_snapshot_cleanup( + table: &Table, + catalog: &C, + plan: CleanupPlan, +) -> Result { + if plan.is_empty() { + return Ok(CleanupResult { + snapshots_removed: 0, + snapshots_retained: plan.snapshots_to_retain.len(), + removed_snapshot_ids: vec![], + orphaned_manifest_lists: vec![], + }); + } + + let snapshot_ids_to_remove: HashSet = plan + .snapshots_to_remove + .iter() + .map(|s| s.snapshot_id) + .collect(); + + // Get manifest lists that will become orphaned + let orphaned_manifest_lists: Vec = table + .metadata() + .snapshots() + .iter() + .filter(|s| snapshot_ids_to_remove.contains(&s.snapshot_id())) + .map(|s| s.manifest_list().to_string()) + .collect(); + + // Build new metadata without the expired snapshots + let old_metadata = table.metadata(); + let new_snapshots: Vec = old_metadata + .snapshots() + .iter() + .filter(|s| !snapshot_ids_to_remove.contains(&s.snapshot_id())) + .cloned() + .collect(); + + // Filter snapshot log as well + let new_snapshot_log: Vec = old_metadata + .snapshot_log() + .iter() + .filter(|entry| !snapshot_ids_to_remove.contains(&entry.snapshot_id())) + .cloned() + .collect(); + + // Build the updated metadata + let mut builder = TableMetadata::builder() + .with_format_version(old_metadata.format_version()) + .with_table_uuid(old_metadata.table_uuid().to_string()) + .with_location(old_metadata.location()) + .with_last_updated_ms( + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map_err(|e| Error::unexpected(format!("Failed to get current time: {}", e)))? + .as_millis() as i64, + ) + .with_last_sequence_number(old_metadata.last_sequence_number()) + .with_snapshot_log(new_snapshot_log) + .with_metadata_log(old_metadata.metadata_log().to_vec()) + .with_partition_specs(old_metadata.partition_specs().to_vec()) + .with_sort_orders(old_metadata.sort_orders().to_vec()) + .with_refs(old_metadata.refs().clone()) + .with_table_features(old_metadata.table_features().to_vec()); + + // Add all schemas + for schema in old_metadata.schemas() { + builder = builder.with_current_schema(schema.clone()); + } + + // Add all properties + for (key, value) in old_metadata.properties() { + builder = builder.with_property(key.clone(), value.clone()); + } + + // Add the retained snapshots + for snapshot in new_snapshots { + // We need to check if this is the current snapshot + if old_metadata.current_snapshot_id() == Some(snapshot.snapshot_id()) { + builder = builder.with_current_snapshot(snapshot); + } else { + // For non-current snapshots, we need a different approach + // The builder only has with_current_snapshot, so we need to + // handle this through the metadata directly after build + } + } + + // This is a simplified version - in production you'd want to use + // the REST catalog's remove-snapshots update type directly + let removed_ids: Vec = plan + .snapshots_to_remove + .iter() + .map(|s| s.snapshot_id) + .collect(); + + // Commit via catalog + catalog + .expire_snapshots(table.identifier(), &removed_ids) + .await?; + + Ok(CleanupResult { + snapshots_removed: plan.snapshots_to_remove.len(), + snapshots_retained: plan.snapshots_to_retain.len(), + removed_snapshot_ids: removed_ids, + orphaned_manifest_lists, + }) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::spec::{Snapshot, Summary}; + + fn create_test_snapshot(id: i64, timestamp_ms: i64, parent_id: Option) -> Snapshot { + let manifest_list = format!("s3://bucket/metadata/snap-{}.avro", id); + let mut builder = Snapshot::builder() + .with_snapshot_id(id) + .with_timestamp_ms(timestamp_ms) + .with_manifest_list(&manifest_list) + .with_summary(Summary::builder().set("operation", "append").build()); + + if let Some(parent) = parent_id { + builder = builder.with_parent_snapshot_id(parent); + } + + builder.build().unwrap() + } + + #[test] + fn test_cleanup_options_defaults() { + let options = CleanupOptions::new(); + assert_eq!(options.older_than_days(), 7); + assert_eq!(options.retain_last(), 10); + assert!(!options.dry_run()); + } + + #[test] + fn test_cleanup_options_builder() { + let options = CleanupOptions::new() + .with_older_than_days(14) + .with_retain_last(5) + .with_dry_run(true); + + assert_eq!(options.older_than_days(), 14); + assert_eq!(options.retain_last(), 5); + assert!(options.dry_run()); + } + + #[test] + fn test_snapshot_info_age_calculation() { + let now_ms = 1700000000000i64; // Some timestamp + let one_day_ago_ms = now_ms - (24 * 60 * 60 * 1000); + + let snapshot = create_test_snapshot(1, one_day_ago_ms, None); + let info = SnapshotInfo::from_snapshot(&snapshot, None, vec![], now_ms); + + assert!((info.age_days - 1.0).abs() < 0.01); + assert!(!info.is_current); + assert!(info.refs.is_empty()); + } + + #[test] + fn test_snapshot_info_current_flag() { + let now_ms = 1700000000000i64; + let snapshot = create_test_snapshot(42, now_ms - 1000, None); + + let info = SnapshotInfo::from_snapshot(&snapshot, Some(42), vec![], now_ms); + assert!(info.is_current); + + let info2 = SnapshotInfo::from_snapshot(&snapshot, Some(99), vec![], now_ms); + assert!(!info2.is_current); + } +} From 4c63519f6d498381ebc39f4252361378a2ed667f Mon Sep 17 00:00:00 2001 From: Claude Date: Sun, 18 Jan 2026 04:53:14 +0000 Subject: [PATCH 2/4] chore: cleanup and format snapshot_cleanup code - Remove unused TableMetadata import - Remove dead code in execute_snapshot_cleanup that built metadata but was never used - Apply rustfmt formatting --- src/cli/commands/snapshot.rs | 7 +--- src/snapshot_cleanup/mod.rs | 73 ++++-------------------------------- 2 files changed, 9 insertions(+), 71 deletions(-) diff --git a/src/cli/commands/snapshot.rs b/src/cli/commands/snapshot.rs index 0d992ba..7538c62 100644 --- a/src/cli/commands/snapshot.rs +++ b/src/cli/commands/snapshot.rs @@ -175,12 +175,7 @@ impl Outputable for CleanupPlanOutput { if !self.snapshots_to_retain.is_empty() { let mut retain_table = ComfyTable::new(); - retain_table.set_header(Row::from(vec![ - "Snapshot ID", - "Timestamp", - "Age", - "Reason", - ])); + retain_table.set_header(Row::from(vec!["Snapshot ID", "Timestamp", "Age", "Reason"])); for snapshot in &self.snapshots_to_retain { let age_str = if snapshot.age_days < 1.0 { diff --git a/src/snapshot_cleanup/mod.rs b/src/snapshot_cleanup/mod.rs index df88687..9dbe06a 100644 --- a/src/snapshot_cleanup/mod.rs +++ b/src/snapshot_cleanup/mod.rs @@ -42,7 +42,7 @@ use crate::catalog::Catalog; use crate::error::{Error, Result}; -use crate::spec::{Snapshot, TableMetadata}; +use crate::spec::Snapshot; use crate::table::Table; use serde::{Deserialize, Serialize}; use std::collections::HashSet; @@ -258,7 +258,10 @@ pub fn plan_snapshot_cleanup(table: &Table, options: &CleanupOptions) -> Result< SnapshotInfo::from_snapshot( s, current_snapshot_id, - snapshot_refs.get(&s.snapshot_id()).cloned().unwrap_or_default(), + snapshot_refs + .get(&s.snapshot_id()) + .cloned() + .unwrap_or_default(), now_ms, ) }) @@ -310,7 +313,7 @@ pub fn plan_snapshot_cleanup(table: &Table, options: &CleanupOptions) -> Result< /// Execute the snapshot cleanup plan /// /// This removes the specified snapshots from the table metadata and commits -/// the changes to the catalog. +/// the changes to the catalog using the REST API's remove-snapshots update. pub async fn execute_snapshot_cleanup( table: &Table, catalog: &C, @@ -331,7 +334,7 @@ pub async fn execute_snapshot_cleanup( .map(|s| s.snapshot_id) .collect(); - // Get manifest lists that will become orphaned + // Get manifest lists that will become orphaned (for garbage collection info) let orphaned_manifest_lists: Vec = table .metadata() .snapshots() @@ -340,73 +343,13 @@ pub async fn execute_snapshot_cleanup( .map(|s| s.manifest_list().to_string()) .collect(); - // Build new metadata without the expired snapshots - let old_metadata = table.metadata(); - let new_snapshots: Vec = old_metadata - .snapshots() - .iter() - .filter(|s| !snapshot_ids_to_remove.contains(&s.snapshot_id())) - .cloned() - .collect(); - - // Filter snapshot log as well - let new_snapshot_log: Vec = old_metadata - .snapshot_log() - .iter() - .filter(|entry| !snapshot_ids_to_remove.contains(&entry.snapshot_id())) - .cloned() - .collect(); - - // Build the updated metadata - let mut builder = TableMetadata::builder() - .with_format_version(old_metadata.format_version()) - .with_table_uuid(old_metadata.table_uuid().to_string()) - .with_location(old_metadata.location()) - .with_last_updated_ms( - std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) - .map_err(|e| Error::unexpected(format!("Failed to get current time: {}", e)))? - .as_millis() as i64, - ) - .with_last_sequence_number(old_metadata.last_sequence_number()) - .with_snapshot_log(new_snapshot_log) - .with_metadata_log(old_metadata.metadata_log().to_vec()) - .with_partition_specs(old_metadata.partition_specs().to_vec()) - .with_sort_orders(old_metadata.sort_orders().to_vec()) - .with_refs(old_metadata.refs().clone()) - .with_table_features(old_metadata.table_features().to_vec()); - - // Add all schemas - for schema in old_metadata.schemas() { - builder = builder.with_current_schema(schema.clone()); - } - - // Add all properties - for (key, value) in old_metadata.properties() { - builder = builder.with_property(key.clone(), value.clone()); - } - - // Add the retained snapshots - for snapshot in new_snapshots { - // We need to check if this is the current snapshot - if old_metadata.current_snapshot_id() == Some(snapshot.snapshot_id()) { - builder = builder.with_current_snapshot(snapshot); - } else { - // For non-current snapshots, we need a different approach - // The builder only has with_current_snapshot, so we need to - // handle this through the metadata directly after build - } - } - - // This is a simplified version - in production you'd want to use - // the REST catalog's remove-snapshots update type directly let removed_ids: Vec = plan .snapshots_to_remove .iter() .map(|s| s.snapshot_id) .collect(); - // Commit via catalog + // Commit via catalog's expire_snapshots which uses REST API's remove-snapshots catalog .expire_snapshots(table.identifier(), &removed_ids) .await?; From fb670f9f691e9f46e765025b8859a500ed12f80a Mon Sep 17 00:00:00 2001 From: Claude Date: Sun, 18 Jan 2026 04:55:55 +0000 Subject: [PATCH 3/4] refactor: remove dead code and DRY up snapshot cleanup - Remove unused `dry_run` field from CleanupOptions (CLI handles it) - Remove unused `parent_snapshot_id` field from SnapshotInfo - Remove unused `removal_count()` and `retention_count()` methods - Extract `format_age()` helper to DRY age formatting logic - Update tests to match simplified API --- src/cli/commands/snapshot.rs | 27 +++++++++++---------------- src/snapshot_cleanup/mod.rs | 32 +------------------------------- 2 files changed, 12 insertions(+), 47 deletions(-) diff --git a/src/cli/commands/snapshot.rs b/src/cli/commands/snapshot.rs index 7538c62..9712e53 100644 --- a/src/cli/commands/snapshot.rs +++ b/src/cli/commands/snapshot.rs @@ -77,16 +77,10 @@ impl Outputable for SnapshotList { ])); for snapshot in &self.snapshots { - let age_str = if snapshot.age_days < 1.0 { - format!("{:.1}h", snapshot.age_days * 24.0) - } else { - format!("{:.1}d", snapshot.age_days) - }; - table.add_row(Row::from(vec![ snapshot.snapshot_id.to_string(), snapshot.timestamp.clone(), - age_str, + format_age(snapshot.age_days), snapshot.operation.clone(), if snapshot.is_current { "yes" } else { "" }.to_string(), snapshot.refs.join(", "), @@ -178,16 +172,10 @@ impl Outputable for CleanupPlanOutput { retain_table.set_header(Row::from(vec!["Snapshot ID", "Timestamp", "Age", "Reason"])); for snapshot in &self.snapshots_to_retain { - let age_str = if snapshot.age_days < 1.0 { - format!("{:.1}h", snapshot.age_days * 24.0) - } else { - format!("{:.1}d", snapshot.age_days) - }; - retain_table.add_row(Row::from(vec![ snapshot.snapshot_id.to_string(), snapshot.timestamp.clone(), - age_str, + format_age(snapshot.age_days), snapshot.reason.clone(), ])); } @@ -260,6 +248,14 @@ fn format_timestamp(timestamp_ms: i64) -> String { .unwrap_or_else(|| "Invalid timestamp".to_string()) } +fn format_age(age_days: f64) -> String { + if age_days < 1.0 { + format!("{:.1}h", age_days * 24.0) + } else { + format!("{:.1}d", age_days) + } +} + fn format_retention_reason(reason: &RetentionReason) -> String { match reason { RetentionReason::CurrentSnapshot => "Current snapshot".to_string(), @@ -387,8 +383,7 @@ pub async fn execute( // Build cleanup options let options = CleanupOptions::new() .with_older_than_days(older_than_days) - .with_retain_last(retain_last) - .with_dry_run(dry_run); + .with_retain_last(retain_last); // Create cleanup plan let plan = plan_snapshot_cleanup(&table, &options) diff --git a/src/snapshot_cleanup/mod.rs b/src/snapshot_cleanup/mod.rs index 9dbe06a..b79f62e 100644 --- a/src/snapshot_cleanup/mod.rs +++ b/src/snapshot_cleanup/mod.rs @@ -54,8 +54,6 @@ pub struct CleanupOptions { older_than_days: u32, /// Minimum number of snapshots to always retain (most recent) retain_last: usize, - /// Whether this is a dry run (no actual changes) - dry_run: bool, } impl Default for CleanupOptions { @@ -63,7 +61,6 @@ impl Default for CleanupOptions { Self { older_than_days: 7, retain_last: 10, - dry_run: false, } } } @@ -86,12 +83,6 @@ impl CleanupOptions { self } - /// Set whether this is a dry run - pub fn with_dry_run(mut self, dry_run: bool) -> Self { - self.dry_run = dry_run; - self - } - /// Get the older than days threshold pub fn older_than_days(&self) -> u32 { self.older_than_days @@ -101,11 +92,6 @@ impl CleanupOptions { pub fn retain_last(&self) -> usize { self.retain_last } - - /// Check if this is a dry run - pub fn dry_run(&self) -> bool { - self.dry_run - } } /// Information about a snapshot for cleanup planning @@ -115,8 +101,6 @@ pub struct SnapshotInfo { pub snapshot_id: i64, /// Timestamp when snapshot was created (milliseconds since epoch) pub timestamp_ms: i64, - /// Parent snapshot ID if any - pub parent_snapshot_id: Option, /// Age in days pub age_days: f64, /// Operation that created this snapshot @@ -140,7 +124,6 @@ impl SnapshotInfo { Self { snapshot_id: snapshot.snapshot_id(), timestamp_ms: snapshot.timestamp_ms(), - parent_snapshot_id: snapshot.parent_snapshot_id(), age_days, operation: snapshot.summary().operation().to_string(), is_current: current_snapshot_id == Some(snapshot.snapshot_id()), @@ -190,16 +173,6 @@ impl CleanupPlan { pub fn is_empty(&self) -> bool { self.snapshots_to_remove.is_empty() } - - /// Get the number of snapshots that will be removed - pub fn removal_count(&self) -> usize { - self.snapshots_to_remove.len() - } - - /// Get the number of snapshots that will be retained - pub fn retention_count(&self) -> usize { - self.snapshots_to_retain.len() - } } /// Result of snapshot cleanup execution @@ -387,19 +360,16 @@ mod tests { let options = CleanupOptions::new(); assert_eq!(options.older_than_days(), 7); assert_eq!(options.retain_last(), 10); - assert!(!options.dry_run()); } #[test] fn test_cleanup_options_builder() { let options = CleanupOptions::new() .with_older_than_days(14) - .with_retain_last(5) - .with_dry_run(true); + .with_retain_last(5); assert_eq!(options.older_than_days(), 14); assert_eq!(options.retain_last(), 5); - assert!(options.dry_run()); } #[test] From e0545b2a3b990cd077c48d4725048001b851a542 Mon Sep 17 00:00:00 2001 From: Clay Smith Date: Sun, 18 Jan 2026 10:27:46 -0800 Subject: [PATCH 4/4] docs: add snapshot cleanup documentation Document the new snapshot cleanup feature in AGENTS.md and README.md: - CLI commands (snapshot list, snapshot cleanup) in quick start - New public API items (plan_snapshot_cleanup, execute_snapshot_cleanup) - Pattern 8 code example for snapshot cleanup workflow - Performance profile and comparison matrix entries - Full README section with Rust example and CLI usage Co-Authored-By: Claude Opus 4.5 --- AGENTS.md | 49 +++++++++++++++++++++++++++++++++++++++++++--- README.md | 58 +++++++++++++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 104 insertions(+), 3 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index 75de6be..1cdbf8a 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -83,6 +83,11 @@ icepick table scan my_namespace.my_table --filter "date >= '2024-01-01'" # Compact small files (dry run first) icepick compact my_namespace.my_table --dry-run icepick compact my_namespace.my_table --target-size 268435456 + +# Snapshot management +icepick snapshot list my_namespace.my_table +icepick snapshot cleanup my_namespace.my_table --dry-run +icepick snapshot cleanup my_namespace.my_table --older-than-days 7 --retain-last 10 ``` ## CORE CONCEPTS @@ -104,6 +109,7 @@ Module Structure: ├── cli/ # CLI commands (native only, behind "cli" feature) │ └── commands/ # catalog, namespace, table, compact subcommands ├── compact/ # Bin-pack compaction for small files +├── snapshot_cleanup/ # Snapshot expiration and cleanup ├── expr/ # Predicate expressions for partition pruning ├── spec/ # Iceberg specification types (Schema, TableIdent, etc.) ├── table/ # Table representation and operations @@ -132,6 +138,9 @@ Module Structure: 12. **arrow_to_parquet()** - Write Arrow data directly to S3 without Iceberg metadata 13. **register_data_files()** - Register existing Parquet files without rewriting data 14. **introspect_parquet_file()** - Extract schema, row count, and metrics from Parquet footer +15. **plan_snapshot_cleanup()** - Plan which snapshots to expire based on retention policy +16. **execute_snapshot_cleanup()** - Execute a cleanup plan to remove expired snapshots +17. **CleanupOptions** - Configure snapshot retention (older_than_days, retain_last) ## COMMON PATTERNS @@ -326,6 +335,34 @@ let result = compact_table(&table, &catalog, &options).await?; println!("Compacted {} files into {}", result.files_removed, result.files_added); ``` +### Pattern 8: Snapshot cleanup + +```rust +use icepick::snapshot_cleanup::{plan_snapshot_cleanup, execute_snapshot_cleanup, CleanupOptions}; + +let table = catalog.load_table(&table_id).await?; + +// Configure cleanup options +let options = CleanupOptions::new() + .with_older_than_days(7) // Expire snapshots older than 7 days + .with_retain_last(10); // Always keep at least 10 most recent + +// Option A: Dry run - see what would be expired +let plan = plan_snapshot_cleanup(&table, &options)?; +println!("Would remove {} of {} snapshots", + plan.snapshots_to_remove.len(), plan.total_snapshots); + +for snapshot in &plan.snapshots_to_remove { + println!(" Remove: {} ({:.1} days old)", snapshot.snapshot_id, snapshot.age_days); +} + +// Option B: Execute cleanup +if !plan.snapshots_to_remove.is_empty() { + let result = execute_snapshot_cleanup(&table, &catalog, plan).await?; + println!("Removed {} snapshots", result.snapshots_removed); +} +``` + ## INTEGRATION POINTS - **Async Runtime**: tokio (required for examples/tests, not enforced as dependency) @@ -463,7 +500,8 @@ When working with this library: 5. Error pattern: All errors implement Display with context - use `?` operator and let errors propagate 6. Use predicates for scan filtering: `table.scan().filter(predicate).build()?` 7. Compaction is available via `compact_table()` or `plan_compaction()` + `execute_compaction()` -8. CLI is behind the `cli` feature flag (native only) +8. Snapshot cleanup via `plan_snapshot_cleanup()` + `execute_snapshot_cleanup()` +9. CLI is behind the `cli` feature flag (native only) ### Key Invariants to Maintain @@ -481,7 +519,8 @@ When working with this library: - Add field IDs to Iceberg schemas (required for Parquet field mapping) - Use `parse_filter()` for user-provided filter strings; use `Predicate::*` for programmatic filters - Call `plan_compaction()` with `dry_run` first to preview changes before `compact_table()` -- Use `CompactOptions::with_*()` builder pattern (methods return `Result`) +- Call `plan_snapshot_cleanup()` first to preview before `execute_snapshot_cleanup()` +- Use `CompactOptions::with_*()` and `CleanupOptions::with_*()` builder patterns **Never:** - Construct `Table` directly (use catalog methods) @@ -490,6 +529,7 @@ When working with this library: - Assume tables have snapshots (check with `current_snapshot()`) - Hardcode credentials in examples (use env vars or function parameters) - Run compaction without checking `plan.is_empty()` first +- Run snapshot cleanup without checking `plan.snapshots_to_remove.is_empty()` first - Use CLI features in WASM builds (cli module is `#[cfg(not(target_family = "wasm"))]`) ## PERFORMANCE PROFILE @@ -505,8 +545,10 @@ When working with this library: | `plan_compaction()` | O(m) | Reads manifests and groups small files | | `execute_compaction()` | O(g×f) | Reads/writes g groups × f files per group | | `arrow_to_parquet()` | O(n) | Full buffer in memory before upload | +| `plan_snapshot_cleanup()` | O(s) | Iterates snapshots and refs | +| `execute_snapshot_cleanup()` | O(1) | Single REST API call to update metadata | -Where m = number of manifest files, n = number of data files, k = files after pruning +Where m = number of manifest files, n = number of data files, k = files after pruning, s = number of snapshots ## COMPARISON MATRIX @@ -520,6 +562,7 @@ Where m = number of manifest files, n = number of data files, k = files after pr | Transaction API | Simplified (append only) | Full (delete, overwrite, etc.) | | Query Optimization | Partition/bounds pruning | Predicate pushdown, projection | | Compaction | ✅ Bin-pack | ✅ Multiple strategies | +| Snapshot Cleanup | ✅ Automatic expiration | ✅ expire_snapshots API | | CLI Tool | ✅ icepick binary | ❌ | **When to use icepick**: WASM deployment, serverless environments (Cloudflare Workers), simpler API for append-only workloads, R2 Data Catalog support, CLI-based table maintenance diff --git a/README.md b/README.md index 4e4b675..eb1eaae 100644 --- a/README.md +++ b/README.md @@ -23,6 +23,11 @@ - **Generic REST Catalog** — Build clients for any Iceberg REST endpoint (Nessie, Glue REST, custom) - **Direct S3 Parquet Writes** — Write Arrow data directly to S3 without Iceberg metadata +### Table Maintenance +- **Bin-pack Compaction** — Merge small files into larger ones for better query performance +- **Snapshot Cleanup** — Automatically expire old snapshots based on retention policies +- **Partition Pruning** — Filter scans by partition values and column statistics + ### Developer Experience - **Clean API** — Simple factory methods, no complex builders - **Type-safe errors** — Comprehensive error handling with context @@ -243,6 +248,59 @@ This is useful for: - Registering files written by external tools (Spark, DuckDB, etc.) - "Write to S3, register later" workflows in serverless environments +## Snapshot Cleanup + +Automatically expire old snapshots to reduce metadata overhead and storage costs: + +```rust +use icepick::{R2Catalog, snapshot_cleanup::{plan_snapshot_cleanup, execute_snapshot_cleanup, CleanupOptions}}; + +#[tokio::main] +async fn main() -> Result<(), Box> { + let catalog = R2Catalog::new("my-catalog", "account-id", "bucket", "token").await?; + let table = catalog.load_table(&"namespace.table".parse()?).await?; + + // Configure retention policy + let options = CleanupOptions::new() + .with_older_than_days(7) // Expire snapshots older than 7 days + .with_retain_last(10); // Always keep at least 10 most recent + + // Preview what would be removed + let plan = plan_snapshot_cleanup(&table, &options)?; + println!("Will remove {} of {} snapshots", + plan.snapshots_to_remove.len(), plan.total_snapshots); + + // Execute cleanup + if !plan.snapshots_to_remove.is_empty() { + let result = execute_snapshot_cleanup(&table, &catalog, plan).await?; + println!("Removed {} snapshots", result.snapshots_removed); + } + + Ok(()) +} +``` + +### CLI Usage + +```bash +# List all snapshots with age and status +icepick snapshot list my_namespace.my_table + +# Preview cleanup (dry run) +icepick snapshot cleanup my_namespace.my_table --dry-run + +# Execute cleanup with custom retention +icepick snapshot cleanup my_namespace.my_table \ + --older-than-days 7 \ + --retain-last 10 +``` + +Snapshot cleanup respects: +- **Current snapshot** - Never expired (it's the current table state) +- **Referenced snapshots** - Never expired if referenced by branches or tags +- **Retention count** - Keeps the N most recent regardless of age +- **Age threshold** - Only expires snapshots older than the threshold + ## Examples Explore complete working examples in the [`examples/`](examples/) directory: