diff --git a/weed/worker/tasks/iceberg/compact.go b/weed/worker/tasks/iceberg/compact.go index d8f10ebec..7d6ef3dec 100644 --- a/weed/worker/tasks/iceberg/compact.go +++ b/weed/worker/tasks/iceberg/compact.go @@ -110,8 +110,8 @@ func (h *Handler) compactDataFiles( // Collect delete entries if we need to apply deletes var positionDeletes map[string][]int64 var eqDeleteGroups []equalityDeleteGroup + var allDeleteEntries []iceberg.ManifestEntry if config.ApplyDeletes && len(deleteManifests) > 0 { - var allDeleteEntries []iceberg.ManifestEntry for _, mf := range deleteManifests { manifestData, err := loadFileByIcebergPath(ctx, filerClient, bucketName, dataPath, mf.FilePath()) if err != nil { @@ -213,6 +213,7 @@ func (h *Handler) compactDataFiles( // Process each bin: read source Parquet files, merge, write output var newManifestEntries []iceberg.ManifestEntry var deletedManifestEntries []iceberg.ManifestEntry + var summary snapshotSummary totalMerged := 0 entrySeqNum := func(entry iceberg.ManifestEntry) *int64 { @@ -312,16 +313,19 @@ func (h *Handler) compactDataFiles( } writtenArtifacts = append(writtenArtifacts, artifact{dir: dataDir, fileName: mergedFileName}) + mergedDataFile := dfBuilder.Build() + summary.addFile(mergedDataFile) newEntry := iceberg.NewManifestEntry( iceberg.EntryStatusADDED, &newSnapID, nil, nil, - dfBuilder.Build(), + mergedDataFile, ) newManifestEntries = append(newManifestEntries, newEntry) // Mark original entries as deleted for _, entry := range bin.Entries { + summary.removeFile(entry.DataFile()) delEntry := iceberg.NewManifestEntry( iceberg.EntryStatusDELETED, &newSnapID, @@ -433,27 +437,23 @@ func (h *Handler) compactDataFiles( // Position deletes reference specific data files — if all those files // were compacted, the deletes are fully consumed. Equality deletes // apply broadly, so they're only consumed if all data files were compacted. - if !config.ApplyDeletes || (len(positionDeletes) == 0 && len(eqDeleteGroups) == 0) { - for _, mf := range deleteManifests { - allManifests = append(allManifests, mf) - } - } else { - // Check if any non-compacted data files remain - hasUncompactedFiles := false + carryDeletes := !config.ApplyDeletes || (len(positionDeletes) == 0 && len(eqDeleteGroups) == 0) + if !carryDeletes { + // Deletes still apply to any file that wasn't compacted. for _, entry := range allEntries { if _, compacted := compactedPaths[entry.DataFile().FilePath()]; !compacted { - hasUncompactedFiles = true + carryDeletes = true break } } - if hasUncompactedFiles { - // Some files weren't compacted — carry forward delete manifests - // since deletes may still apply to those files. - for _, mf := range deleteManifests { - allManifests = append(allManifests, mf) - } + } + if carryDeletes { + allManifests = append(allManifests, deleteManifests...) + } else { + // All files were compacted, so the deletes leave the table with them. + for _, entry := range allDeleteEntries { + summary.removeFile(entry.DataFile()) } - // If all files were compacted, deletes are fully consumed — don't carry forward. } // Write new manifest list @@ -485,16 +485,13 @@ func (h *Handler) compactDataFiles( SequenceNumber: seqNum, TimestampMs: newSnapID, ManifestList: manifestListLocation, - Summary: &table.Summary{ - Operation: table.OpReplace, - Properties: map[string]string{ - "maintenance": "compact_data_files", - "merged-files": fmt.Sprintf("%d", totalMerged), - "new-files": fmt.Sprintf("%d", len(newManifestEntries)), - "compaction-bins": fmt.Sprintf("%d", len(bins)), - "rewrite-strategy": rewritePlan.strategy, - }, - }, + Summary: summary.build(table.OpReplace, cs, map[string]string{ + "maintenance": "compact_data_files", + "merged-files": fmt.Sprintf("%d", totalMerged), + "new-files": fmt.Sprintf("%d", len(newManifestEntries)), + "compaction-bins": fmt.Sprintf("%d", len(bins)), + "rewrite-strategy": rewritePlan.strategy, + }), SchemaID: func() *int { id := schema.ID return &id diff --git a/weed/worker/tasks/iceberg/delete_rewrite.go b/weed/worker/tasks/iceberg/delete_rewrite.go index e633cefee..9c4ef0721 100644 --- a/weed/worker/tasks/iceberg/delete_rewrite.go +++ b/weed/worker/tasks/iceberg/delete_rewrite.go @@ -367,6 +367,7 @@ func (h *Handler) rewritePositionDeleteFiles( artifactSuffix := compactRandomSuffix() replacedPaths := make(map[string]struct{}) + var summary snapshotSummary var rewrittenGroups int64 var skippedGroups int64 var deleteFilesRewritten int64 @@ -457,12 +458,14 @@ func (h *Handler) rewritePositionDeleteFiles( if err != nil { return "", nil, fmt.Errorf("build rewritten delete file: %w", err) } - entry := iceberg.NewManifestEntry(iceberg.EntryStatusADDED, &newSnapID, nil, nil, dfBuilder.Build()) - addToSpec(group.SpecID, entry) + rewrittenDeleteFile := dfBuilder.Build() + summary.addFile(rewrittenDeleteFile) + addToSpec(group.SpecID, iceberg.NewManifestEntry(iceberg.EntryStatusADDED, &newSnapID, nil, nil, rewrittenDeleteFile)) deleteFilesWritten++ } for _, input := range group.Inputs { + summary.removeFile(input.Entry.DataFile()) delEntry := iceberg.NewManifestEntry( iceberg.EntryStatusDELETED, &newSnapID, @@ -561,15 +564,12 @@ func (h *Handler) rewritePositionDeleteFiles( SequenceNumber: seqNum, TimestampMs: newSnapID, ManifestList: manifestListLocation, - Summary: &table.Summary{ - Operation: table.OpReplace, - Properties: map[string]string{ - "maintenance": "rewrite_position_delete_files", - "delete-files-rewritten": fmt.Sprintf("%d", deleteFilesRewritten), - "delete-files-written": fmt.Sprintf("%d", deleteFilesWritten), - "delete-groups": fmt.Sprintf("%d", rewrittenGroups), - }, - }, + Summary: summary.build(table.OpReplace, cs, map[string]string{ + "maintenance": "rewrite_position_delete_files", + "delete-files-rewritten": fmt.Sprintf("%d", deleteFilesRewritten), + "delete-files-written": fmt.Sprintf("%d", deleteFilesWritten), + "delete-groups": fmt.Sprintf("%d", rewrittenGroups), + }), SchemaID: func() *int { id := meta.CurrentSchema().ID return &id diff --git a/weed/worker/tasks/iceberg/exec_test.go b/weed/worker/tasks/iceberg/exec_test.go index a1b4aea65..01881e474 100644 --- a/weed/worker/tasks/iceberg/exec_test.go +++ b/weed/worker/tasks/iceberg/exec_test.go @@ -9,6 +9,7 @@ import ( "net" "path" "sort" + "strconv" "strings" "sync" "testing" @@ -263,8 +264,11 @@ type tableSetup struct { // it differs from the catalog path, as it does for tables an external REST // client created at their own location (e.g. "ns/table-"). Such a // table records absolute s3:// URIs for every file it references. - DataPath string - Snapshots []table.Snapshot + DataPath string + // SnapshotSummary is recorded on the table's snapshot, the way the writer + // that created it would record the table's running totals. + SnapshotSummary map[string]string + Snapshots []table.Snapshot } func (ts tableSetup) tablePath() string { @@ -861,7 +865,14 @@ func TestRewriteManifestsExecution(t *testing.T) { Namespace: "analytics", TableName: "events", Snapshots: []table.Snapshot{ - {SnapshotID: 1, TimestampMs: now, ManifestList: "metadata/snap-1.avro"}, + {SnapshotID: 1, TimestampMs: now, ManifestList: "metadata/snap-1.avro", Summary: &table.Summary{ + Operation: table.OpAppend, + Properties: map[string]string{ + "total-data-files": "5", + "total-records": "5", + "total-files-size": "5120", + }, + }}, }, } meta := populateTable(t, fs, setup) @@ -992,6 +1003,16 @@ func TestRewriteManifestsExecution(t *testing.T) { if !strings.HasPrefix(xattrMeta.MetadataLocation, wantPrefix+"v2-") { t.Errorf("xattr metadataLocation should be absolute, got %q", xattrMeta.MetadataLocation) } + + // Merging manifests moves no data, so the table's totals are unchanged. + if newSnap.Summary == nil { + t.Fatal("new snapshot has no summary") + } + for key, want := range map[string]string{"total-data-files": "5", "total-records": "5", "total-files-size": "5120"} { + if got := newSnap.Summary.Properties[key]; got != want { + t.Errorf("%s = %q, want %q", key, got, want) + } + } } func TestRewriteManifestsBelowThreshold(t *testing.T) { @@ -2447,6 +2468,9 @@ func populateTableWithDeleteFilesAndSortOrder( // Build final metadata with snapshot now := time.Now().UnixMilli() snap := table.Snapshot{SnapshotID: 1, TimestampMs: now, ManifestList: setup.fileRef("metadata", "snap-1.avro")} + if setup.SnapshotSummary != nil { + snap.Summary = &table.Summary{Operation: table.OpAppend, Properties: setup.SnapshotSummary} + } builder, err := table.MetadataBuilderFromBase(meta, "s3://"+setup.BucketName+"/"+setup.dataPath()) if err != nil { t.Fatalf("create metadata builder: %v", err) @@ -2712,6 +2736,208 @@ func TestCompactDataFilesMetrics(t *testing.T) { } } +// summaryDataFiles is the two-file input every snapshot-summary test compacts. +func summaryDataFiles() []struct { + Name string + Rows []struct { + ID int64 + Name string + } +} { + return []struct { + Name string + Rows []struct { + ID int64 + Name string + } + }{ + {"d1.parquet", []struct { + ID int64 + Name string + }{{1, "a"}, {2, "b"}}}, + {"d2.parquet", []struct { + ID int64 + Name string + }{{3, "c"}}}, + } +} + +func snapshotSummaryProps(t *testing.T, client filer_pb.SeaweedFilerClient, setup tableSetup) map[string]string { + t.Helper() + state, err := loadCurrentMetadata(context.Background(), client, setup.BucketName, setup.tablePath()) + if err != nil { + t.Fatalf("loadCurrentMetadata: %v", err) + } + snap := state.Metadata.CurrentSnapshot() + if snap == nil || snap.Summary == nil { + t.Fatalf("new snapshot has no summary: %+v", snap) + } + return snap.Summary.Properties +} + +func requireSummaryValue(t *testing.T, props map[string]string, key, want string) { + t.Helper() + if got := props[key]; got != want { + t.Errorf("%s = %q, want %q", key, got, want) + } +} + +func summaryInt(t *testing.T, props map[string]string, key string) int64 { + t.Helper() + raw, ok := props[key] + if !ok { + t.Fatalf("summary is missing %s: %v", key, props) + } + value, err := strconv.ParseInt(raw, 10, 64) + if err != nil { + t.Fatalf("summary %s = %q: %v", key, raw, err) + } + return value +} + +// Engines report a table's size from the current snapshot summary, so a +// compaction has to carry the running totals forward the way the writer that +// created the table did. +func TestCompactDataFilesRecordsSnapshotTotals(t *testing.T) { + fs, client := startFakeFiler(t) + + const parentFilesSize = 4096 + setup := tableSetup{ + BucketName: "tb", Namespace: "ns", TableName: "tbl", + SnapshotSummary: map[string]string{ + "total-data-files": "2", + "total-records": "3", + "total-files-size": strconv.Itoa(parentFilesSize), + }, + } + populateTableWithDeleteFiles(t, fs, setup, summaryDataFiles(), nil, nil) + + handler := NewHandler(nil) + config := Config{ + TargetFileSizeBytes: 256 * 1024 * 1024, + MinInputFiles: 2, + MaxCommitRetries: 3, + ApplyDeletes: true, + } + if _, _, err := handler.compactDataFiles(context.Background(), client, setup.BucketName, setup.tablePath(), config, nil); err != nil { + t.Fatalf("compactDataFiles: %v", err) + } + + props := snapshotSummaryProps(t, client, setup) + requireSummaryValue(t, props, "added-data-files", "1") + requireSummaryValue(t, props, "deleted-data-files", "2") + requireSummaryValue(t, props, "added-records", "3") + requireSummaryValue(t, props, "deleted-records", "3") + + // Two files became one, holding the same rows. + requireSummaryValue(t, props, "total-data-files", "1") + requireSummaryValue(t, props, "total-records", "3") + wantSize := parentFilesSize + summaryInt(t, props, "added-files-size") - summaryInt(t, props, "removed-files-size") + requireSummaryValue(t, props, "total-files-size", strconv.FormatInt(wantSize, 10)) + + // The operation's own labels survive alongside the counters. + requireSummaryValue(t, props, "maintenance", "compact_data_files") +} + +// A table whose parent snapshot never recorded totals gets none invented for +// it: "total-records: 0" on a table with rows is worse than no answer. +func TestCompactDataFilesLeavesOutTotalsParentNeverRecorded(t *testing.T) { + fs, client := startFakeFiler(t) + + setup := tableSetup{BucketName: "tb", Namespace: "ns", TableName: "tbl"} + populateTableWithDeleteFiles(t, fs, setup, summaryDataFiles(), nil, nil) + + handler := NewHandler(nil) + config := Config{ + TargetFileSizeBytes: 256 * 1024 * 1024, + MinInputFiles: 2, + MaxCommitRetries: 3, + ApplyDeletes: true, + } + if _, _, err := handler.compactDataFiles(context.Background(), client, setup.BucketName, setup.tablePath(), config, nil); err != nil { + t.Fatalf("compactDataFiles: %v", err) + } + + props := snapshotSummaryProps(t, client, setup) + requireSummaryValue(t, props, "added-data-files", "1") + requireSummaryValue(t, props, "deleted-data-files", "2") + for _, key := range []string{"total-data-files", "total-records", "total-files-size"} { + if got, ok := props[key]; ok { + t.Errorf("%s = %q, want it left out", key, got) + } + } +} + +func TestRewritePositionDeleteFilesRecordsSnapshotTotals(t *testing.T) { + fs, client := startFakeFiler(t) + + setup := tableSetup{ + BucketName: "tb", Namespace: "ns", TableName: "tbl", + SnapshotSummary: map[string]string{ + "total-data-files": "1", + "total-records": "3", + "total-delete-files": "2", + "total-position-deletes": "3", + }, + } + populateTableWithDeleteFiles(t, fs, setup, + []struct { + Name string + Rows []struct { + ID int64 + Name string + } + }{ + {"d1.parquet", []struct { + ID int64 + Name string + }{{1, "alice"}, {2, "bob"}, {3, "charlie"}}}, + }, + []struct { + Name string + Rows []struct { + FilePath string + Pos int64 + } + }{ + {"pd1.parquet", []struct { + FilePath string + Pos int64 + }{{"data/d1.parquet", 0}, {"data/d1.parquet", 2}}}, + {"pd2.parquet", []struct { + FilePath string + Pos int64 + }{{"data/d1.parquet", 1}}}, + }, + nil, + ) + + handler := NewHandler(nil) + config := Config{ + DeleteTargetFileSizeBytes: 64 * 1024 * 1024, + DeleteMinInputFiles: 2, + DeleteMaxFileGroupSizeBytes: 128 * 1024 * 1024, + DeleteMaxOutputFiles: 4, + MaxCommitRetries: 3, + } + if _, _, err := handler.rewritePositionDeleteFiles(context.Background(), client, setup.BucketName, setup.tablePath(), config); err != nil { + t.Fatalf("rewritePositionDeleteFiles: %v", err) + } + + props := snapshotSummaryProps(t, client, setup) + requireSummaryValue(t, props, "added-delete-files", "1") + requireSummaryValue(t, props, "removed-delete-files", "2") + requireSummaryValue(t, props, "added-position-delete-files", "1") + requireSummaryValue(t, props, "added-position-deletes", "3") + requireSummaryValue(t, props, "removed-position-deletes", "3") + + // Two delete files became one; the deleted rows and the data are untouched. + requireSummaryValue(t, props, "total-delete-files", "1") + requireSummaryValue(t, props, "total-position-deletes", "3") + requireSummaryValue(t, props, "total-data-files", "1") + requireSummaryValue(t, props, "total-records", "3") +} + func TestExpireSnapshotsMetrics(t *testing.T) { fs, client := startFakeFiler(t) diff --git a/weed/worker/tasks/iceberg/operations.go b/weed/worker/tasks/iceberg/operations.go index ae994e46e..fc1733974 100644 --- a/weed/worker/tasks/iceberg/operations.go +++ b/weed/worker/tasks/iceberg/operations.go @@ -520,10 +520,8 @@ func (h *Handler) rewriteManifests( SequenceNumber: cs.SequenceNumber + 1, TimestampMs: time.Now().UnixMilli(), ManifestList: manifestListLocation, - Summary: &table.Summary{ - Operation: table.OpReplace, - Properties: map[string]string{"maintenance": "rewrite_manifests"}, - }, + // Merging manifests changes no file, so the totals carry over. + Summary: snapshotSummary{}.build(table.OpReplace, cs, map[string]string{"maintenance": "rewrite_manifests"}), SchemaID: func() *int { id := schema.ID return &id diff --git a/weed/worker/tasks/iceberg/snapshot_summary.go b/weed/worker/tasks/iceberg/snapshot_summary.go new file mode 100644 index 000000000..88c85621e --- /dev/null +++ b/weed/worker/tasks/iceberg/snapshot_summary.go @@ -0,0 +1,152 @@ +package iceberg + +import ( + "strconv" + + "github.com/apache/iceberg-go" + "github.com/apache/iceberg-go/table" +) + +// Snapshot summary property names from the Iceberg spec. iceberg-go keeps its +// own copies unexported, so the maintenance operations spell them here. +const ( + summaryAddedDataFiles = "added-data-files" + summaryDeletedDataFiles = "deleted-data-files" + summaryAddedDeleteFiles = "added-delete-files" + summaryRemovedDeleteFiles = "removed-delete-files" + summaryAddedPosDeleteFiles = "added-position-delete-files" + summaryRemovedPosDelFiles = "removed-position-delete-files" + summaryAddedEqDeleteFiles = "added-equality-delete-files" + summaryRemovedEqDelFiles = "removed-equality-delete-files" + summaryAddedRecords = "added-records" + summaryDeletedRecords = "deleted-records" + summaryAddedPosDeletes = "added-position-deletes" + summaryRemovedPosDeletes = "removed-position-deletes" + summaryAddedEqDeletes = "added-equality-deletes" + summaryRemovedEqDeletes = "removed-equality-deletes" + summaryAddedFilesSize = "added-files-size" + summaryRemovedFilesSize = "removed-files-size" + + summaryTotalDataFiles = "total-data-files" + summaryTotalDeleteFiles = "total-delete-files" + summaryTotalRecords = "total-records" + summaryTotalFilesSize = "total-files-size" + summaryTotalPosDeletes = "total-position-deletes" + summaryTotalEqDeletes = "total-equality-deletes" +) + +// snapshotSummary accumulates the files a maintenance operation adds to and +// removes from a table, and renders them as an Iceberg snapshot summary. +// +// Engines read a table's size straight out of the current snapshot summary — +// PyIceberg's inspect.snapshots, Trino's $snapshots and Spark's DESCRIBE all +// report total-records, total-data-files and total-files-size verbatim — so a +// maintenance snapshot carrying only its own labels blanks those numbers for +// the whole table until the next writer commits. +type snapshotSummary struct { + addedFilesSize, removedFilesSize int64 + addedDataFiles, removedDataFiles int64 + addedRecords, removedRecords int64 + addedDeleteFiles, removedDeleteFiles int64 + addedPosDeleteFiles, removedPosFiles int64 + addedEqDeleteFiles, removedEqFiles int64 + addedPosDeletes, removedPosDeletes int64 + addedEqDeletes, removedEqDeletes int64 +} + +func (s *snapshotSummary) addFile(df iceberg.DataFile) { + s.addedFilesSize += df.FileSizeBytes() + switch df.ContentType() { + case iceberg.EntryContentData: + s.addedDataFiles++ + s.addedRecords += df.Count() + case iceberg.EntryContentPosDeletes: + s.addedDeleteFiles++ + s.addedPosDeleteFiles++ + s.addedPosDeletes += df.Count() + case iceberg.EntryContentEqDeletes: + s.addedDeleteFiles++ + s.addedEqDeleteFiles++ + s.addedEqDeletes += df.Count() + } +} + +func (s *snapshotSummary) removeFile(df iceberg.DataFile) { + s.removedFilesSize += df.FileSizeBytes() + switch df.ContentType() { + case iceberg.EntryContentData: + s.removedDataFiles++ + s.removedRecords += df.Count() + case iceberg.EntryContentPosDeletes: + s.removedDeleteFiles++ + s.removedPosFiles++ + s.removedPosDeletes += df.Count() + case iceberg.EntryContentEqDeletes: + s.removedDeleteFiles++ + s.removedEqFiles++ + s.removedEqDeletes += df.Count() + } +} + +// build renders the summary for a snapshot replacing parent: the operation's +// own labels, the added/removed counters, and the running totals. +// +// A total is carried forward only when the parent recorded it. Iceberg treats a +// missing total as zero, which turns a compaction that replaces two files with +// one into "total-data-files: -1" — dropped as negative — or worse, a table +// with millions of rows into "total-records: 0". Leaving the field out lets a +// reader fall back to the manifests instead of believing a made-up number. +func (s snapshotSummary) build(operation table.Operation, parent *table.Snapshot, labels map[string]string) *table.Summary { + props := iceberg.Properties{} + for k, v := range labels { + props[k] = v + } + + setPositive := func(key string, value int64) { + if value > 0 { + props[key] = strconv.FormatInt(value, 10) + } + } + setPositive(summaryAddedFilesSize, s.addedFilesSize) + setPositive(summaryRemovedFilesSize, s.removedFilesSize) + setPositive(summaryAddedDataFiles, s.addedDataFiles) + setPositive(summaryDeletedDataFiles, s.removedDataFiles) + setPositive(summaryAddedRecords, s.addedRecords) + setPositive(summaryDeletedRecords, s.removedRecords) + setPositive(summaryAddedDeleteFiles, s.addedDeleteFiles) + setPositive(summaryRemovedDeleteFiles, s.removedDeleteFiles) + setPositive(summaryAddedPosDeleteFiles, s.addedPosDeleteFiles) + setPositive(summaryRemovedPosDelFiles, s.removedPosFiles) + setPositive(summaryAddedEqDeleteFiles, s.addedEqDeleteFiles) + setPositive(summaryRemovedEqDelFiles, s.removedEqFiles) + setPositive(summaryAddedPosDeletes, s.addedPosDeletes) + setPositive(summaryRemovedPosDeletes, s.removedPosDeletes) + setPositive(summaryAddedEqDeletes, s.addedEqDeletes) + setPositive(summaryRemovedEqDeletes, s.removedEqDeletes) + + var previous iceberg.Properties + if parent != nil && parent.Summary != nil { + previous = parent.Summary.Properties + } + carryTotal := func(key string, added, removed int64) { + raw, ok := previous[key] + if !ok { + return + } + before, err := strconv.ParseInt(raw, 10, 64) + if err != nil { + return + } + if total := before + added - removed; total >= 0 { + props[key] = strconv.FormatInt(total, 10) + } + } + carryTotal(summaryTotalDataFiles, s.addedDataFiles, s.removedDataFiles) + carryTotal(summaryTotalDeleteFiles, s.addedDeleteFiles, s.removedDeleteFiles) + carryTotal(summaryTotalRecords, s.addedRecords, s.removedRecords) + carryTotal(summaryTotalFilesSize, s.addedFilesSize, s.removedFilesSize) + carryTotal(summaryTotalPosDeletes, s.addedPosDeletes, s.removedPosDeletes) + carryTotal(summaryTotalEqDeletes, s.addedEqDeletes, s.removedEqDeletes) + + return &table.Summary{Operation: operation, Properties: props} +}