perf(table): skip manifests with no live entries

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

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++

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
+}

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