diff --git a/weed/plugin/worker/iceberg/config.go b/weed/plugin/worker/iceberg/config.go index 8da2ce268..cc51c7461 100644 --- a/weed/plugin/worker/iceberg/config.go +++ b/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 diff --git a/weed/plugin/worker/iceberg/exec_test.go b/weed/plugin/worker/iceberg/exec_test.go index d262c4f91..82235227a 100644 --- a/weed/plugin/worker/iceberg/exec_test.go +++ b/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) diff --git a/weed/plugin/worker/iceberg/handler.go b/weed/plugin/worker/iceberg/handler.go index 5ef03c185..e71fa7c48 100644 --- a/weed/plugin/worker/iceberg/handler.go +++ b/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}}, }, } } diff --git a/weed/plugin/worker/iceberg/handler_test.go b/weed/plugin/worker/iceberg/handler_test.go index 42a9c8b4c..07b0c4198 100644 --- a/weed/plugin/worker/iceberg/handler_test.go +++ b/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) diff --git a/weed/plugin/worker/iceberg/operations.go b/weed/plugin/worker/iceberg/operations.go index fa8b89af0..28fd4f25f 100644 --- a/weed/plugin/worker/iceberg/operations.go +++ b/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,