iceberg: add lifecycle maintenance parity options

weed/plugin/worker/iceberg/config.go

@@ -23,15 +23,18 @@ const (
 	defaultOperations             = "all"
 
 	// Metric keys returned by maintenance operations.
-	MetricFilesMerged        = "files_merged"
-	MetricFilesWritten       = "files_written"
-	MetricBins               = "bins"
-	MetricSnapshotsExpired   = "snapshots_expired"
-	MetricFilesDeleted       = "files_deleted"
-	MetricOrphansRemoved     = "orphans_removed"
-	MetricManifestsRewritten = "manifests_rewritten"
-	MetricEntriesTotal       = "entries_total"
-	MetricDurationMs         = "duration_ms"
+	MetricFilesMerged          = "files_merged"
+	MetricFilesWritten         = "files_written"
+	MetricBins                 = "bins"
+	MetricSnapshotsExpired     = "snapshots_expired"
+	MetricSnapshotRefsRemoved  = "snapshot_refs_removed"
+	MetricFilesDeleted         = "files_deleted"
+	MetricOrphansRemoved       = "orphans_removed"
+	MetricOrphansFound         = "orphans_found"
+	MetricMetadataFilesDeleted = "metadata_files_deleted"
+	MetricManifestsRewritten   = "manifests_rewritten"
+	MetricEntriesTotal         = "entries_total"
+	MetricDurationMs           = "duration_ms"
 )
 
 // Config holds parsed worker config values.
@@ -45,6 +48,8 @@ type Config struct {
 	MinManifestsToRewrite  int64
 	Operations             string
 	ApplyDeletes           bool
+	RemoveOrphansDryRun    bool
+	CleanExpiredMetadata   bool
 }
 
 // ParseConfig extracts an iceberg maintenance Config from plugin config values.
@@ -60,6 +65,8 @@ func ParseConfig(values map[string]*plugin_pb.ConfigValue) Config {
 		MinManifestsToRewrite:  readInt64Config(values, "min_manifests_to_rewrite", defaultMinManifestsToRewrite),
 		Operations:             readStringConfig(values, "operations", defaultOperations),
 		ApplyDeletes:           readBoolConfig(values, "apply_deletes", true),
+		RemoveOrphansDryRun:    readBoolConfig(values, "remove_orphans_dry_run", false),
+		CleanExpiredMetadata:   readBoolConfig(values, "clean_expired_metadata", false),
 	}
 
 	// Clamp to safe minimums using the default constants

weed/plugin/worker/iceberg/exec_test.go

@@ -263,24 +263,6 @@ func populateTable(t *testing.T, fs *fakeFilerServer, setup tableSetup) table.Me
 	t.Helper()
 
 	meta := buildTestMetadata(t, setup.Snapshots)
-	fullMetadataJSON, err := json.Marshal(meta)
-	if err != nil {
-		t.Fatalf("marshal metadata: %v", err)
-	}
-
-	// Build internal metadata xattr
-	const metadataVersion = 1
-	internalMeta := map[string]interface{}{
-		"metadataVersion":  metadataVersion,
-		"metadataLocation": path.Join("metadata", fmt.Sprintf("v%d.metadata.json", metadataVersion)),
-		"metadata": map[string]interface{}{
-			"fullMetadata": json.RawMessage(fullMetadataJSON),
-		},
-	}
-	xattr, err := json.Marshal(internalMeta)
-	if err != nil {
-		t.Fatalf("marshal xattr: %v", err)
-	}
 
 	bucketsPath := s3tables.TablesPath // "/buckets"
 	bucketPath := path.Join(bucketsPath, setup.BucketName)
@@ -306,10 +288,9 @@ func populateTable(t *testing.T, fs *fakeFilerServer, setup tableSetup) table.Me
 	fs.putEntry(nsPath, setup.TableName, &filer_pb.Entry{
 		Name:        setup.TableName,
 		IsDirectory: true,
-		Extended: map[string][]byte{
-			s3tables.ExtendedKeyMetadata: xattr,
-		},
+		Extended:    map[string][]byte{},
 	})
+	setTableMetadataForTest(t, fs, setup, meta, 1, path.Join("metadata", "v1.metadata.json"))
 
 	// Create metadata/ and data/ directory placeholders
 	metaDir := path.Join(tableFilerPath, "metadata")
@@ -398,6 +379,39 @@ func populateTable(t *testing.T, fs *fakeFilerServer, setup tableSetup) table.Me
 	return meta
 }
 
+func setTableMetadataForTest(t *testing.T, fs *fakeFilerServer, setup tableSetup, meta table.Metadata, metadataVersion int, metadataLocation string) {
+	t.Helper()
+
+	fullMetadataJSON, err := json.Marshal(meta)
+	if err != nil {
+		t.Fatalf("marshal metadata: %v", err)
+	}
+
+	internalMeta := map[string]interface{}{
+		"metadataVersion":  metadataVersion,
+		"metadataLocation": metadataLocation,
+		"metadata": map[string]interface{}{
+			"fullMetadata": json.RawMessage(fullMetadataJSON),
+		},
+	}
+	xattr, err := json.Marshal(internalMeta)
+	if err != nil {
+		t.Fatalf("marshal xattr: %v", err)
+	}
+
+	tableDir := path.Join(s3tables.TablesPath, setup.BucketName, setup.Namespace)
+	entry := fs.getEntry(tableDir, setup.TableName)
+	if entry == nil {
+		t.Fatalf("table entry not found: %s/%s", tableDir, setup.TableName)
+	}
+	updatedEntry := cloneEntryForTest(t, entry)
+	if updatedEntry.Extended == nil {
+		updatedEntry.Extended = make(map[string][]byte)
+	}
+	updatedEntry.Extended[s3tables.ExtendedKeyMetadata] = xattr
+	fs.putEntry(tableDir, setup.TableName, updatedEntry)
+}
+
 func writeCurrentSnapshotManifests(t *testing.T, fs *fakeFilerServer, setup tableSetup, meta table.Metadata, manifestEntries [][]iceberg.ManifestEntry) {
 	t.Helper()
 
@@ -571,6 +585,118 @@ func TestExpireSnapshotsNothingToExpire(t *testing.T) {
 	}
 }
 
+func TestExpireSnapshotsPreservesTaggedSnapshot(t *testing.T) {
+	fs, client := startFakeFiler(t)
+
+	now := time.Now().UnixMilli()
+	setup := tableSetup{
+		BucketName: "test-bucket",
+		Namespace:  "analytics",
+		TableName:  "events",
+		Snapshots: []table.Snapshot{
+			{SnapshotID: 1, TimestampMs: now - 3_000},
+			{SnapshotID: 2, TimestampMs: now - 2_000},
+			{SnapshotID: 3, TimestampMs: now - 1_000},
+		},
+	}
+	meta := populateTable(t, fs, setup)
+
+	builder, err := table.MetadataBuilderFromBase(meta, "s3://test-bucket/test-table")
+	if err != nil {
+		t.Fatalf("metadata builder: %v", err)
+	}
+	if err := builder.SetSnapshotRef("audit-tag", 1, table.TagRef, table.WithMaxRefAgeMs(24*3600*1000)); err != nil {
+		t.Fatalf("set snapshot ref: %v", err)
+	}
+	taggedMeta, err := builder.Build()
+	if err != nil {
+		t.Fatalf("build tagged metadata: %v", err)
+	}
+	setTableMetadataForTest(t, fs, setup, taggedMeta, 1, path.Join("metadata", "v1.metadata.json"))
+
+	handler := NewHandler(nil)
+	config := Config{
+		SnapshotRetentionHours: 0,
+		MaxSnapshotsToKeep:     1,
+		MaxCommitRetries:       3,
+	}
+
+	result, _, err := handler.expireSnapshots(context.Background(), client, setup.BucketName, setup.tablePath(), config)
+	if err != nil {
+		t.Fatalf("expireSnapshots failed: %v", err)
+	}
+	if !strings.Contains(result, "expired 1 snapshot") {
+		t.Fatalf("expected one snapshot to expire, got %q", result)
+	}
+
+	currentMeta, _, err := loadCurrentMetadata(context.Background(), client, setup.BucketName, setup.tablePath())
+	if err != nil {
+		t.Fatalf("loadCurrentMetadata: %v", err)
+	}
+	if currentMeta.SnapshotByID(1) == nil {
+		t.Fatal("expected tagged snapshot 1 to be preserved")
+	}
+	if currentMeta.SnapshotByID(2) != nil {
+		t.Fatal("expected unreferenced snapshot 2 to expire")
+	}
+	refs := snapshotRefs(currentMeta)
+	if ref, ok := refs["audit-tag"]; !ok || ref.SnapshotID != 1 {
+		t.Fatalf("expected audit-tag to remain on snapshot 1, got %+v", refs["audit-tag"])
+	}
+}
+
+func TestExpireSnapshotsCleansExpiredMetadata(t *testing.T) {
+	fs, client := startFakeFiler(t)
+
+	now := time.Now().UnixMilli()
+	setup := tableSetup{
+		BucketName: "test-bucket",
+		Namespace:  "analytics",
+		TableName:  "events",
+		Snapshots: []table.Snapshot{
+			{SnapshotID: 1, TimestampMs: now, ManifestList: "metadata/snap-1.avro"},
+		},
+	}
+	populateTable(t, fs, setup)
+
+	metaDir := path.Join(s3tables.TablesPath, setup.BucketName, setup.tablePath(), "metadata")
+	fs.putEntry(metaDir, "v1.metadata.json", &filer_pb.Entry{
+		Name:       "v1.metadata.json",
+		Attributes: &filer_pb.FuseAttributes{Mtime: time.Now().Add(-time.Hour).Unix()},
+		Content:    []byte(`{"current":true}`),
+	})
+	fs.putEntry(metaDir, "v0.metadata.json", &filer_pb.Entry{
+		Name:       "v0.metadata.json",
+		Attributes: &filer_pb.FuseAttributes{Mtime: time.Now().Add(-2 * time.Hour).Unix()},
+		Content:    []byte(`{"stale":true}`),
+	})
+
+	handler := NewHandler(nil)
+	config := Config{
+		SnapshotRetentionHours: 24 * 365,
+		MaxSnapshotsToKeep:     10,
+		MaxCommitRetries:       3,
+		CleanExpiredMetadata:   true,
+	}
+
+	result, metrics, err := handler.expireSnapshots(context.Background(), client, setup.BucketName, setup.tablePath(), config)
+	if err != nil {
+		t.Fatalf("expireSnapshots failed: %v", err)
+	}
+	if !strings.Contains(result, "deleted 1 expired metadata file") {
+		t.Fatalf("expected expired metadata cleanup, got %q", result)
+	}
+	if got := metrics[MetricMetadataFilesDeleted]; got != 1 {
+		t.Fatalf("expected one expired metadata file deleted, got %d", got)
+	}
+	if fs.getEntry(metaDir, "v0.metadata.json") != nil {
+		t.Fatal("expected stale metadata file to be deleted")
+	}
+	if fs.getEntry(metaDir, "v1.metadata.json") == nil {
+		t.Fatal("expected current metadata file to be preserved")
+	}
+}
+
 func TestRemoveOrphansExecution(t *testing.T) {
 	fs, client := startFakeFiler(t)
 
@@ -632,6 +758,52 @@ func TestRemoveOrphansExecution(t *testing.T) {
 	}
 }
 
+func TestRemoveOrphansDryRun(t *testing.T) {
+	fs, client := startFakeFiler(t)
+
+	now := time.Now().UnixMilli()
+	setup := tableSetup{
+		BucketName: "test-bucket",
+		Namespace:  "analytics",
+		TableName:  "events",
+		Snapshots: []table.Snapshot{
+			{SnapshotID: 1, TimestampMs: now, ManifestList: "metadata/snap-1.avro"},
+		},
+	}
+	populateTable(t, fs, setup)
+
+	dataDir := path.Join(s3tables.TablesPath, setup.BucketName, setup.tablePath(), "data")
+	oldTime := time.Now().Add(-200 * time.Hour).Unix()
+	fs.putEntry(dataDir, "orphan-data.parquet", &filer_pb.Entry{
+		Name:       "orphan-data.parquet",
+		Attributes: &filer_pb.FuseAttributes{Mtime: oldTime},
+	})
+
+	handler := NewHandler(nil)
+	config := Config{
+		OrphanOlderThanHours: 72,
+		MaxCommitRetries:     3,
+		RemoveOrphansDryRun:  true,
+	}
+
+	result, metrics, err := handler.removeOrphans(context.Background(), client, setup.BucketName, setup.tablePath(), config)
+	if err != nil {
+		t.Fatalf("removeOrphans failed: %v", err)
+	}
+	if !strings.Contains(result, "dry run") {
+		t.Fatalf("expected dry-run summary, got %q", result)
+	}
+	if got := metrics[MetricOrphansFound]; got != 1 {
+		t.Fatalf("expected one orphan found, got %d", got)
+	}
+	if got := metrics[MetricOrphansRemoved]; got != 0 {
+		t.Fatalf("expected zero orphan deletions, got %d", got)
+	}
+	if fs.getEntry(dataDir, "orphan-data.parquet") == nil {
+		t.Fatal("expected orphan file to remain during dry run")
+	}
+}
+
 func TestRemoveOrphansPreservesReferencedFiles(t *testing.T) {
 	fs, client := startFakeFiler(t)
 

weed/plugin/worker/iceberg/handler.go

@@ -129,6 +129,13 @@ func (h *Handler) Descriptor() *plugin_pb.JobTypeDescriptor {
 							Widget:      plugin_pb.ConfigWidget_CONFIG_WIDGET_NUMBER,
 							MinValue:    &plugin_pb.ConfigValue{Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 1}},
 						},
+						{
+							Name:        "clean_expired_metadata",
+							Label:       "Clean Expired Metadata",
+							Description: "When true, delete older unreferenced metadata JSON files after snapshot expiration planning finishes.",
+							FieldType:   plugin_pb.ConfigFieldType_CONFIG_FIELD_TYPE_BOOL,
+							Widget:      plugin_pb.ConfigWidget_CONFIG_WIDGET_TOGGLE,
+						},
 					},
 				},
 				{
@@ -174,6 +181,13 @@ func (h *Handler) Descriptor() *plugin_pb.JobTypeDescriptor {
 							Widget:      plugin_pb.ConfigWidget_CONFIG_WIDGET_NUMBER,
 							MinValue:    &plugin_pb.ConfigValue{Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 1}},
 						},
+						{
+							Name:        "remove_orphans_dry_run",
+							Label:       "Dry Run",
+							Description: "When true, report orphan files without deleting them.",
+							FieldType:   plugin_pb.ConfigFieldType_CONFIG_FIELD_TYPE_BOOL,
+							Widget:      plugin_pb.ConfigWidget_CONFIG_WIDGET_TOGGLE,
+						},
 					},
 				},
 				{
@@ -225,6 +239,8 @@ func (h *Handler) Descriptor() *plugin_pb.JobTypeDescriptor {
 				"max_commit_retries":       {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: defaultMaxCommitRetries}},
 				"operations":               {Kind: &plugin_pb.ConfigValue_StringValue{StringValue: defaultOperations}},
 				"apply_deletes":            {Kind: &plugin_pb.ConfigValue_BoolValue{BoolValue: true}},
+				"remove_orphans_dry_run":   {Kind: &plugin_pb.ConfigValue_BoolValue{BoolValue: false}},
+				"clean_expired_metadata":   {Kind: &plugin_pb.ConfigValue_BoolValue{BoolValue: false}},
 			},
 		},
 		AdminRuntimeDefaults: &plugin_pb.AdminRuntimeDefaults{
@@ -247,6 +263,8 @@ func (h *Handler) Descriptor() *plugin_pb.JobTypeDescriptor {
 			"max_commit_retries":       {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: defaultMaxCommitRetries}},
 			"operations":               {Kind: &plugin_pb.ConfigValue_StringValue{StringValue: defaultOperations}},
 			"apply_deletes":            {Kind: &plugin_pb.ConfigValue_BoolValue{BoolValue: true}},
+			"remove_orphans_dry_run":   {Kind: &plugin_pb.ConfigValue_BoolValue{BoolValue: false}},
+			"clean_expired_metadata":   {Kind: &plugin_pb.ConfigValue_BoolValue{BoolValue: false}},
 		},
 	}
 }

weed/plugin/worker/iceberg/handler_test.go

@@ -761,6 +761,27 @@ func TestParseConfigApplyDeletes(t *testing.T) {
 	}
 }
 
+func TestParseConfigLifecycleFlags(t *testing.T) {
+	config := ParseConfig(nil)
+	if config.RemoveOrphansDryRun {
+		t.Error("expected RemoveOrphansDryRun=false by default")
+	}
+	if config.CleanExpiredMetadata {
+		t.Error("expected CleanExpiredMetadata=false by default")
+	}
+
+	config = ParseConfig(map[string]*plugin_pb.ConfigValue{
+		"remove_orphans_dry_run": {Kind: &plugin_pb.ConfigValue_BoolValue{BoolValue: true}},
+		"clean_expired_metadata": {Kind: &plugin_pb.ConfigValue_StringValue{StringValue: "true"}},
+	})
+	if !config.RemoveOrphansDryRun {
+		t.Error("expected RemoveOrphansDryRun=true when explicitly set")
+	}
+	if !config.CleanExpiredMetadata {
+		t.Error("expected CleanExpiredMetadata=true when set via string 'true'")
+	}
+}
+
 func TestCollectPositionDeletes(t *testing.T) {
 	fs, client := startFakeFiler(t)
 

weed/plugin/worker/iceberg/operations.go

@@ -8,7 +8,6 @@ import (
 	"fmt"
 	"math/rand/v2"
 	"path"
-	"sort"
 	"strings"
 	"time"
 
@@ -51,49 +50,11 @@ func (h *Handler) expireSnapshots(
 		return "no snapshots", nil, nil
 	}
 
-	// Determine which snapshots to expire
-	currentSnap := meta.CurrentSnapshot()
-	var currentSnapID int64
-	if currentSnap != nil {
-		currentSnapID = currentSnap.SnapshotID
-	}
-
-	retentionMs := config.SnapshotRetentionHours * 3600 * 1000
-	nowMs := time.Now().UnixMilli()
-
-	// Sort snapshots by timestamp descending (most recent first) so that
-	// the keep-count logic always preserves the newest snapshots.
-	sorted := make([]table.Snapshot, len(snapshots))
-	copy(sorted, snapshots)
-	sort.Slice(sorted, func(i, j int) bool {
-		return sorted[i].TimestampMs > sorted[j].TimestampMs
-	})
-
-	// Walk from newest to oldest. The current snapshot is always kept.
-	// Among the remaining, keep up to MaxSnapshotsToKeep-1 (since current
-	// counts toward the quota). Expire the rest only if they exceed the
-	// retention window; snapshots within the window are kept regardless.
-	var toExpire []int64
-	var kept int64
-	for _, snap := range sorted {
-		if snap.SnapshotID == currentSnapID {
-			kept++
-			continue
-		}
-		age := nowMs - snap.TimestampMs
-		if kept < config.MaxSnapshotsToKeep {
-			kept++
-			continue
-		}
-		if age > retentionMs {
-			toExpire = append(toExpire, snap.SnapshotID)
-		} else {
-			kept++
-		}
-	}
-
-	if len(toExpire) == 0 {
-		return "no snapshots expired", nil, nil
+	plannedSnapshotIDs := snapshotIDSet(snapshots)
+	plannedRefs := snapshotRefs(meta)
+	toExpire, refsToRemove, err := planSnapshotExpiration(meta, config, start)
+	if err != nil {
+		return "", nil, fmt.Errorf("plan snapshot expiration: %w", err)
 	}
 
 	// Split snapshots into expired and kept sets
@@ -102,7 +63,7 @@ func (h *Handler) expireSnapshots(
 		expireSet[id] = struct{}{}
 	}
 	var expiredSnaps, keptSnaps []table.Snapshot
-	for _, snap := range sorted {
+	for _, snap := range snapshots {
 		if _, ok := expireSet[snap.SnapshotID]; ok {
 			expiredSnaps = append(expiredSnaps, snap)
 		} else {
@@ -110,15 +71,19 @@ func (h *Handler) expireSnapshots(
 		}
 	}
 
-	// Collect all files referenced by each set before modifying metadata.
-	// This lets us determine which files become unreferenced.
-	expiredFiles, err := collectSnapshotFiles(ctx, filerClient, bucketName, tablePath, expiredSnaps)
-	if err != nil {
-		return "", nil, fmt.Errorf("collect expired snapshot files: %w", err)
-	}
-	keptFiles, err := collectSnapshotFiles(ctx, filerClient, bucketName, tablePath, keptSnaps)
-	if err != nil {
-		return "", nil, fmt.Errorf("collect kept snapshot files: %w", err)
+	expiredFiles := make(map[string]struct{})
+	keptFiles := make(map[string]struct{})
+	if len(toExpire) > 0 {
+		// Collect all files referenced by each set before modifying metadata.
+		// This lets us determine which files become unreferenced.
+		expiredFiles, err = collectSnapshotFiles(ctx, filerClient, bucketName, tablePath, expiredSnaps)
+		if err != nil {
+			return "", nil, fmt.Errorf("collect expired snapshot files: %w", err)
+		}
+		keptFiles, err = collectSnapshotFiles(ctx, filerClient, bucketName, tablePath, keptSnaps)
+		if err != nil {
+			return "", nil, fmt.Errorf("collect kept snapshot files: %w", err)
+		}
 	}
 
 	// Normalize kept file paths for consistent comparison
@@ -127,17 +92,25 @@ func (h *Handler) expireSnapshots(
 		normalizedKept[normalizeIcebergPath(f, bucketName, tablePath)] = struct{}{}
 	}
 
-	// Use MetadataBuilder to remove snapshots and create new metadata
-	err = h.commitWithRetry(ctx, filerClient, bucketName, tablePath, metadataFileName, config, func(currentMeta table.Metadata, builder *table.MetadataBuilder) error {
-		// Guard: verify table head hasn't changed since we planned
-		cs := currentMeta.CurrentSnapshot()
-		if (cs == nil) != (currentSnapID == 0) || (cs != nil && cs.SnapshotID != currentSnapID) {
-			return errStalePlan
+	if len(toExpire) > 0 || len(refsToRemove) > 0 {
+		// Use MetadataBuilder to remove refs/snapshots and create new metadata.
+		err = h.commitWithRetry(ctx, filerClient, bucketName, tablePath, metadataFileName, config, func(currentMeta table.Metadata, builder *table.MetadataBuilder) error {
+			if !sameSnapshotSet(currentMeta, plannedSnapshotIDs) || !sameSnapshotRefs(currentMeta, plannedRefs) {
+				return errStalePlan
+			}
+			for _, refName := range refsToRemove {
+				if err := builder.RemoveSnapshotRef(refName); err != nil {
+					return err
+				}
+			}
+			if len(toExpire) == 0 {
+				return nil
+			}
+			return builder.RemoveSnapshots(toExpire)
+		})
+		if err != nil {
+			return "", nil, fmt.Errorf("commit snapshot expiration: %w", err)
 		}
-		return builder.RemoveSnapshots(toExpire)
-	})
-	if err != nil {
-		return "", nil, fmt.Errorf("commit snapshot expiration: %w", err)
 	}
 
 	// Delete files exclusively referenced by expired snapshots (best-effort)
@@ -157,12 +130,164 @@ func (h *Handler) expireSnapshots(
 		}
 	}
 
-	metrics := map[string]int64{
-		MetricSnapshotsExpired: int64(len(toExpire)),
-		MetricFilesDeleted:     int64(deletedCount),
-		MetricDurationMs:       time.Since(start).Milliseconds(),
+	metadataFilesDeleted := 0
+	if config.CleanExpiredMetadata {
+		metadataFilesDeleted, err = cleanupExpiredMetadataFiles(ctx, filerClient, bucketName, tablePath)
+		if err != nil {
+			return "", nil, fmt.Errorf("clean expired metadata: %w", err)
+		}
 	}
-	return fmt.Sprintf("expired %d snapshot(s), deleted %d unreferenced file(s)", len(toExpire), deletedCount), metrics, nil
+
+	if len(toExpire) == 0 && len(refsToRemove) == 0 && metadataFilesDeleted == 0 {
+		return "no snapshots expired", nil, nil
+	}
+
+	metrics := map[string]int64{
+		MetricSnapshotsExpired:     int64(len(toExpire)),
+		MetricSnapshotRefsRemoved:  int64(len(refsToRemove)),
+		MetricFilesDeleted:         int64(deletedCount),
+		MetricMetadataFilesDeleted: int64(metadataFilesDeleted),
+		MetricDurationMs:           time.Since(start).Milliseconds(),
+	}
+
+	var summaryParts []string
+	if len(toExpire) > 0 {
+		summaryParts = append(summaryParts, fmt.Sprintf("expired %d snapshot(s)", len(toExpire)))
+	}
+	if len(refsToRemove) > 0 {
+		summaryParts = append(summaryParts, fmt.Sprintf("removed %d snapshot ref(s)", len(refsToRemove)))
+	}
+	if len(toExpire) > 0 {
+		summaryParts = append(summaryParts, fmt.Sprintf("deleted %d unreferenced file(s)", deletedCount))
+	}
+	if metadataFilesDeleted > 0 {
+		summaryParts = append(summaryParts, fmt.Sprintf("deleted %d expired metadata file(s)", metadataFilesDeleted))
+	}
+	return strings.Join(summaryParts, ", "), metrics, nil
+}
+
+func planSnapshotExpiration(meta table.Metadata, config Config, now time.Time) ([]int64, []string, error) {
+	retentionMs := config.SnapshotRetentionHours * 3600 * 1000
+	minSnapshotsToKeep := int(config.MaxSnapshotsToKeep)
+	if minSnapshotsToKeep < 1 {
+		minSnapshotsToKeep = 1
+	}
+
+	snapsToKeep := make(map[int64]struct{})
+	var refsToRemove []string
+	nowMs := now.UnixMilli()
+	refCount := 0
+
+	for refName, ref := range meta.Refs() {
+		refCount++
+		snap := meta.SnapshotByID(ref.SnapshotID)
+		if snap == nil {
+			return nil, nil, fmt.Errorf("snapshot ref %q points to missing snapshot %d", refName, ref.SnapshotID)
+		}
+
+		maxRefAgeMs := retentionMs
+		if ref.MaxRefAgeMs != nil {
+			maxRefAgeMs = *ref.MaxRefAgeMs
+		}
+		if refName != table.MainBranch && nowMs-snap.TimestampMs > maxRefAgeMs {
+			refsToRemove = append(refsToRemove, refName)
+			continue
+		}
+
+		if ref.SnapshotRefType != table.BranchRef {
+			snapsToKeep[ref.SnapshotID] = struct{}{}
+			continue
+		}
+
+		maxSnapshotAgeMs := retentionMs
+		if ref.MaxSnapshotAgeMs != nil {
+			maxSnapshotAgeMs = *ref.MaxSnapshotAgeMs
+		}
+		refMinSnapshotsToKeep := minSnapshotsToKeep
+		if ref.MinSnapshotsToKeep != nil && *ref.MinSnapshotsToKeep > 0 {
+			refMinSnapshotsToKeep = *ref.MinSnapshotsToKeep
+		}
+
+		numSnapshots := 0
+		snapshotID := ref.SnapshotID
+		for {
+			snap = meta.SnapshotByID(snapshotID)
+			if snap == nil {
+				return nil, nil, fmt.Errorf("snapshot %d not found while traversing ref %q", snapshotID, refName)
+			}
+
+			snapshotAgeMs := nowMs - snap.TimestampMs
+			if snapshotAgeMs > maxSnapshotAgeMs && numSnapshots >= refMinSnapshotsToKeep {
+				break
+			}
+
+			snapsToKeep[snap.SnapshotID] = struct{}{}
+			if snap.ParentSnapshotID == nil {
+				break
+			}
+
+			snapshotID = *snap.ParentSnapshotID
+			numSnapshots++
+		}
+	}
+
+	if refCount == 0 {
+		if currentSnap := meta.CurrentSnapshot(); currentSnap != nil {
+			snapsToKeep[currentSnap.SnapshotID] = struct{}{}
+		}
+	}
+
+	var toExpire []int64
+	for _, snap := range meta.Snapshots() {
+		if _, ok := snapsToKeep[snap.SnapshotID]; !ok {
+			toExpire = append(toExpire, snap.SnapshotID)
+		}
+	}
+
+	return toExpire, refsToRemove, nil
+}
+
+func snapshotIDSet(snapshots []table.Snapshot) map[int64]struct{} {
+	ids := make(map[int64]struct{}, len(snapshots))
+	for _, snap := range snapshots {
+		ids[snap.SnapshotID] = struct{}{}
+	}
+	return ids
+}
+
+func sameSnapshotSet(meta table.Metadata, expected map[int64]struct{}) bool {
+	current := meta.Snapshots()
+	if len(current) != len(expected) {
+		return false
+	}
+	for _, snap := range current {
+		if _, ok := expected[snap.SnapshotID]; !ok {
+			return false
+		}
+	}
+	return true
+}
+
+func snapshotRefs(meta table.Metadata) map[string]table.SnapshotRef {
+	refs := make(map[string]table.SnapshotRef)
+	for refName, ref := range meta.Refs() {
+		refs[refName] = ref
+	}
+	return refs
+}
+
+func sameSnapshotRefs(meta table.Metadata, expected map[string]table.SnapshotRef) bool {
+	current := snapshotRefs(meta)
+	if len(current) != len(expected) {
+		return false
+	}
+	for refName, expectedRef := range expected {
+		currentRef, ok := current[refName]
+		if !ok || !currentRef.Equals(expectedRef) {
+			return false
+		}
+	}
+	return true
 }
 
 // collectSnapshotFiles returns all file paths (manifest lists, manifest files,
@@ -234,6 +359,15 @@ func (h *Handler) removeOrphans(
 		return "", nil, fmt.Errorf("collect orphan candidates: %w", err)
 	}
 
+	if config.RemoveOrphansDryRun {
+		metrics := map[string]int64{
+			MetricOrphansFound:   int64(len(orphanCandidates)),
+			MetricOrphansRemoved: 0,
+			MetricDurationMs:     time.Since(start).Milliseconds(),
+		}
+		return fmt.Sprintf("found %d orphan file(s) (dry run)", len(orphanCandidates)), metrics, nil
+	}
+
 	orphanCount := 0
 	for _, candidate := range orphanCandidates {
 		if delErr := deleteFilerFile(ctx, filerClient, candidate.Dir, candidate.Entry.Name); delErr != nil {
@@ -244,12 +378,60 @@ func (h *Handler) removeOrphans(
 	}
 
 	metrics := map[string]int64{
+		MetricOrphansFound:   int64(len(orphanCandidates)),
 		MetricOrphansRemoved: int64(orphanCount),
 		MetricDurationMs:     time.Since(start).Milliseconds(),
 	}
 	return fmt.Sprintf("removed %d orphan file(s)", orphanCount), metrics, nil
 }
 
+func cleanupExpiredMetadataFiles(
+	ctx context.Context,
+	filerClient filer_pb.SeaweedFilerClient,
+	bucketName, tablePath string,
+) (int, error) {
+	meta, currentMetadataFileName, err := loadCurrentMetadata(ctx, filerClient, bucketName, tablePath)
+	if err != nil {
+		return 0, fmt.Errorf("load metadata: %w", err)
+	}
+
+	referencedFiles := map[string]struct{}{
+		path.Join("metadata", currentMetadataFileName): {},
+	}
+	for entry := range meta.PreviousFiles() {
+		referencedFiles[normalizeIcebergPath(entry.MetadataFile, bucketName, tablePath)] = struct{}{}
+	}
+
+	currentVersion := extractMetadataVersion(currentMetadataFileName)
+	metadataDir := path.Join(s3tables.TablesPath, bucketName, tablePath, "metadata")
+	entries, err := listFilerEntries(ctx, filerClient, metadataDir, "")
+	if err != nil {
+		return 0, fmt.Errorf("list metadata dir: %w", err)
+	}
+
+	deleted := 0
+	for _, entry := range entries {
+		if entry == nil || entry.IsDirectory || !strings.HasSuffix(entry.Name, ".metadata.json") {
+			continue
+		}
+
+		relPath := path.Join("metadata", entry.Name)
+		if _, ok := referencedFiles[relPath]; ok {
+			continue
+		}
+		if extractMetadataVersion(entry.Name) >= currentVersion {
+			continue
+		}
+		if err := deleteFilerFile(ctx, filerClient, metadataDir, entry.Name); err != nil {
+			glog.Warningf("iceberg maintenance: failed to delete expired metadata file %s/%s: %v", metadataDir, entry.Name, err)
+			continue
+		}
+		deleted++
+	}
+
+	return deleted, nil
+}
+
 func collectOrphanCandidates(
 	ctx context.Context,
 	filerClient filer_pb.SeaweedFilerClient,