feat: add snapshot cleanup command for automatic snapshot expiration

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

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<dyn std::error::Error>> {
+    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:

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,
     };
 

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",
+        ))
+    }
 }

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(())
+    }
+}

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
+    }
 }

src/catalog/rest/commit_types.rs

@@ -75,6 +75,13 @@ pub enum TableUpdate {
         max_ref_age_ms: Option<i64>,
     },
 
+    /// Remove snapshots by their IDs
+    #[serde(rename = "remove-snapshots")]
+    RemoveSnapshots {
+        #[serde(rename = "snapshot-ids")]
+        snapshot_ids: Vec<i64>,
+    },
+
     #[serde(rename = "upgrade-format-version")]
     UpgradeFormatVersion {
         #[serde(rename = "format-version")]

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)]

src/cli/commands/mod.rs

@@ -3,4 +3,5 @@
 pub mod catalog;
 pub mod compact;
 pub mod namespace;
+pub mod snapshot;
 pub mod table;