This is an automated email from the ASF dual-hosted git repository.
zeroshade 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 5cee5eb30 feat(metrics): report skipped-data-files and
skipped-delete-files in ScanReport (#2430)
5cee5eb30 is described below
commit 5cee5eb305c75bb312408863b04e787aed6d80c8
Author: Revanth <[email protected]>
AuthorDate: Wed Nov 4 11:34:14 2026 -0600
feat(metrics): report skipped-data-files and skipped-delete-files in
ScanReport (#2430)
---
table/incremental_append_scan.go | 1 +
table/incremental_changelog_scan.go | 1 +
table/scan_metrics.go | 44 +++++-
table/scan_metrics_test.go | 7 +-
table/scan_skipped_files_test.go | 245 ++++++++++++++++++++++++++++++
table/scanner.go | 35 ++++-
table/scanner_manifest_projection_test.go | 2 +
7 files changed, 323 insertions(+), 12 deletions(-)
diff --git a/table/incremental_append_scan.go b/table/incremental_append_scan.go
index 277caec92..dfce4336c 100644
--- a/table/incremental_append_scan.go
+++ b/table/incremental_append_scan.go
@@ -205,6 +205,7 @@ func (s *IncrementalAppendScan) PlanFiles(ctx
context.Context) ([]FileScanTask,
if err != nil {
return nil, err
}
+ acc.skippedDataFiles += entries.metrics.skippedDataFiles
tasks := make([]FileScanTask, 0, len(entries.dataEntries))
for _, entry := range entries.dataEntries {
diff --git a/table/incremental_changelog_scan.go
b/table/incremental_changelog_scan.go
index d1918b60f..1711eafc6 100644
--- a/table/incremental_changelog_scan.go
+++ b/table/incremental_changelog_scan.go
@@ -270,6 +270,7 @@ func (s *IncrementalChangelogScan) PlanFiles(ctx
context.Context) ([]ChangelogSc
if err != nil {
return nil, err
}
+ acc.skippedDataFiles += entries.metrics.skippedDataFiles
dataEntries = entries.dataEntries
}
diff --git a/table/scan_metrics.go b/table/scan_metrics.go
index a541d548d..194df7745 100644
--- a/table/scan_metrics.go
+++ b/table/scan_metrics.go
@@ -35,10 +35,13 @@ import (
// fields remain plain integers without atomics.
//
// 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.
+// filter applied while listing manifests). The skipped data/delete file counts
+// measure the next step: entries the scan reads from scanned manifests (live
+// entries for table scans) that the partition or inclusive metrics evaluator
+// rejected. Each manifest worker counts its own rejections and the planner
+// merges them after the workers finish. Entries in pruned manifests count only
+// toward the skipped manifest counts. indexed-delete-files is left for a
+// follow-up and omitted rather than reported as zero.
type scanMetricsAccumulator struct {
totalDataManifests int64
totalDeleteManifests int64
@@ -46,6 +49,8 @@ type scanMetricsAccumulator struct {
scannedDeleteManifests int64
skippedDataManifests int64
skippedDeleteManifests int64
+ skippedDataFiles int64
+ skippedDeleteFiles int64
resultDataFiles int64
resultDeleteFiles int64
@@ -56,6 +61,31 @@ type scanMetricsAccumulator struct {
dvs int64
}
+// countRejections wraps a data-file filter so every file it rejects increments
+// *skipped. Errors are passed through and not counted. Build the wrapper
inside
+// the manifest worker that owns skipped, never in the evaluator caches that
+// workers share.
+func countRejections(filter func(iceberg.DataFile) (bool, error), skipped
*int64) func(iceberg.DataFile) (bool, error) {
+ return func(df iceberg.DataFile) (bool, error) {
+ use, err := filter(df)
+ if err == nil && !use {
+ *skipped++
+ }
+
+ return use, err
+ }
+}
+
+// addSkippedFiles merges one manifest's rejected-entry count into the counter
+// for that manifest's content type, as Java does in ManifestReader.entries.
+func (acc *scanMetricsAccumulator) addSkippedFiles(manifest
iceberg.ManifestFile, skipped int64) {
+ if manifest.ManifestContent() == iceberg.ManifestContentDeletes {
+ acc.skippedDeleteFiles += skipped
+ } else {
+ acc.skippedDataFiles += skipped
+ }
+}
+
func counterCount(v int64) *metrics.CounterResult {
return metrics.NewCounterResult(metrics.UnitCount, v)
}
@@ -107,8 +137,8 @@ func (acc *scanMetricsAccumulator) addResultDeleteMetrics(t
FileScanTask) {
// describes exactly what was used rather than re-resolving it here. projected
may
// be nil when the projection could not be resolved, in which case the report
// simply carries no projected fields. Only the metrics that are actually
measured
-// are populated; unmeasured ones (e.g. skipped-data-files,
indexed-delete-files)
-// are left unset and omitted.
+// are populated; unmeasured ones (e.g. indexed-delete-files) are left unset
and
+// omitted.
func (scan *Scan) buildScanReport(acc *scanMetricsAccumulator, schema,
projected *iceberg.Schema, planning time.Duration) metrics.ScanReport {
ids, names := projectedFields(projected)
envCtx := iceberg.EnvironmentContext()
@@ -154,6 +184,8 @@ func (scan *Scan) buildScanReport(acc
*scanMetricsAccumulator, schema, projected
ScannedDeleteManifests:
counterCount(acc.scannedDeleteManifests),
SkippedDataManifests:
counterCount(acc.skippedDataManifests),
SkippedDeleteManifests:
counterCount(acc.skippedDeleteManifests),
+ SkippedDataFiles:
counterCount(acc.skippedDataFiles),
+ SkippedDeleteFiles:
counterCount(acc.skippedDeleteFiles),
TotalFileSizeInBytes:
counterBytes(acc.totalFileSize),
TotalDeleteFileSizeInBytes:
counterBytes(acc.totalDeleteFileSize),
EqualityDeleteFiles:
counterCount(acc.equalityDeleteFiles),
diff --git a/table/scan_metrics_test.go b/table/scan_metrics_test.go
index ec6abc204..29e1ec2ff 100644
--- a/table/scan_metrics_test.go
+++ b/table/scan_metrics_test.go
@@ -64,6 +64,8 @@ func TestBuildScanReport(t *testing.T) {
dvs: 1,
resultDeleteFiles: 4,
totalDeleteFileSize: 512,
+ skippedDataFiles: 7,
+ skippedDeleteFiles: 3,
}
sr := scan.buildScanReport(acc, meta.CurrentSchema(),
meta.CurrentSchema(), 5*time.Millisecond)
@@ -91,6 +93,10 @@ func TestBuildScanReport(t *testing.T) {
assert.Equal(t, int64(1), m.PositionalDeleteFiles.Value)
assert.Equal(t, int64(2), m.EqualityDeleteFiles.Value)
assert.Equal(t, int64(1), m.DVs.Value)
+ require.NotNil(t, m.SkippedDataFiles)
+ assert.Equal(t, int64(7), m.SkippedDataFiles.Value)
+ require.NotNil(t, m.SkippedDeleteFiles)
+ assert.Equal(t, int64(3), m.SkippedDeleteFiles.Value)
require.NotNil(t, m.TotalFileSizeInBytes)
assert.Equal(t, metrics.UnitBytes, m.TotalFileSizeInBytes.Unit)
@@ -98,7 +104,6 @@ func TestBuildScanReport(t *testing.T) {
assert.Equal(t, int64(512), m.TotalDeleteFileSizeInBytes.Value)
// Unmeasured metrics are left unset (omitted on the wire).
- assert.Nil(t, m.SkippedDataFiles)
assert.Nil(t, m.IndexedDeleteFiles)
}
diff --git a/table/scan_skipped_files_test.go b/table/scan_skipped_files_test.go
new file mode 100644
index 000000000..7cb77977f
--- /dev/null
+++ b/table/scan_skipped_files_test.go
@@ -0,0 +1,245 @@
+// 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"
+ iceio "github.com/apache/iceberg-go/io"
+ "github.com/apache/iceberg-go/metrics"
+ "github.com/stretchr/testify/require"
+)
+
+// skippedFilesTestTable builds a single append snapshot over a table
+// partitioned by identity(category):
+//
+// - data manifest A: categories 0, 1 and 2, four files each, with id bounds
+// [0,9], [10,19], [20,29] and [30,39], plus one DELETED entry in category
0;
+// - data manifest B: two files in category 5, so a category filter that
+// excludes 5 prunes the whole manifest;
+// - a delete manifest with one position delete file per category 0, 1 and 2.
+//
+// A filter of category == 1 AND id >= 25 then rejects 8 live entries in A by
+// partition, 2 more by column bounds, and keeps 2.
+func skippedFilesTestTable(t *testing.T, rep metrics.Reporter) *Table {
+ t.Helper()
+ ctx := context.Background()
+
+ const (
+ location = "mem://default/skipped-files"
+ snapshotID = int64(1)
+ )
+ schema := iceberg.NewSchema(0,
+ iceberg.NestedField{ID: 1, Name: "id", Type:
iceberg.PrimitiveTypes.Int32, Required: true},
+ iceberg.NestedField{ID: 2, Name: "category", Type:
iceberg.PrimitiveTypes.Int32, Required: true},
+ )
+ spec := iceberg.NewPartitionSpec(iceberg.PartitionField{
+ SourceIDs: []int{2}, FieldID: 1000, Name: "category",
Transform: iceberg.IdentityTransform{},
+ })
+
+ fsys, err := iceio.LoadFS(ctx, nil, location)
+ require.NoError(t, err)
+ fs := fsys.(iceio.WriteFileIO)
+
+ bound := func(v int32) []byte {
+ b, err := iceberg.Int32Literal(v).MarshalBinary()
+ require.NoError(t, err)
+
+ return b
+ }
+ dataFile := func(category, lower int32) iceberg.DataFile {
+ builder, err := iceberg.NewDataFileBuilder(spec,
iceberg.EntryContentData,
+ fmt.Sprintf("%s/data/category=%d/id-%d.parquet",
location, category, lower),
+ iceberg.ParquetFile, map[int]any{1000: category}, nil,
nil, 10, 100)
+ require.NoError(t, err)
+
+ return builder.
+ LowerBoundValues(map[int][]byte{1: bound(lower)}).
+ UpperBoundValues(map[int][]byte{1: bound(lower + 9)}).
+ Build()
+ }
+ added := func(df iceberg.DataFile) iceberg.ManifestEntry {
+ return iceberg.NewManifestEntry(iceberg.EntryStatusADDED,
new(snapshotID), nil, nil, df)
+ }
+ writeManifest := func(path string, content iceberg.ManifestContent,
entries []iceberg.ManifestEntry) iceberg.ManifestFile {
+ var buf bytes.Buffer
+ writer, err := iceberg.NewManifestWriter(2, &buf, spec, schema,
snapshotID,
+ iceberg.WithManifestWriterContent(content))
+ require.NoError(t, err)
+ for _, entry := range entries {
+ if entry.Status() == iceberg.EntryStatusDELETED {
+ require.NoError(t, writer.Delete(entry))
+ } else {
+ require.NoError(t, writer.Add(entry))
+ }
+ }
+ require.NoError(t, writer.Close())
+ manifest, err := writer.ToManifestFile(path, int64(buf.Len()),
+ iceberg.WithManifestFileContent(content))
+ require.NoError(t, err)
+ require.NoError(t, fs.WriteFile(ctx, path, buf.Bytes()))
+
+ return manifest
+ }
+
+ var entriesA []iceberg.ManifestEntry
+ for category := range int32(3) {
+ for lower := int32(0); lower < 40; lower += 10 {
+ entriesA = append(entriesA, added(dataFile(category,
lower)))
+ }
+ }
+ // Rejected by the partition filter, but deleted entries are not live,
so
+ // they must not count as skipped.
+ entriesA = append(entriesA, iceberg.NewManifestEntry(
+ iceberg.EntryStatusDELETED, new(snapshotID), new(int64(1)),
new(int64(1)), dataFile(0, 100)))
+ manifestA := writeManifest(location+"/metadata/data-a.avro",
iceberg.ManifestContentData, entriesA)
+ manifestB := writeManifest(location+"/metadata/data-b.avro",
iceberg.ManifestContentData,
+ []iceberg.ManifestEntry{added(dataFile(5, 0)),
added(dataFile(5, 10))})
+
+ var deleteEntries []iceberg.ManifestEntry
+ for category := range int32(3) {
+ builder, err := iceberg.NewDataFileBuilder(spec,
iceberg.EntryContentPosDeletes,
+ fmt.Sprintf("%s/data/category=%d/pos-delete.parquet",
location, category),
+ iceberg.ParquetFile, map[int]any{1000: category}, nil,
nil, 1, 10)
+ require.NoError(t, err)
+ deleteEntries = append(deleteEntries, added(builder.Build()))
+ }
+ deleteManifest := writeManifest(location+"/metadata/deletes.avro",
iceberg.ManifestContentDeletes, deleteEntries)
+
+ listPath := location + "/metadata/snap-1.avro"
+ var listBuf bytes.Buffer
+ seqNum := int64(1)
+ require.NoError(t, iceberg.WriteManifestList(2, &listBuf, snapshotID,
nil, &seqNum, 0,
+ []iceberg.ManifestFile{manifestA, manifestB, deleteManifest}))
+ require.NoError(t, fs.WriteFile(ctx, listPath, listBuf.Bytes()))
+
+ meta, err := NewMetadata(schema, &spec, UnsortedSortOrder, location,
nil)
+ require.NoError(t, err)
+ txn := New(Identifier{"db", "tbl"}, meta, "metadata.json", nil,
nil).NewTransaction()
+ txn.meta.snapshotList = []Snapshot{{
+ SnapshotID: snapshotID,
+ TimestampMs: 1000,
+ ManifestList: listPath,
+ SequenceNumber: seqNum,
+ SchemaID: &schema.ID,
+ Summary: &Summary{Operation: OpAppend},
+ }}
+ txn.meta.snapshotLog = []SnapshotLogEntry{{SnapshotID: snapshotID,
TimestampMs: 1000}}
+ txn.meta.currentSnapshotID = new(snapshotID)
+ built, err := txn.meta.Build()
+ require.NoError(t, err)
+
+ return New(Identifier{"db", "tbl"}, built, "metadata.json",
+ func(context.Context) (iceio.IO, error) { return fs, nil }, nil,
+ WithMetricsReporter(rep))
+}
+
+var skippedFilesTestFilter = iceberg.NewAnd(
+ iceberg.EqualTo(iceberg.Reference("category"), int32(1)),
+ iceberg.GreaterThanEqual(iceberg.Reference("id"), int32(25)),
+)
+
+func singleScanReport(t *testing.T, rep *metrics.InMemoryReporter)
metrics.ScanReport {
+ t.Helper()
+ reports := rep.Reports()
+ require.Len(t, reports, 1)
+ report, ok := reports[0].(metrics.ScanReport)
+ require.True(t, ok)
+
+ return report
+}
+
+func TestPlanFilesReportsSkippedDataAndDeleteFiles(t *testing.T) {
+ rep := &metrics.InMemoryReporter{}
+ tbl := skippedFilesTestTable(t, rep)
+
+ tasks, err :=
tbl.Scan(WithRowFilter(skippedFilesTestFilter)).PlanFiles(context.Background())
+ require.NoError(t, err)
+ require.Len(t, tasks, 2)
+
+ m := singleScanReport(t, rep).Metrics
+ // Manifest B is pruned as a whole, so its two files count only toward
+ // skipped-data-manifests.
+ require.Equal(t, int64(1), m.ScannedDataManifests.Value)
+ require.Equal(t, int64(1), m.SkippedDataManifests.Value)
+ // 8 live entries rejected by partition plus 2 by id bounds. The DELETED
+ // entry is not counted.
+ require.NotNil(t, m.SkippedDataFiles)
+ require.Equal(t, metrics.UnitCount, m.SkippedDataFiles.Unit)
+ require.Equal(t, int64(10), m.SkippedDataFiles.Value)
+ require.Equal(t, int64(2), m.ResultDataFiles.Value)
+ // The position deletes in categories 0 and 2 are rejected by partition.
+ require.NotNil(t, m.SkippedDeleteFiles)
+ require.Equal(t, int64(2), m.SkippedDeleteFiles.Value)
+}
+
+func TestPlanFilesReportsZeroSkippedFilesWithoutFilter(t *testing.T) {
+ rep := &metrics.InMemoryReporter{}
+ tbl := skippedFilesTestTable(t, rep)
+
+ tasks, err := tbl.Scan().PlanFiles(context.Background())
+ require.NoError(t, err)
+ require.Len(t, tasks, 14)
+
+ // Java always reports every counter, so zero is emitted rather than
omitted.
+ m := singleScanReport(t, rep).Metrics
+ require.NotNil(t, m.SkippedDataFiles)
+ require.Equal(t, int64(0), m.SkippedDataFiles.Value)
+ require.NotNil(t, m.SkippedDeleteFiles)
+ require.Equal(t, int64(0), m.SkippedDeleteFiles.Value)
+}
+
+func TestIncrementalAppendScanReportsSkippedDataFiles(t *testing.T) {
+ rep := &metrics.InMemoryReporter{}
+ tbl := skippedFilesTestTable(t, rep)
+
+ tasks, err :=
tbl.NewIncrementalAppendScan(WithRowFilter(skippedFilesTestFilter)).
+ ToSnapshot(1).PlanFiles(context.Background())
+ require.NoError(t, err)
+ require.Len(t, tasks, 2)
+
+ m := singleScanReport(t, rep).Metrics
+ require.Equal(t, int64(10), m.SkippedDataFiles.Value)
+ require.Equal(t, int64(0), m.SkippedDeleteFiles.Value)
+}
+
+func TestIncrementalChangelogScanReportsSkippedDataFiles(t *testing.T) {
+ rep := &metrics.InMemoryReporter{}
+ tbl := incrementalChangelogTestTable(t)
+
+ // Manifest 2 holds the DELETED entry for partition 1 and the ADDED
entry for
+ // partition 2, so it survives manifest pruning. A changelog scan reads
+ // deleted entries as changes, so the partition evaluator rejecting that
+ // entry counts as a skipped file. The other manifests hold only other
+ // partitions and are pruned whole.
+ tasks, err := tbl.NewIncrementalChangelogScan(
+ WithRowFilter(iceberg.EqualTo(iceberg.Reference("id"),
int32(2))),
+ WithReporter(rep),
+ ).PlanFiles(context.Background())
+ require.NoError(t, err)
+ require.Len(t, tasks, 1)
+
+ m := singleScanReport(t, rep).Metrics
+ require.Equal(t, int64(1), m.ScannedDataManifests.Value)
+ require.Equal(t, int64(1), m.SkippedDataFiles.Value)
+ require.Equal(t, int64(0), m.SkippedDeleteFiles.Value)
+}
diff --git a/table/scanner.go b/table/scanner.go
index c8ad9effd..5d9ff9ed5 100644
--- a/table/scanner.go
+++ b/table/scanner.go
@@ -157,6 +157,10 @@ type manifestEntries struct {
equalityDeleteEntries []iceberg.ManifestEntry
dvEntries []iceberg.ManifestEntry
mu sync.Mutex
+
+ // metrics carries only the skipped data and delete file counts from
+ // collection, for callers that merge them into a scan report.
+ metrics scanMetricsAccumulator
}
type classifiedManifestEntries struct {
@@ -1282,6 +1286,7 @@ func (scan *Scan)
collectManifestEntriesWithSchemaOptionsAndMinSequenceNum(
manifestIO := newManifestIOBatch(scan.ioF, concurrencyLimit)
manifestResults := make([]classifiedManifestEntries, len(manifestList))
+ skippedFiles := make([]int64, len(manifestList))
g, gctx := errgroup.WithContext(ctx)
g.SetLimit(concurrencyLimit)
@@ -1310,11 +1315,14 @@ func (scan *Scan)
collectManifestEntriesWithSchemaOptionsAndMinSequenceNum(
// Keep pruning stats until equality and
positional deletes are indexed.
projection =
&iceberg.ManifestEntryProjection{IncludePruningStats: true}
}
+ var skipped int64
manifestEntries, err := openManifestWithReadOptions(
- fs, mf, partEval, metricsEval, projection,
false, discardDeleted, discardExisting)
+ fs, mf, countRejections(partEval, &skipped),
countRejections(metricsEval, &skipped),
+ projection, false, discardDeleted,
discardExisting)
if err != nil {
return err
}
+ skippedFiles[manifestIndex] = skipped
classified, err :=
classifyManifestEntries(manifestEntries)
if err != nil {
return err
@@ -1329,7 +1337,12 @@ func (scan *Scan)
collectManifestEntriesWithSchemaOptionsAndMinSequenceNum(
return nil, err
}
- return flattenClassifiedManifestEntries(manifestResults), nil
+ entries := flattenClassifiedManifestEntries(manifestResults)
+ for i, mf := range manifestList {
+ entries.metrics.addSkippedFiles(mf, skippedFiles[i])
+ }
+
+ return entries, nil
}
func (scan *Scan) dataManifestProjection(retainDataFileStats bool)
(iceberg.ManifestEntryProjection, bool) {
@@ -1383,13 +1396,16 @@ func (scan *Scan) planDataManifestTasks(
eqDeleteIndex *equalityDeleteIndex,
) ([]FileScanTask, error) {
return scan.planDataManifestTasksWithOptions(
- ctx, manifestList, schema, minSeqNum, posDeleteIndex, dvIndex,
eqDeleteIndex, false, false)
+ ctx, manifestList, schema, &scanMetricsAccumulator{},
minSeqNum, posDeleteIndex, dvIndex, eqDeleteIndex, false, false)
}
+// planDataManifestTasksWithOptions merges the data entries its manifest
+// workers reject into acc.skippedDataFiles once all workers finish.
func (scan *Scan) planDataManifestTasksWithOptions(
ctx context.Context,
manifestList []iceberg.ManifestFile,
schema *iceberg.Schema,
+ acc *scanMetricsAccumulator,
minSeqNum int64,
posDeleteIndex *positionalDeleteIndex,
dvIndex map[string]iceberg.ManifestEntry,
@@ -1408,6 +1424,7 @@ func (scan *Scan) planDataManifestTasksWithOptions(
}
taskBatches := make([][]FileScanTask, len(manifestList))
+ skippedFiles := make([]int64, len(manifestList))
taskCapacities := make([]int, len(manifestList))
taskOffsets := make([]int, len(manifestList))
directBuffer := scan.rowFilter == nil ||
scan.rowFilter.Equals(iceberg.AlwaysTrue{})
@@ -1480,7 +1497,10 @@ func (scan *Scan) planDataManifestTasksWithOptions(
stripTaskColumnStats = p.IncludePruningStats &&
!dropColumnStats
projection = &p
}
- err = streamManifest(fs, manifest, partEval,
metricsEval, projection, dropColumnStats, func(entry iceberg.ManifestEntry)
error {
+ var skipped int64
+ countedPartEval := countRejections(partEval, &skipped)
+ countedMetricsEval := countRejections(metricsEval,
&skipped)
+ err = streamManifest(fs, manifest, countedPartEval,
countedMetricsEval, projection, dropColumnStats, func(entry
iceberg.ManifestEntry) error {
dataFile := entry.DataFile()
if dataFile.ContentType() !=
iceberg.EntryContentData {
return fmt.Errorf("%w: data manifest
contains %s file %q",
@@ -1505,6 +1525,7 @@ func (scan *Scan) planDataManifestTasksWithOptions(
}
taskBatches[index] = tasks
+ skippedFiles[index] = skipped
return nil
})
@@ -1513,6 +1534,9 @@ func (scan *Scan) planDataManifestTasksWithOptions(
if err := g.Wait(); err != nil {
return nil, err
}
+ for index, manifest := range manifestList {
+ acc.addSkippedFiles(manifest, skippedFiles[index])
+ }
var results []FileScanTask
if directBuffer {
@@ -1761,6 +1785,7 @@ func (scan *Scan) planFilesLocal(
if err != nil {
return nil, err
}
+ acc.skippedDeleteFiles +=
deleteEntries.metrics.skippedDeleteFiles
}
// Equality-delete pruning is the only data-manifest consumer of column
// metrics after the delete entries have been indexed. Positional
deletes and
@@ -1785,7 +1810,7 @@ func (scan *Scan) planFilesLocal(
// Step 4: Stream data entries into per-manifest task batches, then
flatten
// them in manifest order.
plannedTasks, err := scan.planDataManifestTasksWithOptions(
- ctx, dataManifests, schema, minSeqNum, posDeleteIndex, dvIndex,
eqDeleteIndex,
+ ctx, dataManifests, schema, acc, minSeqNum, posDeleteIndex,
dvIndex, eqDeleteIndex,
projectScanColumns, retainDataFileStats)
if err != nil {
return nil, err
diff --git a/table/scanner_manifest_projection_test.go
b/table/scanner_manifest_projection_test.go
index dc90b1ae1..1a65a6202 100644
--- a/table/scanner_manifest_projection_test.go
+++ b/table/scanner_manifest_projection_test.go
@@ -268,6 +268,7 @@ func
TestPlanDataManifestTasksWithProjectionRetainsEqualityDeletePruning(t *test
t.Context(),
[]iceberg.ManifestFile{manifest},
schema,
+ &scanMetricsAccumulator{},
0,
posDeleteIndex,
dvIndex,
@@ -314,6 +315,7 @@ func
TestPlanDataManifestTasksWithProjectionRetainsEqualityDeletePruning(t *test
t.Context(),
[]iceberg.ManifestFile{manifest},
schema,
+ &scanMetricsAccumulator{},
0,
posDeleteIndex,
dvIndex,
