You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iceberg.apache.org by at...@apache.org on 2026/09/01 10:33:01 UTC
(iceberg-go) branch main updated: perf(table): stream data manifest entries into tasks (#1974)
This is an automated email from the ASF dual-hosted git repository.
laskoviymishka pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg-go.git
The following commit(s) were added to refs/heads/main by this push:
new 719394a23 perf(table): stream data manifest entries into tasks (#1974)
719394a23 is described below
commit 719394a231554b427a54d7f2b6674c4ead539934
Author: Minh Vu <vu...@gmail.com>
AuthorDate: Tue Sep 1 12:32:56 2026 +0200
perf(table): stream data manifest entries into tasks (#1974)
* perf(table): stream data manifest entries into tasks
Signed-off-by: Minh Vu <vu...@gmail.com>
* fix(table): satisfy lint checks for streamed planning
Signed-off-by: Minh Vu <vu...@gmail.com>
* test(table): cover inaccurate manifest entry counts
Signed-off-by: Minh Vu <vu...@gmail.com>
* fix(table): preserve scan planning behavior after rebase
---------
Signed-off-by: Minh Vu <vu...@gmail.com>
---
table/scan_metrics.go | 16 +-
table/scanner.go | 389 ++++++++++++++++++++++++++-------
table/scanner_internal_test.go | 88 ++++++++
table/scanner_plan_tasks_bench_test.go | 193 ++++++++++++++++
4 files changed, 604 insertions(+), 82 deletions(-)
diff --git a/table/scan_metrics.go b/table/scan_metrics.go
index 9ab7a5ae3..94022576f 100644
--- a/table/scan_metrics.go
+++ b/table/scan_metrics.go
@@ -32,16 +32,14 @@ import (
// scanMetricsAccumulator gathers scan-planning counts. Every field is written
// from a single goroutine — the manifest counts in
// fetchPartitionSpecFilteredManifestsWithSchema and the result/delete counts in
-// planFilesLocal after collectManifestEntriesWithSchema's concurrency barrier —
-// so plain integers are race-free; no field is touched from the concurrent
-// manifest workers.
+// planFilesLocal after its concurrent manifest/task barrier — so plain
+// integers are race-free; no field is touched from the concurrent workers.
//
-// 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.
+// 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
+// manifest-entry 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 aec98bfea..c30c76ed8 100644
--- a/table/scanner.go
+++ b/table/scanner.go
@@ -375,30 +375,53 @@ func openManifest(io io.IO, manifest iceberg.ManifestFile,
) ([]iceberg.ManifestEntry, error) {
// Counts may be -1 (unset) on V1 manifests, so clamp before allocating.
out := make([]iceberg.ManifestEntry, 0, max(0, int(manifest.AddedDataFiles())+int(manifest.ExistingDataFiles())))
- for entry, err := range manifest.Entries(io, true) {
+ if err := streamManifest(io, manifest, partitionFilter, metricsEval, func(entry iceberg.ManifestEntry) error {
+ out = append(out, entry)
+
+ return nil
+ }); err != nil {
+ return nil, err
+ }
+
+ return out, nil
+}
+
+// streamManifest reads live entries from a manifest, applying partition and
+// metrics filters before passing each match to visit. It deliberately does not
+// retain the entries, so callers can choose the smallest representation needed
+// for the next planning step.
+func streamManifest(manifestIO io.IO, manifest iceberg.ManifestFile,
+ partitionFilter, metricsEval func(iceberg.DataFile) (bool, error),
+ visit func(iceberg.ManifestEntry) error,
+) error {
+ for entry, err := range manifest.Entries(manifestIO, true) {
if err != nil {
- return nil, err
+ return err
}
- p, err := partitionFilter(entry.DataFile())
+ dataFile := entry.DataFile()
+ use, err := partitionFilter(dataFile)
if err != nil {
- return nil, err
+ return err
}
- if !p {
+ if !use {
continue
}
- m, err := metricsEval(entry.DataFile())
+ use, err = metricsEval(dataFile)
if err != nil {
- return nil, err
+ return err
+ }
+ if !use {
+ continue
}
- if m {
- out = append(out, entry)
+ if err := visit(entry); err != nil {
+ return err
}
}
- return out, nil
+ return nil
}
// IsDeletionVector reports whether df is a deletion vector: a Puffin file with
@@ -1073,6 +1096,17 @@ func (scan *Scan) collectManifestEntriesWithSchema(
manifestList []iceberg.ManifestFile,
schema *iceberg.Schema,
partitionFilters *keyDefaultMapErr[int, iceberg.BooleanExpression],
+) (*manifestEntries, error) {
+ return scan.collectManifestEntriesWithSchemaMinSequenceNum(
+ ctx, manifestList, schema, partitionFilters, minSequenceNum(manifestList))
+}
+
+func (scan *Scan) collectManifestEntriesWithSchemaMinSequenceNum(
+ ctx context.Context,
+ manifestList []iceberg.ManifestFile,
+ schema *iceberg.Schema,
+ partitionFilters *keyDefaultMapErr[int, iceberg.BooleanExpression],
+ minSeqNum int64,
) (*manifestEntries, error) {
metricsEval, err := newInclusiveMetricsEvaluator(
schema,
@@ -1084,7 +1118,6 @@ func (scan *Scan) collectManifestEntriesWithSchema(
return nil, err
}
- minSeqNum := minSequenceNum(manifestList)
concurrencyLimit := min(scan.concurrency, len(manifestList))
manifestIO := newManifestIOBatch(scan.ioF, concurrencyLimit)
@@ -1131,6 +1164,238 @@ func (scan *Scan) collectManifestEntriesWithSchema(
return flattenClassifiedManifestEntries(manifestResults), nil
}
+func splitManifestList(manifestList []iceberg.ManifestFile) (dataManifests, deleteManifests []iceberg.ManifestFile) {
+ dataManifests = make([]iceberg.ManifestFile, 0, len(manifestList))
+ deleteManifests = make([]iceberg.ManifestFile, 0, len(manifestList))
+ for _, manifest := range manifestList {
+ if manifest.ManifestContent() == iceberg.ManifestContentDeletes {
+ deleteManifests = append(deleteManifests, manifest)
+ } else {
+ dataManifests = append(dataManifests, manifest)
+ }
+ }
+
+ return dataManifests, deleteManifests
+}
+
+func manifestTaskCapacity(manifest iceberg.ManifestFile) (int, bool) {
+ addedFiles, existingFiles := manifest.AddedDataFiles(), manifest.ExistingDataFiles()
+ if addedFiles < 0 || existingFiles < 0 {
+ return 0, false
+ }
+
+ capacity := int64(addedFiles) + int64(existingFiles)
+ if capacity > int64(^uint(0)>>1) {
+ return 0, false
+ }
+
+ return int(capacity), true
+}
+
+// planDataManifestTasks streams filtered data entries into task batches. The
+// batches are kept per manifest while reads run concurrently, then flattened
+// in manifest-list order so task ordering does not depend on read completion.
+// Manifests with reliable live-entry counts share one task buffer; manifests
+// with unknown counts use independent batches and the same final flattening.
+func (scan *Scan) planDataManifestTasks(
+ ctx context.Context,
+ manifestList []iceberg.ManifestFile,
+ schema *iceberg.Schema,
+ minSeqNum int64,
+ posDeleteIndex *positionalDeleteIndex,
+ dvIndex map[string]iceberg.ManifestEntry,
+ eqDeleteIndex *equalityDeleteIndex,
+) ([]FileScanTask, error) {
+ metricsEval, err := newInclusiveMetricsEvaluator(
+ schema,
+ scan.rowFilter,
+ scan.caseSensitive,
+ scan.options["include_empty_files"] == "true",
+ )
+ if err != nil {
+ return nil, err
+ }
+
+ taskBatches := make([][]FileScanTask, len(manifestList))
+ taskCapacities := make([]int, len(manifestList))
+ taskOffsets := make([]int, len(manifestList))
+ directBuffer := scan.rowFilter == nil || scan.rowFilter.Equals(iceberg.AlwaysTrue{})
+ totalCapacity := 0
+ maxInt := int(^uint(0) >> 1)
+ if directBuffer {
+ for index, manifest := range manifestList {
+ if !scan.checkSequenceNumber(minSeqNum, manifest) {
+ continue
+ }
+
+ capacity, ok := manifestTaskCapacity(manifest)
+ if !ok || capacity > maxInt-totalCapacity {
+ directBuffer = false
+
+ break
+ }
+ taskCapacities[index] = capacity
+ taskOffsets[index] = totalCapacity
+ totalCapacity += capacity
+ }
+ }
+ var taskBuffer []FileScanTask
+ if directBuffer {
+ taskBuffer = make([]FileScanTask, totalCapacity)
+ }
+
+ concurrencyLimit := min(scan.concurrency, len(manifestList))
+ manifestIO := newManifestIOBatch(scan.ioF, concurrencyLimit)
+ g, gctx := errgroup.WithContext(ctx)
+ g.SetLimit(concurrencyLimit)
+
+ partitionFilters := scan.partitionFiltersForSchema(schema)
+ partitionEvaluators := newKeyDefaultMapWrapErr(func(specID int) (func(iceberg.DataFile) (bool, error), error) {
+ return buildPartitionEvaluator(specID, scan.metadata, schema, partitionFilters, scan.caseSensitive)
+ })
+
+ for index, manifest := range manifestList {
+ if !scan.checkSequenceNumber(minSeqNum, manifest) {
+ continue
+ }
+
+ g.Go(func() error {
+ fs, err := manifestIO.acquire(gctx)
+ if err != nil {
+ return err
+ }
+ partEval, err := partitionEvaluators.Get(int(manifest.PartitionSpecID()))
+ if err != nil {
+ return fmt.Errorf("failed to build partition evaluator for spec %d: %w", manifest.PartitionSpecID(), err)
+ }
+
+ var tasks []FileScanTask
+ if directBuffer {
+ offset := taskOffsets[index]
+ capacity := taskCapacities[index]
+ tasks = taskBuffer[offset : offset+capacity : offset+capacity]
+ tasks = tasks[:0]
+ } else {
+ tasks = make([]FileScanTask, 0,
+ max(0, int(manifest.AddedDataFiles())+int(manifest.ExistingDataFiles())))
+ }
+ err = streamManifest(fs, manifest, partEval, metricsEval, func(entry iceberg.ManifestEntry) error {
+ dataFile := entry.DataFile()
+ if dataFile.ContentType() != iceberg.EntryContentData {
+ return fmt.Errorf("%w: data manifest contains %s file %q",
+ ErrInvalidMetadata, dataFile.ContentType(), dataFile.FilePath())
+ }
+
+ task, err := fileScanTaskForDataEntry(entry, posDeleteIndex, dvIndex, eqDeleteIndex)
+ if err != nil {
+ return err
+ }
+ tasks = append(tasks, task)
+
+ return nil
+ })
+ if err != nil {
+ return err
+ }
+
+ taskBatches[index] = tasks
+
+ return nil
+ })
+ }
+
+ if err := g.Wait(); err != nil {
+ return nil, err
+ }
+
+ if directBuffer {
+ overflowed := false
+ for index, tasks := range taskBatches {
+ if len(tasks) > taskCapacities[index] {
+ overflowed = true
+
+ break
+ }
+ }
+ if !overflowed {
+ writeIndex := 0
+ for index, tasks := range taskBatches {
+ count := len(tasks)
+ if count > 0 {
+ sourceOffset := taskOffsets[index]
+ copy(taskBuffer[writeIndex:writeIndex+count],
+ taskBuffer[sourceOffset:sourceOffset+count])
+ writeIndex += count
+ }
+ taskBatches[index] = nil
+ }
+
+ return taskBuffer[:writeIndex:writeIndex], nil
+ }
+ }
+
+ totalTasks := 0
+ for _, tasks := range taskBatches {
+ totalTasks += len(tasks)
+ }
+ results := make([]FileScanTask, 0, totalTasks)
+ for i, tasks := range taskBatches {
+ results = append(results, tasks...)
+ // Release the per-manifest backing array as soon as it has been copied
+ // into the final result slice.
+ taskBatches[i] = nil
+ }
+
+ return results, nil
+}
+
+func fileScanTaskForDataEntry(
+ entry iceberg.ManifestEntry,
+ posDeleteIndex *positionalDeleteIndex,
+ dvIndex map[string]iceberg.ManifestEntry,
+ eqDeleteIndex *equalityDeleteIndex,
+) (FileScanTask, error) {
+ // Spec §Scan Planning: when a deletion vector applies to a data file,
+ // positional-delete files must NOT be applied. The DV is guaranteed to
+ // encode all prior pos-delete positions; reading the pos-delete Parquet too
+ // would be wasteful I/O, and on a buggy writer whose DV omits prior
+ // positions, applying both would over-delete. Mirrors Java's
+ // DeleteFileIndex.forDataFile.
+ dvFiles := matchDVToData(entry, dvIndex)
+ var deleteFiles []iceberg.DataFile
+ if len(dvFiles) == 0 {
+ var err error
+ deleteFiles, err = posDeleteIndex.forDataFile(entry)
+ if err != nil {
+ return FileScanTask{}, err
+ }
+ }
+ eqDeleteFiles, err := eqDeleteIndex.forDataFile(entry)
+ if err != nil {
+ return FileScanTask{}, err
+ }
+
+ dataFile := entry.DataFile()
+ task := FileScanTask{
+ File: dataFile,
+ DeleteFiles: deleteFiles,
+ EqualityDeleteFiles: eqDeleteFiles,
+ DeletionVectorFiles: dvFiles,
+ Start: 0,
+ Length: dataFile.FileSizeBytes(),
+ }
+ // Row lineage constants: readers use these to synthesize _row_id and
+ // _last_updated_sequence_number when requested. Per spec the synthesized
+ // _last_updated_sequence_number is the manifest entry's data sequence
+ // number (field id 3), not file sequence number (field id 4).
+ task.FirstRowID = dataFile.FirstRowID()
+ if seq := entry.SequenceNum(); seq >= 0 {
+ task.DataSequenceNumber = &seq
+ }
+
+ return task, nil
+}
+
// PlanFiles orchestrates the fetching and filtering of manifests, building a
// list of FileScanTasks that match the current Scan criteria. When planning
// happens locally it times the whole operation and emits a ScanReport to the
@@ -1250,72 +1515,63 @@ func (scan *Scan) planFilesLocal(ctx context.Context, acc *scanMetricsAccumulato
}
}
- // Step 2: Read manifest entries concurrently, accumulating data and positional deletes.
- entries, err := scan.collectManifestEntriesWithSchema(ctx, manifestList, schema, partitionFilters)
- if err != nil {
- return nil, err
- }
-
- var boundRowFilter iceberg.BooleanExpression
- if scan.rowFilter != nil && !scan.rowFilter.Equals(iceberg.AlwaysTrue{}) {
- boundRowFilter, err = iceberg.BindExpr(schema, scan.rowFilter, scan.caseSensitive)
+ // Step 2: Read delete manifests first so data entries can be turned into
+ // tasks immediately after their delete indexes are ready.
+ dataManifests, deleteManifests := splitManifestList(manifestList)
+ minSeqNum := minSequenceNum(manifestList)
+ deleteEntries := newManifestEntries()
+ if len(deleteManifests) > 0 {
+ deleteEntries, err = scan.collectManifestEntriesWithSchemaMinSequenceNum(
+ ctx, deleteManifests, schema, partitionFilters, minSeqNum)
if err != nil {
return nil, err
}
}
- var residualEvaluators map[int]*partitionResidualEvaluator
- if boundRowFilter != nil {
- residualEvaluators = make(map[int]*partitionResidualEvaluator)
- }
-
// Step 3: Index positional deletes and match them to data files.
- posDeleteIndex, err := buildPositionalDeleteIndex(entries.positionalDeleteEntries)
+ posDeleteIndex, err := buildPositionalDeleteIndex(deleteEntries.positionalDeleteEntries)
if err != nil {
return nil, err
}
- dvIndex, err := buildDVIndex(entries.dvEntries)
+ dvIndex, err := buildDVIndex(deleteEntries.dvEntries)
if err != nil {
return nil, err
}
- eqDeleteIndex, err := buildEqualityDeleteIndex(entries.equalityDeleteEntries, scan.metadata, schema)
+ eqDeleteIndex, err := buildEqualityDeleteIndex(deleteEntries.equalityDeleteEntries, scan.metadata, schema)
if err != nil {
return nil, err
}
- results = make([]FileScanTask, 0, len(entries.dataEntries))
- splitTargetSize := scan.metadata.Properties().GetInt64(
- ReadSplitTargetSizeKey, ReadSplitTargetSizeDefault)
- for _, e := range entries.dataEntries {
- // Spec §Scan Planning: when a deletion vector applies to a data
- // file, positional-delete files must NOT be applied. The DV is
- // guaranteed to encode all prior pos-delete positions; reading the
- // pos-delete Parquet too would be wasteful I/O, and on a buggy
- // writer whose DV omits prior positions, applying both would
- // over-delete. Mirrors Java's DeleteFileIndex.forDataFile.
- dvFiles := matchDVToData(e, dvIndex)
- var deleteFiles []iceberg.DataFile
- if len(dvFiles) == 0 {
- deleteFiles, err = posDeleteIndex.forDataFile(e)
- if err != nil {
- return nil, err
- }
- }
- eqDeleteFiles, err := eqDeleteIndex.forDataFile(e)
+ // Step 4: Stream data entries into per-manifest task batches, then flatten
+ // them in manifest order.
+ plannedTasks, err := scan.planDataManifestTasks(
+ ctx, dataManifests, schema, minSeqNum, posDeleteIndex, dvIndex, eqDeleteIndex)
+ if err != nil {
+ return nil, err
+ }
+
+ var boundRowFilter iceberg.BooleanExpression
+ if scan.rowFilter != nil && !scan.rowFilter.Equals(iceberg.AlwaysTrue{}) {
+ boundRowFilter, err = iceberg.BindExpr(schema, scan.rowFilter, scan.caseSensitive)
if err != nil {
return nil, err
}
+ }
+ var residualEvaluators map[int]*partitionResidualEvaluator
+ if boundRowFilter != nil {
+ residualEvaluators = make(map[int]*partitionResidualEvaluator)
+ }
- task := FileScanTask{
- File: e.DataFile(),
- DeleteFiles: deleteFiles,
- EqualityDeleteFiles: eqDeleteFiles,
- DeletionVectorFiles: dvFiles,
- Start: 0,
- Length: e.DataFile().FileSizeBytes(),
- }
+ // Apply residuals, metrics, and parquet range splitting in manifest order.
+ // These operations stay outside the concurrent manifest workers so the
+ // residual evaluator cache and scan metrics remain race-free.
+ results = make([]FileScanTask, 0, len(plannedTasks))
+ splitTargetSize := scan.metadata.Properties().GetInt64(
+ ReadSplitTargetSizeKey, ReadSplitTargetSizeDefault)
+ acc.resultDataFiles = int64(len(plannedTasks))
+ for _, task := range plannedTasks {
if boundRowFilter != nil {
- specID := int(e.DataFile().SpecID())
+ specID := int(task.File.SpecID())
residualEvaluator, found := residualEvaluators[specID]
if !found {
residualEvaluator, err = newPartitionResidualEvaluator(
@@ -1327,36 +1583,23 @@ func (scan *Scan) planFilesLocal(ctx context.Context, acc *scanMetricsAccumulato
}
if residualEvaluator != nil {
var simplified bool
- task.Residual, simplified, err = residualEvaluator.residual(dataFilePartition(e.DataFile()))
+ task.Residual, simplified, err = residualEvaluator.residual(dataFilePartition(task.File))
if err != nil {
- return nil, fmt.Errorf("evaluate partition residual for %s: %w", e.DataFile().FilePath(), err)
+ return nil, fmt.Errorf("evaluate partition residual for %s: %w", task.File.FilePath(), err)
}
if !simplified {
task.Residual = nil
}
}
}
- // Row lineage constants: readers use these to synthesize _row_id and
- // _last_updated_sequence_number when requested. Per spec the
- // synthesized _last_updated_sequence_number is the manifest entry's
- // data sequence number (field id 3), not file sequence number
- // (field id 4); back-dated EXISTING entries can have the two
- // diverge and Java/iceberg-rust use the data sequence number.
- task.FirstRowID = e.DataFile().FirstRowID()
- if seq := e.SequenceNum(); seq >= 0 {
- s := seq
- task.DataSequenceNumber = &s
- }
- // Metrics count data files and their deletes once per base file. The
- // execution task can contain one range per row group.
- acc.resultDataFiles++
+
acc.addResultDeleteMetrics(task)
+ acc.totalFileSize += task.File.FileSizeBytes()
if splitTasks, split := splitParquetScanTask(task, splitTargetSize); split {
results = append(results, splitTasks...)
} else {
results = append(results, task)
}
- acc.totalFileSize += e.DataFile().FileSizeBytes()
}
return results, nil
diff --git a/table/scanner_internal_test.go b/table/scanner_internal_test.go
index 206642109..1380a648d 100644
--- a/table/scanner_internal_test.go
+++ b/table/scanner_internal_test.go
@@ -893,6 +893,71 @@ func TestMinSequenceNum(t *testing.T) {
}
}
+func TestSplitManifestListKeepsDataManifestOrder(t *testing.T) {
+ manifests := []iceberg.ManifestFile{
+ newDeleteManifest(1),
+ newDataManifest(2),
+ newDeleteManifest(3),
+ newDataManifest(4),
+ }
+
+ dataManifests, deleteManifests := splitManifestList(manifests)
+
+ assert.Equal(t, []iceberg.ManifestFile{manifests[1], manifests[3]}, dataManifests)
+ assert.Equal(t, []iceberg.ManifestFile{manifests[0], manifests[2]}, deleteManifests)
+}
+
+func TestPlanDataManifestTasksPreservesManifestOrder(t *testing.T) {
+ fs := newTrackingIO()
+ schema := simpleSchema()
+ meta, err := NewMetadata(schema, iceberg.UnpartitionedSpec, UnsortedSortOrder,
+ "mem://streamed-planning", nil)
+ require.NoError(t, err)
+
+ const snapshotID = int64(1)
+ manifests := []iceberg.ManifestFile{
+ writeManifest(t, fs, snapshotID, 1,
+ "mem://streamed-planning/metadata/manifest-a.avro",
+ "mem://streamed-planning/data-a.parquet"),
+ writeManifest(t, fs, snapshotID, 2,
+ "mem://streamed-planning/metadata/manifest-b.avro",
+ "mem://streamed-planning/data-b.parquet"),
+ }
+ scan := &Scan{
+ metadata: meta,
+ rowFilter: iceberg.AlwaysTrue{},
+ caseSensitive: true,
+ concurrency: 2,
+ ioF: testFSF(fs),
+ }
+
+ posDeleteIndex, err := buildPositionalDeleteIndex(nil)
+ require.NoError(t, err)
+ dvIndex, err := buildDVIndex(nil)
+ require.NoError(t, err)
+ eqDeleteIndex, err := buildEqualityDeleteIndex(nil, meta, schema)
+ require.NoError(t, err)
+
+ tasks, err := scan.planDataManifestTasks(context.Background(), manifests, schema, 0,
+ posDeleteIndex, dvIndex, eqDeleteIndex)
+ require.NoError(t, err)
+ require.Len(t, tasks, 2)
+ assert.Equal(t, "mem://streamed-planning/data-a.parquet", tasks[0].File.FilePath())
+ assert.Equal(t, "mem://streamed-planning/data-b.parquet", tasks[1].File.FilePath())
+
+ unknownCounts := iceberg.NewManifestFile(
+ 2, manifests[0].FilePath(), manifests[0].Length(), 0, snapshotID,
+ ).
+ SequenceNum(1, 1).
+ AddedFiles(-1).
+ ExistingFiles(-1).
+ Build()
+ tasks, err = scan.planDataManifestTasks(context.Background(), []iceberg.ManifestFile{unknownCounts}, schema, 0,
+ posDeleteIndex, dvIndex, eqDeleteIndex)
+ require.NoError(t, err)
+ assert.Len(t, tasks, 1, "unknown manifest counts should use the safe append path")
+}
+
func TestSplitLineageMetadataFields(t *testing.T) {
tests := []struct {
name string
@@ -2335,3 +2400,26 @@ func TestArrowScanFiltersMissingColumnInitialDefault(t *testing.T) {
}
})
}
+
+func TestPlanDataManifestTasksHandlesInaccurateCounts(t *testing.T) {
+ for _, counts := range [][]int32{{5, 5, 5}, {0, 2, 1}, {-1, -1, -1}} {
+ scan, schema, manifests, posIndex, dvIndex, eqIndex := newPlanTasksBenchmarkFixture(t, 3, 3)
+ for i, manifest := range manifests {
+ manifests[i] = iceberg.NewManifestFile(2, manifest.FilePath(), manifest.Length(),
+ manifest.PartitionSpecID(), manifest.SnapshotID()).
+ SequenceNum(manifest.SequenceNum(), manifest.MinSequenceNum()).
+ AddedFiles(counts[i]).
+ Build()
+ }
+
+ tasks, err := scan.planDataManifestTasks(t.Context(), manifests, schema,
+ minSequenceNum(manifests), posIndex, dvIndex, eqIndex)
+ require.NoError(t, err)
+ require.Len(t, tasks, 9)
+ for i, task := range tasks {
+ assert.Equal(t, "mem://plan-tasks-benchmark/data-"+strconv.Itoa(i/3)+"-"+strconv.Itoa(i%3)+".parquet",
+ task.File.FilePath(), "manifest counts: %v", counts)
+ assert.Equal(t, int64(i/3+1), *task.DataSequenceNumber)
+ }
+ }
+}
diff --git a/table/scanner_plan_tasks_bench_test.go b/table/scanner_plan_tasks_bench_test.go
new file mode 100644
index 000000000..052696012
--- /dev/null
+++ b/table/scanner_plan_tasks_bench_test.go
@@ -0,0 +1,193 @@
+// 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"
+ "runtime"
+ "testing"
+
+ "github.com/apache/iceberg-go"
+)
+
+var planTasksBenchmarkSink int
+
+func BenchmarkPlanDataManifestTasks(b *testing.B) {
+ for _, workload := range []struct {
+ manifestCount int
+ entryCount int
+ }{
+ {manifestCount: 8, entryCount: 1_000},
+ {manifestCount: 32, entryCount: 1_000},
+ {manifestCount: 8, entryCount: 10_000},
+ } {
+ b.Run(fmt.Sprintf("manifests=%d/entries=%d", workload.manifestCount, workload.entryCount), func(b *testing.B) {
+ scan, schema, manifests, posDeleteIndex, dvIndex, eqDeleteIndex := newPlanTasksBenchmarkFixture(b, workload.manifestCount, workload.entryCount)
+
+ for _, mode := range []struct {
+ name string
+ plan func(context.Context) ([]FileScanTask, error)
+ }{
+ {
+ name: "buffered",
+ plan: func(ctx context.Context) ([]FileScanTask, error) {
+ return planBufferedDataManifestTasks(scan, ctx, manifests, schema,
+ posDeleteIndex, dvIndex, eqDeleteIndex)
+ },
+ },
+ {
+ name: "streamed",
+ plan: func(ctx context.Context) ([]FileScanTask, error) {
+ return scan.planDataManifestTasks(ctx, manifests, schema,
+ minSequenceNum(manifests), posDeleteIndex, dvIndex, eqDeleteIndex)
+ },
+ },
+ } {
+ b.Run(mode.name, func(b *testing.B) {
+ b.ReportAllocs()
+ b.ResetTimer()
+ for b.Loop() {
+ tasks, err := mode.plan(context.Background())
+ if err != nil {
+ b.Fatal(err)
+ }
+ planTasksBenchmarkSink = len(tasks)
+ runtime.KeepAlive(tasks)
+ }
+ })
+ }
+ if planTasksBenchmarkSink != workload.manifestCount*workload.entryCount {
+ b.Fatalf("planned %d tasks, want %d", planTasksBenchmarkSink,
+ workload.manifestCount*workload.entryCount)
+ }
+ })
+ }
+}
+
+func newPlanTasksBenchmarkFixture(
+ b testing.TB,
+ manifestCount, entryCount int,
+) (*Scan, *iceberg.Schema, []iceberg.ManifestFile, *positionalDeleteIndex, map[string]iceberg.ManifestEntry, *equalityDeleteIndex) {
+ b.Helper()
+
+ fs := newTrackingIO()
+ schema := simpleSchema()
+ spec := iceberg.NewPartitionSpec()
+ meta, err := NewMetadata(schema, &spec, UnsortedSortOrder, "mem://plan-tasks-benchmark", nil)
+ if err != nil {
+ b.Fatal(err)
+ }
+
+ snapshotID := int64(1)
+ manifests := make([]iceberg.ManifestFile, manifestCount)
+ for manifestIndex := range manifestCount {
+ entries := make([]iceberg.ManifestEntry, entryCount)
+ sequenceNumber := int64(manifestIndex + 1)
+ for entryIndex := range entryCount {
+ dataPath := fmt.Sprintf("mem://plan-tasks-benchmark/data-%d-%d.parquet", manifestIndex, entryIndex)
+ dataFile, err := iceberg.NewDataFileBuilder(
+ spec,
+ iceberg.EntryContentData,
+ dataPath,
+ iceberg.ParquetFile,
+ nil,
+ nil,
+ nil,
+ 1,
+ 1024,
+ )
+ if err != nil {
+ b.Fatal(err)
+ }
+ entries[entryIndex] = iceberg.NewManifestEntryBuilder(
+ iceberg.EntryStatusADDED,
+ &snapshotID,
+ dataFile.Build(),
+ ).SequenceNum(sequenceNumber).Build()
+ }
+
+ manifestPath := fmt.Sprintf("mem://plan-tasks-benchmark/metadata/manifest-%d.avro", manifestIndex)
+ var manifestBuf bytes.Buffer
+ _, err = iceberg.WriteManifest(manifestPath, &manifestBuf, 2, spec, schema, snapshotID, entries)
+ if err != nil {
+ b.Fatal(err)
+ }
+ if err := fs.WriteFile(manifestPath, manifestBuf.Bytes()); err != nil {
+ b.Fatal(err)
+ }
+ manifests[manifestIndex] = iceberg.NewManifestFile(
+ 2, manifestPath, int64(manifestBuf.Len()), int32(spec.ID()), snapshotID,
+ ).
+ SequenceNum(sequenceNumber, sequenceNumber).
+ AddedFiles(int32(entryCount)).
+ AddedRows(int64(entryCount)).
+ Build()
+ }
+
+ posDeleteIndex, err := buildPositionalDeleteIndex(nil)
+ if err != nil {
+ b.Fatal(err)
+ }
+ dvIndex, err := buildDVIndex(nil)
+ if err != nil {
+ b.Fatal(err)
+ }
+ eqDeleteIndex, err := buildEqualityDeleteIndex(nil, meta, schema)
+ if err != nil {
+ b.Fatal(err)
+ }
+
+ scan := &Scan{
+ metadata: meta,
+ rowFilter: iceberg.AlwaysTrue{},
+ caseSensitive: true,
+ concurrency: 4,
+ ioF: testFSF(fs),
+ }
+
+ return scan, schema, manifests, posDeleteIndex, dvIndex, eqDeleteIndex
+}
+
+func planBufferedDataManifestTasks(
+ scan *Scan,
+ ctx context.Context,
+ manifests []iceberg.ManifestFile,
+ schema *iceberg.Schema,
+ posDeleteIndex *positionalDeleteIndex,
+ dvIndex map[string]iceberg.ManifestEntry,
+ eqDeleteIndex *equalityDeleteIndex,
+) ([]FileScanTask, error) {
+ entries, err := scan.collectManifestEntriesWithSchema(
+ ctx, manifests, schema, scan.partitionFiltersForSchema(schema))
+ if err != nil {
+ return nil, err
+ }
+
+ tasks := make([]FileScanTask, 0, len(entries.dataEntries))
+ for _, entry := range entries.dataEntries {
+ task, err := fileScanTaskForDataEntry(entry, posDeleteIndex, dvIndex, eqDeleteIndex)
+ if err != nil {
+ return nil, err
+ }
+ tasks = append(tasks, task)
+ }
+
+ return tasks, nil
+}