mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-10-26 15:00:48 +01:00
iceberg maintenance: record file counters on the snapshots it commits (#10420)
A maintenance snapshot carried only its own labels — merged-files, delete-groups and friends — and no summary counters, so every engine that reads a table's size out of the current snapshot summary reported nothing for it: PyIceberg's inspect.snapshots, Trino's $snapshots and Spark's DESCRIBE all read total-records, total-data-files and total-files-size verbatim, and a table lost them the moment compaction touched it. Accumulate the files each operation adds and removes, and render them the way the spec defines: the added-*/removed-* counters from the files themselves, then the totals carried over from the parent snapshot. Carry a total only when the parent recorded it. Iceberg treats a missing total as zero, which turns a compaction replacing two files with one into a negative total-data-files, or 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. Compaction also accounts for the delete files it consumes, so a run that folds every delete into the rewritten data reports them as removed.
This commit is contained in:
1 parent
b4b0346f95
commit
1e1b2bb2f9
5 files changed
+418
-45
No files matched your search
@@ -110,8 +110,8 @@ func (h *Handler) compactDataFiles(
|
|||||||
// Collect delete entries if we need to apply deletes
|
// Collect delete entries if we need to apply deletes
|
||||||
var positionDeletes map[string][]int64
|
var positionDeletes map[string][]int64
|
||||||
var eqDeleteGroups []equalityDeleteGroup
|
var eqDeleteGroups []equalityDeleteGroup
|
||||||
|
var allDeleteEntries []iceberg.ManifestEntry
|
||||||
if config.ApplyDeletes && len(deleteManifests) > 0 {
|
if config.ApplyDeletes && len(deleteManifests) > 0 {
|
||||||
var allDeleteEntries []iceberg.ManifestEntry
|
|
||||||
for _, mf := range deleteManifests {
|
for _, mf := range deleteManifests {
|
||||||
manifestData, err := loadFileByIcebergPath(ctx, filerClient, bucketName, dataPath, mf.FilePath())
|
manifestData, err := loadFileByIcebergPath(ctx, filerClient, bucketName, dataPath, mf.FilePath())
|
||||||
if err != nil {
|
if err != nil {
|
||||||
@@ -213,6 +213,7 @@ func (h *Handler) compactDataFiles(
|
|||||||
// Process each bin: read source Parquet files, merge, write output
|
// Process each bin: read source Parquet files, merge, write output
|
||||||
var newManifestEntries []iceberg.ManifestEntry
|
var newManifestEntries []iceberg.ManifestEntry
|
||||||
var deletedManifestEntries []iceberg.ManifestEntry
|
var deletedManifestEntries []iceberg.ManifestEntry
|
||||||
|
var summary snapshotSummary
|
||||||
totalMerged := 0
|
totalMerged := 0
|
||||||
|
|
||||||
entrySeqNum := func(entry iceberg.ManifestEntry) *int64 {
|
entrySeqNum := func(entry iceberg.ManifestEntry) *int64 {
|
||||||
@@ -312,16 +313,19 @@ func (h *Handler) compactDataFiles(
|
|||||||
}
|
}
|
||||||
writtenArtifacts = append(writtenArtifacts, artifact{dir: dataDir, fileName: mergedFileName})
|
writtenArtifacts = append(writtenArtifacts, artifact{dir: dataDir, fileName: mergedFileName})
|
||||||
|
|
||||||
|
mergedDataFile := dfBuilder.Build()
|
||||||
|
summary.addFile(mergedDataFile)
|
||||||
newEntry := iceberg.NewManifestEntry(
|
newEntry := iceberg.NewManifestEntry(
|
||||||
iceberg.EntryStatusADDED,
|
iceberg.EntryStatusADDED,
|
||||||
&newSnapID,
|
&newSnapID,
|
||||||
nil, nil,
|
nil, nil,
|
||||||
dfBuilder.Build(),
|
mergedDataFile,
|
||||||
)
|
)
|
||||||
newManifestEntries = append(newManifestEntries, newEntry)
|
newManifestEntries = append(newManifestEntries, newEntry)
|
||||||
|
|
||||||
// Mark original entries as deleted
|
// Mark original entries as deleted
|
||||||
for _, entry := range bin.Entries {
|
for _, entry := range bin.Entries {
|
||||||
|
summary.removeFile(entry.DataFile())
|
||||||
delEntry := iceberg.NewManifestEntry(
|
delEntry := iceberg.NewManifestEntry(
|
||||||
iceberg.EntryStatusDELETED,
|
iceberg.EntryStatusDELETED,
|
||||||
&newSnapID,
|
&newSnapID,
|
||||||
@@ -433,27 +437,23 @@ func (h *Handler) compactDataFiles(
|
|||||||
// Position deletes reference specific data files — if all those files
|
// Position deletes reference specific data files — if all those files
|
||||||
// were compacted, the deletes are fully consumed. Equality deletes
|
// were compacted, the deletes are fully consumed. Equality deletes
|
||||||
// apply broadly, so they're only consumed if all data files were compacted.
|
// apply broadly, so they're only consumed if all data files were compacted.
|
||||||
if !config.ApplyDeletes || (len(positionDeletes) == 0 && len(eqDeleteGroups) == 0) {
|
carryDeletes := !config.ApplyDeletes || (len(positionDeletes) == 0 && len(eqDeleteGroups) == 0)
|
||||||
for _, mf := range deleteManifests {
|
if !carryDeletes {
|
||||||
allManifests = append(allManifests, mf)
|
// Deletes still apply to any file that wasn't compacted.
|
||||||
}
|
|
||||||
} else {
|
|
||||||
// Check if any non-compacted data files remain
|
|
||||||
hasUncompactedFiles := false
|
|
||||||
for _, entry := range allEntries {
|
for _, entry := range allEntries {
|
||||||
if _, compacted := compactedPaths[entry.DataFile().FilePath()]; !compacted {
|
if _, compacted := compactedPaths[entry.DataFile().FilePath()]; !compacted {
|
||||||
hasUncompactedFiles = true
|
carryDeletes = true
|
||||||
break
|
break
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
if hasUncompactedFiles {
|
}
|
||||||
// Some files weren't compacted — carry forward delete manifests
|
if carryDeletes {
|
||||||
// since deletes may still apply to those files.
|
allManifests = append(allManifests, deleteManifests...)
|
||||||
for _, mf := range deleteManifests {
|
} else {
|
||||||
allManifests = append(allManifests, mf)
|
// 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
|
// Write new manifest list
|
||||||
@@ -485,16 +485,13 @@ func (h *Handler) compactDataFiles(
|
|||||||
SequenceNumber: seqNum,
|
SequenceNumber: seqNum,
|
||||||
TimestampMs: newSnapID,
|
TimestampMs: newSnapID,
|
||||||
ManifestList: manifestListLocation,
|
ManifestList: manifestListLocation,
|
||||||
Summary: &table.Summary{
|
Summary: summary.build(table.OpReplace, cs, map[string]string{
|
||||||
Operation: table.OpReplace,
|
"maintenance": "compact_data_files",
|
||||||
Properties: map[string]string{
|
"merged-files": fmt.Sprintf("%d", totalMerged),
|
||||||
"maintenance": "compact_data_files",
|
"new-files": fmt.Sprintf("%d", len(newManifestEntries)),
|
||||||
"merged-files": fmt.Sprintf("%d", totalMerged),
|
"compaction-bins": fmt.Sprintf("%d", len(bins)),
|
||||||
"new-files": fmt.Sprintf("%d", len(newManifestEntries)),
|
"rewrite-strategy": rewritePlan.strategy,
|
||||||
"compaction-bins": fmt.Sprintf("%d", len(bins)),
|
}),
|
||||||
"rewrite-strategy": rewritePlan.strategy,
|
|
||||||
},
|
|
||||||
},
|
|
||||||
SchemaID: func() *int {
|
SchemaID: func() *int {
|
||||||
id := schema.ID
|
id := schema.ID
|
||||||
return &id
|
return &id
|
||||||
|
|||||||
@@ -367,6 +367,7 @@ func (h *Handler) rewritePositionDeleteFiles(
|
|||||||
artifactSuffix := compactRandomSuffix()
|
artifactSuffix := compactRandomSuffix()
|
||||||
|
|
||||||
replacedPaths := make(map[string]struct{})
|
replacedPaths := make(map[string]struct{})
|
||||||
|
var summary snapshotSummary
|
||||||
var rewrittenGroups int64
|
var rewrittenGroups int64
|
||||||
var skippedGroups int64
|
var skippedGroups int64
|
||||||
var deleteFilesRewritten int64
|
var deleteFilesRewritten int64
|
||||||
@@ -457,12 +458,14 @@ func (h *Handler) rewritePositionDeleteFiles(
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
return "", nil, fmt.Errorf("build rewritten delete file: %w", err)
|
return "", nil, fmt.Errorf("build rewritten delete file: %w", err)
|
||||||
}
|
}
|
||||||
entry := iceberg.NewManifestEntry(iceberg.EntryStatusADDED, &newSnapID, nil, nil, dfBuilder.Build())
|
rewrittenDeleteFile := dfBuilder.Build()
|
||||||
addToSpec(group.SpecID, entry)
|
summary.addFile(rewrittenDeleteFile)
|
||||||
|
addToSpec(group.SpecID, iceberg.NewManifestEntry(iceberg.EntryStatusADDED, &newSnapID, nil, nil, rewrittenDeleteFile))
|
||||||
deleteFilesWritten++
|
deleteFilesWritten++
|
||||||
}
|
}
|
||||||
|
|
||||||
for _, input := range group.Inputs {
|
for _, input := range group.Inputs {
|
||||||
|
summary.removeFile(input.Entry.DataFile())
|
||||||
delEntry := iceberg.NewManifestEntry(
|
delEntry := iceberg.NewManifestEntry(
|
||||||
iceberg.EntryStatusDELETED,
|
iceberg.EntryStatusDELETED,
|
||||||
&newSnapID,
|
&newSnapID,
|
||||||
@@ -561,15 +564,12 @@ func (h *Handler) rewritePositionDeleteFiles(
|
|||||||
SequenceNumber: seqNum,
|
SequenceNumber: seqNum,
|
||||||
TimestampMs: newSnapID,
|
TimestampMs: newSnapID,
|
||||||
ManifestList: manifestListLocation,
|
ManifestList: manifestListLocation,
|
||||||
Summary: &table.Summary{
|
Summary: summary.build(table.OpReplace, cs, map[string]string{
|
||||||
Operation: table.OpReplace,
|
"maintenance": "rewrite_position_delete_files",
|
||||||
Properties: map[string]string{
|
"delete-files-rewritten": fmt.Sprintf("%d", deleteFilesRewritten),
|
||||||
"maintenance": "rewrite_position_delete_files",
|
"delete-files-written": fmt.Sprintf("%d", deleteFilesWritten),
|
||||||
"delete-files-rewritten": fmt.Sprintf("%d", deleteFilesRewritten),
|
"delete-groups": fmt.Sprintf("%d", rewrittenGroups),
|
||||||
"delete-files-written": fmt.Sprintf("%d", deleteFilesWritten),
|
}),
|
||||||
"delete-groups": fmt.Sprintf("%d", rewrittenGroups),
|
|
||||||
},
|
|
||||||
},
|
|
||||||
SchemaID: func() *int {
|
SchemaID: func() *int {
|
||||||
id := meta.CurrentSchema().ID
|
id := meta.CurrentSchema().ID
|
||||||
return &id
|
return &id
|
||||||
|
|||||||
@@ -9,6 +9,7 @@ import (
|
|||||||
"net"
|
"net"
|
||||||
"path"
|
"path"
|
||||||
"sort"
|
"sort"
|
||||||
|
"strconv"
|
||||||
"strings"
|
"strings"
|
||||||
"sync"
|
"sync"
|
||||||
"testing"
|
"testing"
|
||||||
@@ -263,8 +264,11 @@ type tableSetup struct {
|
|||||||
// it differs from the catalog path, as it does for tables an external REST
|
// it differs from the catalog path, as it does for tables an external REST
|
||||||
// client created at their own location (e.g. "ns/table-<uuid>"). Such a
|
// client created at their own location (e.g. "ns/table-<uuid>"). Such a
|
||||||
// table records absolute s3:// URIs for every file it references.
|
// table records absolute s3:// URIs for every file it references.
|
||||||
DataPath string
|
DataPath string
|
||||||
Snapshots []table.Snapshot
|
// 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 {
|
func (ts tableSetup) tablePath() string {
|
||||||
@@ -861,7 +865,14 @@ func TestRewriteManifestsExecution(t *testing.T) {
|
|||||||
Namespace: "analytics",
|
Namespace: "analytics",
|
||||||
TableName: "events",
|
TableName: "events",
|
||||||
Snapshots: []table.Snapshot{
|
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)
|
meta := populateTable(t, fs, setup)
|
||||||
@@ -992,6 +1003,16 @@ func TestRewriteManifestsExecution(t *testing.T) {
|
|||||||
if !strings.HasPrefix(xattrMeta.MetadataLocation, wantPrefix+"v2-") {
|
if !strings.HasPrefix(xattrMeta.MetadataLocation, wantPrefix+"v2-") {
|
||||||
t.Errorf("xattr metadataLocation should be absolute, got %q", xattrMeta.MetadataLocation)
|
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) {
|
func TestRewriteManifestsBelowThreshold(t *testing.T) {
|
||||||
@@ -2447,6 +2468,9 @@ func populateTableWithDeleteFilesAndSortOrder(
|
|||||||
// Build final metadata with snapshot
|
// Build final metadata with snapshot
|
||||||
now := time.Now().UnixMilli()
|
now := time.Now().UnixMilli()
|
||||||
snap := table.Snapshot{SnapshotID: 1, TimestampMs: now, ManifestList: setup.fileRef("metadata", "snap-1.avro")}
|
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())
|
builder, err := table.MetadataBuilderFromBase(meta, "s3://"+setup.BucketName+"/"+setup.dataPath())
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatalf("create metadata builder: %v", err)
|
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) {
|
func TestExpireSnapshotsMetrics(t *testing.T) {
|
||||||
fs, client := startFakeFiler(t)
|
fs, client := startFakeFiler(t)
|
||||||
|
|
||||||
|
|||||||
@@ -520,10 +520,8 @@ func (h *Handler) rewriteManifests(
|
|||||||
SequenceNumber: cs.SequenceNumber + 1,
|
SequenceNumber: cs.SequenceNumber + 1,
|
||||||
TimestampMs: time.Now().UnixMilli(),
|
TimestampMs: time.Now().UnixMilli(),
|
||||||
ManifestList: manifestListLocation,
|
ManifestList: manifestListLocation,
|
||||||
Summary: &table.Summary{
|
// Merging manifests changes no file, so the totals carry over.
|
||||||
Operation: table.OpReplace,
|
Summary: snapshotSummary{}.build(table.OpReplace, cs, map[string]string{"maintenance": "rewrite_manifests"}),
|
||||||
Properties: map[string]string{"maintenance": "rewrite_manifests"},
|
|
||||||
},
|
|
||||||
SchemaID: func() *int {
|
SchemaID: func() *int {
|
||||||
id := schema.ID
|
id := schema.ID
|
||||||
return &id
|
return &id
|
||||||
|
|||||||
@@ -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}
|
||||||
|
}
|
||||||
Reference in new issue
Block a user