From 673ba7a95b48681b22fe27ab66812c6ff47b1667 Mon Sep 17 00:00:00 2001 From: Minh Vu Date: Sat, 29 Aug 2026 11:47:31 +0200 Subject: [PATCH] perf(table): skip manifests with no live entries Signed-off-by: Minh Vu --- table/scan_metrics.go | 11 +- table/scanner.go | 11 ++ table/scanner_empty_manifest_bench_test.go | 165 +++++++++++++++++++++ table/scanner_internal_test.go | 123 +++++++++++++++ 4 files changed, 305 insertions(+), 5 deletions(-) create mode 100644 table/scanner_empty_manifest_bench_test.go diff --git a/table/scan_metrics.go b/table/scan_metrics.go index 996b94fbf..d88a2f90f 100644 --- a/table/scan_metrics.go +++ b/table/scan_metrics.go @@ -36,11 +36,12 @@ import ( // so plain integers are race-free; no field is touched from the concurrent // manifest workers. // -// The scanned/skipped manifest counts measure partition-spec pruning (the -// filter applied while listing manifests). The per-entry skipped-data-files / -// skipped-delete-files counts (which would be incremented inside the concurrent -// openManifest loop, and would use atomics) and indexed-delete-files are left -// for a follow-up and omitted rather than reported as zero. +// The scanned/skipped manifest counts measure manifest selection before opening +// manifests, including partition-spec pruning and known-empty manifests. The +// per-entry skipped-data-files / skipped-delete-files counts (which would be +// incremented inside the concurrent openManifest loop, and would use atomics) +// and indexed-delete-files are left for a follow-up and omitted rather than +// reported as zero. type scanMetricsAccumulator struct { totalDataManifests int64 totalDeleteManifests int64 diff --git a/table/scanner.go b/table/scanner.go index f5599d932..1f701d3da 100644 --- a/table/scanner.go +++ b/table/scanner.go @@ -918,6 +918,17 @@ func (scan *Scan) filterManifestsWithSchema( if err != nil { return nil, fmt.Errorf("failed to evaluate manifest %s: %w", mf.FilePath(), err) } + // Has*Files returns true for unknown counts, so this only skips manifests + // known to contain no added or existing (live) entries. + if use && !mf.HasAddedFiles() && !mf.HasExistingFiles() { + if isDelete { + acc.skippedDeleteManifests++ + } else { + acc.skippedDataManifests++ + } + + continue + } if use { if isDelete { acc.scannedDeleteManifests++ diff --git a/table/scanner_empty_manifest_bench_test.go b/table/scanner_empty_manifest_bench_test.go new file mode 100644 index 000000000..1d21f6f5f --- /dev/null +++ b/table/scanner_empty_manifest_bench_test.go @@ -0,0 +1,165 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package table + +import ( + "bytes" + "context" + "fmt" + "testing" + + "github.com/apache/iceberg-go" + "github.com/stretchr/testify/require" +) + +var skipEmptyManifestsBenchmarkSink int + +type emptyManifestBenchmarkCase struct { + name string + manifestCount int + entriesPerManifest int + liveEvery int +} + +func BenchmarkPlanFilesSkipsKnownEmptyManifests(b *testing.B) { + for _, tc := range []emptyManifestBenchmarkCase{ + {name: "manifests=256/all-empty", manifestCount: 256, entriesPerManifest: 64}, + {name: "manifests=256/10pct-live", manifestCount: 256, entriesPerManifest: 64, liveEvery: 10}, + } { + b.Run(tc.name, func(b *testing.B) { + tbl, fs, manifestPaths, expectedTasks := newEmptyManifestBenchmarkTable(b, tc) + scan := tbl.Scan(WithMaxConcurrency(1)) + + b.ReportAllocs() + b.ResetTimer() + for b.Loop() { + tasks, err := scan.PlanFiles(context.Background()) + if err != nil { + b.Fatal(err) + } + if len(tasks) != expectedTasks { + b.Fatalf("PlanFiles returned %d tasks, want %d", len(tasks), expectedTasks) + } + skipEmptyManifestsBenchmarkSink = len(tasks) + } + b.StopTimer() + + manifestOpens := 0 + for _, path := range manifestPaths { + manifestOpens += fs.openCount[path] + } + b.ReportMetric(float64(manifestOpens)/float64(b.N), "manifest-opens/op") + b.ReportMetric(float64(expectedTasks), "tasks/op") + }) + } +} + +func newEmptyManifestBenchmarkTable( + b testing.TB, + tc emptyManifestBenchmarkCase, +) (*Table, *trackingCallsIO, []string, int) { + b.Helper() + + const ( + tableLocation = "mem://empty-manifest-benchmark" + snapshotID = int64(1) + sequenceNum = int64(1) + ) + + fs := newTrackingCallsIO() + schema := simpleSchema() + spec := iceberg.NewPartitionSpec() + manifests := make([]iceberg.ManifestFile, 0, tc.manifestCount) + manifestPaths := make([]string, 0, tc.manifestCount) + expectedTasks := 0 + + for i := range tc.manifestCount { + manifestPath := fmt.Sprintf("%s/metadata/manifest-%d.avro", tableLocation, i) + dataPath := fmt.Sprintf("%s/data-%d.parquet", tableLocation, i) + live := tc.liveEvery > 0 && i%tc.liveEvery == 0 + manifest := writeEmptyManifestBenchmarkManifest( + b, fs, schema, spec, snapshotID, sequenceNum, manifestPath, dataPath, + tc.entriesPerManifest, live, + ) + manifests = append(manifests, manifest) + manifestPaths = append(manifestPaths, manifestPath) + if live { + expectedTasks += tc.entriesPerManifest + } + } + + manifestListPath := tableLocation + "/metadata/snap-1.avro" + var listBuf bytes.Buffer + require.NoError(b, iceberg.WriteManifestList(2, &listBuf, snapshotID, nil, ptr(sequenceNum), 0, manifests)) + require.NoError(b, fs.WriteFile(manifestListPath, listBuf.Bytes())) + + meta, err := NewMetadata(schema, &spec, UnsortedSortOrder, tableLocation, nil) + require.NoError(b, err) + builder, err := MetadataBuilderFromBase(meta, "") + require.NoError(b, err) + schemaID := meta.CurrentSchema().ID + require.NoError(b, builder.AddSnapshot(&Snapshot{ + SnapshotID: snapshotID, + SequenceNumber: sequenceNum, + TimestampMs: meta.LastUpdatedMillis() + 1, + ManifestList: manifestListPath, + Summary: &Summary{Operation: OpAppend}, + SchemaID: &schemaID, + })) + require.NoError(b, builder.SetSnapshotRef(MainBranch, snapshotID, BranchRef)) + built, err := builder.Build() + require.NoError(b, err) + + return New(Identifier{"db", "empty-manifest-benchmark"}, built, tableLocation+"/metadata/metadata.json", testFSF(fs), nil), fs, manifestPaths, expectedTasks +} + +func writeEmptyManifestBenchmarkManifest( + b testing.TB, + fs *trackingCallsIO, + schema *iceberg.Schema, + spec iceberg.PartitionSpec, + snapshotID, sequenceNum int64, + manifestPath, dataPath string, + entryCount int, + live bool, +) iceberg.ManifestFile { + b.Helper() + + entries := make([]iceberg.ManifestEntry, entryCount) + status := iceberg.EntryStatusDELETED + if live { + status = iceberg.EntryStatusADDED + } + for i := range entries { + dataFile, err := iceberg.NewDataFileBuilder( + spec, iceberg.EntryContentData, dataPath, iceberg.ParquetFile, + nil, nil, nil, 1, 1024, + ) + require.NoError(b, err) + entries[i] = iceberg.NewManifestEntryBuilder(status, &snapshotID, dataFile.Build()). + SequenceNum(sequenceNum). + Build() + } + + var manifestBuf bytes.Buffer + manifest, err := iceberg.WriteManifest(manifestPath, &manifestBuf, 2, spec, schema, snapshotID, entries) + require.NoError(b, err) + require.NoError(b, fs.WriteFile(manifestPath, manifestBuf.Bytes())) + + return manifest +} diff --git a/table/scanner_internal_test.go b/table/scanner_internal_test.go index af1d7e97b..bdec71436 100644 --- a/table/scanner_internal_test.go +++ b/table/scanner_internal_test.go @@ -877,15 +877,18 @@ func TestFetchManifestCountersWithRealSnapshot(t *testing.T) { dataScanned := iceberg.NewManifestFile(2, "mem://default/table-location/metadata/data-scanned.avro", 100, specID, snapshotID). Content(iceberg.ManifestContentData). + AddedFiles(1). Partitions([]iceberg.FieldSummary{{ContainsNull: false, LowerBound: match, UpperBound: match}}). Build() dataSkipped := iceberg.NewManifestFile(2, "mem://default/table-location/metadata/data-skipped.avro", 100, specID, snapshotID). Content(iceberg.ManifestContentData). + AddedFiles(1). Partitions([]iceberg.FieldSummary{{ContainsNull: false, LowerBound: noMatch, UpperBound: noMatch}}). Build() deleteScanned := iceberg.NewManifestFile(2, "mem://default/table-location/metadata/delete-scanned.avro", 100, specID, snapshotID). Content(iceberg.ManifestContentDeletes). SequenceNum(1, 1). + AddedFiles(1). Partitions([]iceberg.FieldSummary{{ContainsNull: false, LowerBound: match, UpperBound: match}}). Build() @@ -934,6 +937,125 @@ func TestFetchManifestCountersWithRealSnapshot(t *testing.T) { assert.Len(t, filtered, 2, "only the overlapping manifests survive partition-spec filtering") } +func TestFilterManifestsWithSchemaSkipsKnownEmptyManifests(t *testing.T) { + schema := simpleSchema() + metadata, err := NewMetadata( + schema, + iceberg.UnpartitionedSpec, + UnsortedSortOrder, + "mem://empty-manifest-filter", + nil, + ) + require.NoError(t, err) + + scan := &Scan{ + metadata: metadata, + rowFilter: iceberg.AlwaysTrue{}, + caseSensitive: true, + } + knownEmptyData := iceberg.NewManifestFile(2, "data-empty.avro", 100, 0, 1). + Content(iceberg.ManifestContentData). + AddedFiles(0). + ExistingFiles(0). + DeletedFiles(2). + Build() + knownEmptyDelete := iceberg.NewManifestFile(2, "delete-empty.avro", 100, 0, 1). + Content(iceberg.ManifestContentDeletes). + AddedFiles(0). + ExistingFiles(0). + DeletedFiles(1). + Build() + unknownCounts := iceberg.NewManifestFile(2, "data-unknown.avro", 100, 0, 1). + Content(iceberg.ManifestContentData). + AddedFiles(-1). + ExistingFiles(-1). + Build() + live := iceberg.NewManifestFile(2, "data-live.avro", 100, 0, 1). + Content(iceberg.ManifestContentData). + AddedFiles(0). + ExistingFiles(1). + Build() + + var acc scanMetricsAccumulator + filtered, err := scan.filterManifestsWithSchema( + []iceberg.ManifestFile{knownEmptyData, knownEmptyDelete, unknownCounts, live}, + schema, + &acc, + ) + require.NoError(t, err) + require.Len(t, filtered, 2) + assert.Equal(t, "data-unknown.avro", filtered[0].FilePath()) + assert.Equal(t, "data-live.avro", filtered[1].FilePath()) + + assert.Equal(t, int64(3), acc.totalDataManifests) + assert.Equal(t, int64(1), acc.totalDeleteManifests) + assert.Equal(t, int64(2), acc.scannedDataManifests) + assert.Equal(t, int64(1), acc.skippedDataManifests) + assert.Equal(t, int64(0), acc.scannedDeleteManifests) + assert.Equal(t, int64(1), acc.skippedDeleteManifests) + assert.Equal(t, acc.totalDataManifests, acc.scannedDataManifests+acc.skippedDataManifests) + assert.Equal(t, acc.totalDeleteManifests, acc.scannedDeleteManifests+acc.skippedDeleteManifests) +} + +func TestPlanFilesSkipsKnownEmptyManifestsBeforeOpening(t *testing.T) { + const ( + tableLocation = "mem://empty-manifest-plan" + snapshotID = int64(1) + manifestListPath = tableLocation + "/metadata/snap-1.avro" + ) + + fs := newTrackingCallsIO() + schema := simpleSchema() + spec := iceberg.NewPartitionSpec() + metadata, err := NewMetadata(schema, &spec, UnsortedSortOrder, tableLocation, nil) + require.NoError(t, err) + builder, err := MetadataBuilderFromBase(metadata, "") + require.NoError(t, err) + + dataPath := tableLocation + "/metadata/data-empty.avro" + deletePath := tableLocation + "/metadata/delete-empty.avro" + dataManifest := iceberg.NewManifestFile(2, dataPath, 100, int32(spec.ID()), snapshotID). + Content(iceberg.ManifestContentData). + SequenceNum(1, 1). + AddedFiles(0). + ExistingFiles(0). + DeletedFiles(2). + Build() + deleteManifest := iceberg.NewManifestFile(2, deletePath, 100, int32(spec.ID()), snapshotID). + Content(iceberg.ManifestContentDeletes). + SequenceNum(1, 1). + AddedFiles(0). + ExistingFiles(0). + DeletedFiles(1). + Build() + + var listBuf bytes.Buffer + sequenceNumber := int64(1) + require.NoError(t, iceberg.WriteManifestList(2, &listBuf, snapshotID, nil, &sequenceNumber, 0, + []iceberg.ManifestFile{dataManifest, deleteManifest})) + require.NoError(t, fs.WriteFile(manifestListPath, listBuf.Bytes())) + + schemaID := metadata.CurrentSchema().ID + require.NoError(t, builder.AddSnapshot(&Snapshot{ + SnapshotID: snapshotID, + SequenceNumber: sequenceNumber, + TimestampMs: metadata.LastUpdatedMillis() + 1, + ManifestList: manifestListPath, + Summary: &Summary{Operation: OpAppend}, + SchemaID: &schemaID, + })) + require.NoError(t, builder.SetSnapshotRef(MainBranch, snapshotID, BranchRef)) + built, err := builder.Build() + require.NoError(t, err) + + tbl := New(Identifier{"db", "empty-manifest-plan"}, built, tableLocation+"/metadata/metadata.json", testFSF(fs), nil) + tasks, err := tbl.Scan().PlanFiles(context.Background()) + require.NoError(t, err) + assert.Empty(t, tasks) + assert.Zero(t, fs.openCount[dataPath]) + assert.Zero(t, fs.openCount[deletePath]) +} + func TestBuildManifestEvaluatorWithInvalidSpecID(t *testing.T) { schema := iceberg.NewSchema( 1, @@ -1763,6 +1885,7 @@ func TestPlanFilesUnknownTransformDoesNotPrune(t *testing.T) { manifest := iceberg.NewManifestFile(2, manifestPath, int64(manifestBytes.Len()), int32(spec.ID()), snapshotID). Partitions([]iceberg.FieldSummary{{ContainsNull: false, LowerBound: &lower, UpperBound: &upper}}). SequenceNum(1, 1). + AddedFiles(1). Build() var listBytes bytes.Buffer