diff --git a/manifest.go b/manifest.go index 3e0c158ed..1130d0b14 100644 --- a/manifest.go +++ b/manifest.go @@ -416,28 +416,7 @@ func (m *manifestFile) HasAddedFiles() bool { return m.AddedFilesCount != 0 } func (m *manifestFile) HasExistingFiles() bool { return m.ExistingFilesCount != 0 } func (m *manifestFile) Entries(fs iceio.IO, discardDeleted bool) iter.Seq2[ManifestEntry, error] { - return func(yield func(ManifestEntry, error) bool) { - f, err := fs.Open(m.FilePath()) - if err != nil { - yield(nil, err) - - return - } - aborted := false - defer func() { - if cerr := f.Close(); cerr != nil && !aborted { - yield(nil, cerr) - } - }() - - for entry, err := range iterManifest(m, f, discardDeleted) { - if !yield(entry, err) { - aborted = true - - return - } - } - } + return manifestEntries(fs, m, discardDeleted, nil) } func (m *manifestFile) FetchEntries(fs iceio.IO, discardDeleted bool) (_ []ManifestEntry, err error) { @@ -723,13 +702,48 @@ type ManifestReader struct { // file. If the caller is interested in the manifest entries in the file, it must call // [ManifestReader.Entries] before closing the provided reader. func NewManifestReader(file ManifestFile, in io.Reader) (*ManifestReader, error) { - rd, err := ocf.NewReader(in) + return newManifestReader(file, in, nil) +} + +// NewManifestReaderWithProjection returns a manifest reader that decodes the +// standard scan fields and optionally the column statistics selected by +// projection. Manifest metadata validation and entry inheritance are the same +// as in NewManifestReader; fields omitted by the projection retain their zero +// values in the returned DataFile. +func NewManifestReaderWithProjection( + file ManifestFile, + in io.Reader, + projection ManifestEntryProjection, +) (*ManifestReader, error) { + return newManifestReader(file, in, &projection) +} + +func newManifestReader( + file ManifestFile, + in io.Reader, + projection *ManifestEntryProjection, +) (*ManifestReader, error) { + var writerSchema *avro.Schema + var rd *ocf.Reader + var err error + if projection == nil { + rd, err = ocf.NewReader(in) + } else { + rd, err = ocf.NewReader(in, ocf.WithReaderSchemaFunc(func(reader *ocf.Reader) (*avro.Schema, error) { + writerSchema = reader.Schema() + + return projectedManifestEntrySchema(writerSchema, *projection) + })) + } if err != nil { return nil, err } metadata := rd.Metadata() - sc := rd.Schema() + if writerSchema == nil { + writerSchema = rd.Schema() + } + sc := writerSchema formatVersion := 1 // format-version is optional for v1 manifest files, so default to v1. @@ -972,9 +986,20 @@ func (c *ManifestReader) ReadEntry() (ManifestEntry, error) { // iterManifest returns an iterator that streams manifest entries from // the provided reader without buffering them. If discardDeleted is true, // entries whose status is "deleted" are skipped. -func iterManifest(m ManifestFile, f io.Reader, discardDeleted bool) iter.Seq2[ManifestEntry, error] { +func iterManifest( + m ManifestFile, + f io.Reader, + discardDeleted bool, + projection *ManifestEntryProjection, +) iter.Seq2[ManifestEntry, error] { return func(yield func(ManifestEntry, error) bool) { - manifestReader, err := NewManifestReader(m, f) + var manifestReader *ManifestReader + var err error + if projection == nil { + manifestReader, err = NewManifestReader(m, f) + } else { + manifestReader, err = NewManifestReaderWithProjection(m, f, *projection) + } if err != nil { yield(nil, err) @@ -1016,7 +1041,7 @@ func iterManifest(m ManifestFile, f io.Reader, discardDeleted bool) iter.Seq2[Ma // is true, the returned slice omits entries whose status is "deleted". func ReadManifest(m ManifestFile, f io.Reader, discardDeleted bool) ([]ManifestEntry, error) { var results []ManifestEntry - for entry, err := range iterManifest(m, f, discardDeleted) { + for entry, err := range iterManifest(m, f, discardDeleted, nil) { if err != nil { return results, err } diff --git a/manifest_projection.go b/manifest_projection.go new file mode 100644 index 000000000..263e9e14e --- /dev/null +++ b/manifest_projection.go @@ -0,0 +1,206 @@ +// 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 iceberg + +import ( + "errors" + "fmt" + "iter" + "slices" + + iceio "github.com/apache/iceberg-go/io" + lru "github.com/hashicorp/golang-lru/v2" + "github.com/twmb/avro" +) + +// ManifestEntryProjection selects the optional data-file fields decoded while +// reading a manifest. The fields needed to build a scan task are always read. +// Column statistics are read only when IncludeColumnStats is true. +// +// A projected read is intended for planning paths that use statistics +// transiently. Callers that need the complete DataFile metadata should use +// ManifestFile.Entries or ReadManifest instead. +type ManifestEntryProjection struct { + IncludeColumnStats bool +} + +const manifestEntryProjectionCacheSize = 256 + +type manifestEntryProjectionCacheKey struct { + writerSchema string + includeColumnStats bool +} + +var manifestEntryProjectionCache = func() *lru.Cache[manifestEntryProjectionCacheKey, *avro.Schema] { + c, err := lru.New[manifestEntryProjectionCacheKey, *avro.Schema](manifestEntryProjectionCacheSize) + if err != nil { + panic(err) + } + + return c +}() + +// EntriesWithProjection streams manifest entries using a reader-schema +// projection. It is the projected counterpart to ManifestFile.Entries and is +// useful when a caller needs only the fields required for scan planning. +func EntriesWithProjection( + fs iceio.IO, + m ManifestFile, + discardDeleted bool, + projection ManifestEntryProjection, +) iter.Seq2[ManifestEntry, error] { + return manifestEntries(fs, m, discardDeleted, &projection) +} + +func manifestEntries( + fs iceio.IO, + m ManifestFile, + discardDeleted bool, + projection *ManifestEntryProjection, +) iter.Seq2[ManifestEntry, error] { + return func(yield func(ManifestEntry, error) bool) { + f, err := fs.Open(m.FilePath()) + if err != nil { + yield(nil, err) + + return + } + aborted := false + defer func() { + if cerr := f.Close(); cerr != nil && !aborted { + yield(nil, cerr) + } + }() + + for entry, err := range iterManifest(m, f, discardDeleted, projection) { + if !yield(entry, err) { + aborted = true + + return + } + } + } +} + +func projectedManifestEntrySchema( + writerSchema *avro.Schema, + projection ManifestEntryProjection, +) (*avro.Schema, error) { + key := manifestEntryProjectionCacheKey{ + writerSchema: writerSchema.String(), + includeColumnStats: projection.IncludeColumnStats, + } + if cached, ok := manifestEntryProjectionCache.Get(key); ok { + return cached, nil + } + + root := writerSchema.Root() + projectedRoot := *root + projectedRoot.Fields = slices.Clone(root.Fields) + dataFileFound := false + for i := range projectedRoot.Fields { + if projectedRoot.Fields[i].Name != "data_file" { + continue + } + + dataFileFound = true + dataFile := projectedRoot.Fields[i].Type + if dataFile.Type != "record" { + return nil, fmt.Errorf("manifest entry data_file has unexpected Avro type %q", dataFile.Type) + } + + fields := make([]avro.SchemaField, 0, len(dataFile.Fields)) + for _, field := range dataFile.Fields { + if manifestScanDataFileField(field.Name, projection.IncludeColumnStats) { + fields = append(fields, field) + } + } + dataFile.Fields = fields + projectedRoot.Fields[i].Type = dataFile + + break + } + if !dataFileFound { + return nil, errors.New("manifest entry schema does not contain a data_file field") + } + + projected, err := projectedRoot.Schema() + if err != nil { + return nil, fmt.Errorf("build projected manifest entry schema: %w", err) + } + manifestEntryProjectionCache.Add(key, projected) + + return projected, nil +} + +func manifestScanDataFileField(name string, includeColumnStats bool) bool { + switch name { + case "content", "file_path", "file_format", "partition", "record_count", + "file_size_in_bytes", "key_metadata", "split_offsets", "equality_ids", + "sort_order_id", "first_row_id", "referenced_data_file", "content_offset", + "content_size_in_bytes": + return true + case "value_counts", "null_value_counts", "nan_value_counts", "lower_bounds", "upper_bounds": + return includeColumnStats + default: + // block_size_in_bytes, column_sizes, and distinct_counts are not + // needed to build or read a FileScanTask. + return false + } +} + +// DataFileWithoutColumnStats returns a copy of the built-in DataFile with +// transient column statistics removed. Other DataFile implementations are +// returned unchanged because the package cannot safely clone their private +// state. +func DataFileWithoutColumnStats(file DataFile) DataFile { + d, ok := file.(*dataFile) + if !ok { + return file + } + + out := cloneDataFileAvroFields(d) + out.ColSizes = nil + out.ValCounts = nil + out.NullCounts = nil + out.NaNCounts = nil + out.DistinctCounts = nil + out.LowerBounds = nil + out.UpperBounds = nil + out.fieldNameToID = d.fieldNameToID + out.fieldIDToLogicalType = d.fieldIDToLogicalType + out.fieldIDToPartitionData = d.fieldIDToPartitionData + out.fieldIDToDecimalScale = d.fieldIDToDecimalScale + out.specID = d.specID + + return out +} + +// ManifestEntryWithoutColumnStats returns a copy of an entry whose built-in +// DataFile has had transient column statistics removed. +func ManifestEntryWithoutColumnStats(entry ManifestEntry) ManifestEntry { + m, ok := entry.(*manifestEntry) + if !ok { + return entry + } + + out := *m + out.Data = DataFileWithoutColumnStats(m.Data) + + return &out +} diff --git a/manifest_projection_bench_test.go b/manifest_projection_bench_test.go new file mode 100644 index 000000000..01e59fa12 --- /dev/null +++ b/manifest_projection_bench_test.go @@ -0,0 +1,164 @@ +// 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 iceberg + +import ( + "bytes" + "fmt" + "iter" + "testing" + + iceio "github.com/apache/iceberg-go/io" +) + +var manifestProjectionBenchmarkSink int + +func BenchmarkManifestEntryProjection(b *testing.B) { + for _, fieldCount := range []int{10, 100, 1_000} { + b.Run(fmt.Sprintf("stats_fields=%d", fieldCount), func(b *testing.B) { + fixture := newManifestProjectionBenchmarkFixture(b, fieldCount, 10_000) + + b.Run("full", func(b *testing.B) { + benchmarkManifestEntryRead(b, fixture, nil) + }) + b.Run("scan", func(b *testing.B) { + benchmarkManifestEntryRead(b, fixture, &ManifestEntryProjection{}) + }) + b.Run("scan_with_stats", func(b *testing.B) { + benchmarkManifestEntryRead(b, fixture, &ManifestEntryProjection{IncludeColumnStats: true}) + }) + }) + } +} + +type manifestProjectionBenchmarkFixture struct { + fs *iceio.MemFS + manifest ManifestFile + count int +} + +func newManifestProjectionBenchmarkFixture( + b *testing.B, + fieldCount, entryCount int, +) manifestProjectionBenchmarkFixture { + b.Helper() + + fields := make([]NestedField, fieldCount) + valueCounts := make(map[int]int64, fieldCount) + nullCounts := make(map[int]int64, fieldCount) + nanCounts := make(map[int]int64, fieldCount) + lowerBounds := make(map[int][]byte, fieldCount) + upperBounds := make(map[int][]byte, fieldCount) + columnSizes := make(map[int]int64, fieldCount) + distinctCounts := make(map[int]int64, fieldCount) + for i := range fieldCount { + id := i + 1 + fields[i] = NestedField{ + ID: id, Name: fmt.Sprintf("field_%d", id), Type: PrimitiveTypes.Int64, Required: true, + } + valueCounts[id] = int64(entryCount) + nullCounts[id] = 0 + nanCounts[id] = 0 + lowerBounds[id] = []byte{0, 0, 0, 0, 0, 0, 0, 0} + upperBounds[id] = []byte{0, 0, 0, 0, 0, 0, 0, 1} + columnSizes[id] = 64 + distinctCounts[id] = 1 + } + schema := NewSchema(1, fields...) + builder, err := NewDataFileBuilder( + *UnpartitionedSpec, + EntryContentData, + "data.parquet", + ParquetFile, + nil, + nil, + nil, + int64(entryCount), + 128, + ) + if err != nil { + b.Fatal(err) + } + builder. + ColumnSizes(columnSizes). + ValueCounts(valueCounts). + NullValueCounts(nullCounts). + NaNValueCounts(nanCounts). + DistinctValueCounts(distinctCounts). + LowerBoundValues(lowerBounds). + UpperBoundValues(upperBounds) + + snapshotID := int64(1) + entries := make([]ManifestEntry, entryCount) + for i := range entries { + entries[i] = NewManifestEntry(EntryStatusADDED, &snapshotID, nil, nil, builder.Build()) + } + + manifestPath := fmt.Sprintf("mem://manifest-%d.avro", fieldCount) + var manifestBytes bytes.Buffer + manifest, err := WriteManifest( + manifestPath, + &manifestBytes, + 2, + *UnpartitionedSpec, + schema, + snapshotID, + entries, + ) + if err != nil { + b.Fatal(err) + } + + fs := iceio.NewMemFS() + if err := fs.WriteFile(manifestPath, manifestBytes.Bytes()); err != nil { + b.Fatal(err) + } + + return manifestProjectionBenchmarkFixture{fs: fs, manifest: manifest, count: entryCount} +} + +func benchmarkManifestEntryRead( + b *testing.B, + fixture manifestProjectionBenchmarkFixture, + projection *ManifestEntryProjection, +) { + b.Helper() + b.ReportAllocs() + b.ReportMetric(float64(fixture.count), "entries/op") + b.ResetTimer() + + for b.Loop() { + count := 0 + var entries iter.Seq2[ManifestEntry, error] + if projection == nil { + entries = fixture.manifest.Entries(fixture.fs, true) + } else { + entries = EntriesWithProjection(fixture.fs, fixture.manifest, true, *projection) + } + for entry, err := range entries { + if err != nil { + b.Fatal(err) + } + count++ + manifestProjectionBenchmarkSink += len(entry.DataFile().FilePath()) + } + if count != fixture.count { + b.Fatalf("read %d entries, want %d", count, fixture.count) + } + } +} diff --git a/manifest_projection_test.go b/manifest_projection_test.go new file mode 100644 index 000000000..c6492afc6 --- /dev/null +++ b/manifest_projection_test.go @@ -0,0 +1,143 @@ +// 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 iceberg + +import ( + "bytes" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestManifestEntryProjectionDropsTransientStats(t *testing.T) { + schema := NewSchema(1, NestedField{ + ID: 1, Name: "id", Type: PrimitiveTypes.Int64, Required: true, + }) + builder, err := NewDataFileBuilder( + *UnpartitionedSpec, + EntryContentData, + "data.parquet", + ParquetFile, + nil, + nil, + nil, + 3, + 128, + ) + require.NoError(t, err) + builder. + ColumnSizes(map[int]int64{1: 10}). + ValueCounts(map[int]int64{1: 3}). + NullValueCounts(map[int]int64{1: 0}). + NaNValueCounts(map[int]int64{1: 0}). + LowerBoundValues(map[int][]byte{1: {0x01}}). + UpperBoundValues(map[int][]byte{1: {0x03}}). + KeyMetadata([]byte{0x04}). + SplitOffsets([]int64{4, 64}). + SortOrderID(7). + FirstRowID(10). + ReferencedDataFile("data.parquet"). + ContentOffset(12). + ContentSizeInBytes(20) + + snapshotID := int64(5) + sequenceNumber := int64(6) + fileSequenceNumber := int64(7) + entry := NewManifestEntry( + EntryStatusADDED, + &snapshotID, + &sequenceNumber, + &fileSequenceNumber, + builder.Build(), + ) + var manifestBytes bytes.Buffer + manifest, err := WriteManifest( + "manifest.avro", &manifestBytes, 3, *UnpartitionedSpec, schema, snapshotID, + []ManifestEntry{entry}, + ) + require.NoError(t, err) + + reader, err := NewManifestReaderWithProjection( + manifest, bytes.NewReader(manifestBytes.Bytes()), ManifestEntryProjection{}, + ) + require.NoError(t, err) + projected, err := reader.ReadEntry() + require.NoError(t, err) + require.NoError(t, reader.Close()) + + assert.Equal(t, entry.Status(), projected.Status()) + assert.Equal(t, entry.SnapshotID(), projected.SnapshotID()) + assert.Equal(t, entry.SequenceNum(), projected.SequenceNum()) + assert.Equal(t, entry.FileSequenceNum(), projected.FileSequenceNum()) + assert.Equal(t, "data.parquet", projected.DataFile().FilePath()) + assert.Equal(t, int64(3), projected.DataFile().Count()) + assert.Equal(t, int64(128), projected.DataFile().FileSizeBytes()) + assert.Equal(t, []int64{4, 64}, projected.DataFile().SplitOffsets()) + assert.Equal(t, 7, *projected.DataFile().SortOrderID()) + assert.Equal(t, int64(10), *projected.DataFile().FirstRowID()) + assert.Equal(t, "data.parquet", *projected.DataFile().ReferencedDataFile()) + assert.Equal(t, int64(12), *projected.DataFile().ContentOffset()) + assert.Equal(t, int64(20), *projected.DataFile().ContentSizeInBytes()) + assert.Equal(t, []byte{0x04}, projected.DataFile().KeyMetadata()) + assert.Empty(t, projected.DataFile().ColumnSizes()) + assert.Empty(t, projected.DataFile().ValueCounts()) + assert.Empty(t, projected.DataFile().NullValueCounts()) + assert.Empty(t, projected.DataFile().NaNValueCounts()) + assert.Empty(t, projected.DataFile().LowerBoundValues()) + assert.Empty(t, projected.DataFile().UpperBoundValues()) + + statsReader, err := NewManifestReaderWithProjection( + manifest, bytes.NewReader(manifestBytes.Bytes()), ManifestEntryProjection{IncludeColumnStats: true}, + ) + require.NoError(t, err) + withStats, err := statsReader.ReadEntry() + require.NoError(t, err) + require.NoError(t, statsReader.Close()) + assert.Equal(t, map[int]int64{1: 3}, withStats.DataFile().ValueCounts()) + assert.Equal(t, map[int][]byte{1: {0x01}}, withStats.DataFile().LowerBoundValues()) + assert.Equal(t, map[int][]byte{1: {0x03}}, withStats.DataFile().UpperBoundValues()) + assert.Empty(t, withStats.DataFile().ColumnSizes()) +} + +func TestManifestEntryWithoutColumnStatsPreservesEntry(t *testing.T) { + builder, err := NewDataFileBuilder( + *UnpartitionedSpec, + EntryContentData, + "data.parquet", + ParquetFile, + nil, + nil, + nil, + 1, + 10, + ) + require.NoError(t, err) + builder.ValueCounts(map[int]int64{1: 1}) + + snapshotID := int64(3) + entry := NewManifestEntry(EntryStatusEXISTING, &snapshotID, nil, nil, builder.Build()) + projected := ManifestEntryWithoutColumnStats(entry) + + assert.NotSame(t, entry, projected) + assert.Equal(t, entry.Status(), projected.Status()) + assert.Equal(t, entry.SnapshotID(), projected.SnapshotID()) + assert.Equal(t, entry.DataFile().FilePath(), projected.DataFile().FilePath()) + assert.Empty(t, projected.DataFile().ValueCounts()) + assert.Equal(t, map[int]int64{1: 1}, entry.DataFile().ValueCounts()) +} diff --git a/table/scanner.go b/table/scanner.go index f5599d932..2b5c2895c 100644 --- a/table/scanner.go +++ b/table/scanner.go @@ -337,10 +337,24 @@ func GetPartitionRecord(dataFile iceberg.DataFile, partitionType *iceberg.Struct func openManifest(io io.IO, manifest iceberg.ManifestFile, partitionFilter, metricsEval func(iceberg.DataFile) (bool, error), +) ([]iceberg.ManifestEntry, error) { + return openManifestWithProjection(io, manifest, partitionFilter, metricsEval, nil, false) +} + +func openManifestWithProjection( + io io.IO, + manifest iceberg.ManifestFile, + partitionFilter, metricsEval func(iceberg.DataFile) (bool, error), + projection *iceberg.ManifestEntryProjection, + dropColumnStats bool, ) ([]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) { + entries := manifest.Entries(io, true) + if projection != nil { + entries = iceberg.EntriesWithProjection(io, manifest, true, *projection) + } + for entry, err := range entries { if err != nil { return nil, err } @@ -359,6 +373,9 @@ func openManifest(io io.IO, manifest iceberg.ManifestFile, } if m { + if dropColumnStats { + entry = iceberg.ManifestEntryWithoutColumnStats(entry) + } out = append(out, entry) } } @@ -953,6 +970,15 @@ func (scan *Scan) collectManifestEntriesWithSchema( ctx context.Context, manifestList []iceberg.ManifestFile, schema *iceberg.Schema, +) (*manifestEntries, error) { + return scan.collectManifestEntriesWithSchemaOptions(ctx, manifestList, schema, false) +} + +func (scan *Scan) collectManifestEntriesWithSchemaOptions( + ctx context.Context, + manifestList []iceberg.ManifestFile, + schema *iceberg.Schema, + projectScanColumns bool, ) (*manifestEntries, error) { metricsEval, err := newInclusiveMetricsEvaluator( schema, @@ -990,7 +1016,18 @@ func (scan *Scan) collectManifestEntriesWithSchema( if err != nil { return fmt.Errorf("failed to build partition evaluator for spec %d: %w", mf.PartitionSpecID(), err) } - manifestEntries, err := openManifest(fs, mf, partEval, metricsEval) + var projection *iceberg.ManifestEntryProjection + dropColumnStats := false + if projectScanColumns { + p := iceberg.ManifestEntryProjection{ + IncludeColumnStats: scan.manifestProjectionNeedsStats(mf), + } + projection = &p + dropColumnStats = p.IncludeColumnStats && + mf.ManifestContent() == iceberg.ManifestContentData + } + manifestEntries, err := openManifestWithProjection( + fs, mf, partEval, metricsEval, projection, dropColumnStats) if err != nil { return err } @@ -1009,12 +1046,25 @@ func (scan *Scan) collectManifestEntriesWithSchema( return entries, nil } +func (scan *Scan) manifestProjectionNeedsStats(manifest iceberg.ManifestFile) bool { + return manifest.ManifestContent() == iceberg.ManifestContentDeletes || + (scan.rowFilter != nil && !scan.rowFilter.Equals(iceberg.AlwaysTrue{})) +} + // 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 // scan's reporter on success; remote (server-side) planning reports its own // metrics and does not emit here. func (scan *Scan) PlanFiles(ctx context.Context) ([]FileScanTask, error) { + return scan.planFiles(ctx, false) +} + +// planFiles performs scan planning. Projected local planning is used by +// ToArrowRecords because manifest column statistics are only needed while +// pruning; PlanFiles keeps the complete DataFile metadata for callers that +// inspect planned tasks. +func (scan *Scan) planFiles(ctx context.Context, projectScanColumns bool) ([]FileScanTask, error) { if atomic.LoadUint32(&scan.closed) != 0 { return nil, fmt.Errorf("%w: scan is closed", ErrInvalidOperation) } @@ -1055,7 +1105,7 @@ func (scan *Scan) PlanFiles(ctx context.Context) ([]FileScanTask, error) { return nil, err } - results, err := scan.planFilesLocal(ctx, &acc, schema) + results, err := scan.planFilesLocal(ctx, &acc, schema, projectScanColumns) if err != nil { return nil, err } @@ -1091,7 +1141,12 @@ func (scan *Scan) PlanFiles(ctx context.Context) ([]FileScanTask, error) { // local plan retires any previous remote plan; a failed local plan leaves it // usable. It returns a nil slice (not an empty one) when there is no snapshot // or every manifest is pruned. -func (scan *Scan) planFilesLocal(ctx context.Context, acc *scanMetricsAccumulator, schema *iceberg.Schema) (results []FileScanTask, err error) { +func (scan *Scan) planFilesLocal( + ctx context.Context, + acc *scanMetricsAccumulator, + schema *iceberg.Schema, + projectScanColumns bool, +) (results []FileScanTask, err error) { defer func() { if err == nil { err = scan.closePlanIO() @@ -1112,7 +1167,8 @@ 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) + entries, err := scan.collectManifestEntriesWithSchemaOptions( + ctx, manifestList, schema, projectScanColumns) if err != nil { return nil, err } @@ -1152,6 +1208,11 @@ func (scan *Scan) planFilesLocal(ctx context.Context, acc *scanMetricsAccumulato if err != nil { return nil, err } + if projectScanColumns { + deleteFiles = dataFilesWithoutColumnStats(deleteFiles) + eqDeleteFiles = dataFilesWithoutColumnStats(eqDeleteFiles) + dvFiles = dataFilesWithoutColumnStats(dvFiles) + } task := FileScanTask{ File: e.DataFile(), @@ -1181,10 +1242,32 @@ func (scan *Scan) planFilesLocal(ctx context.Context, acc *scanMetricsAccumulato // result-scoped, consistent with result-data-files and total-file-size (a // DV-suppressed positional delete never lands on a task, so it is excluded). acc.applyResultDeleteMetrics(results) + if projectScanColumns { + // The indexes retain the full delete entries while matching them to + // data files. Release those references before returning so only the + // compact task metadata remains live after planning. + entries = nil + posDeleteIndex = nil + dvIndex = nil + eqDeleteIndex = nil + } return results, nil } +func dataFilesWithoutColumnStats(files []iceberg.DataFile) []iceberg.DataFile { + if len(files) == 0 { + return files + } + + projected := make([]iceberg.DataFile, len(files)) + for i, file := range files { + projected[i] = iceberg.DataFileWithoutColumnStats(file) + } + + return projected +} + // canLimitLocalPlanning reports whether the manifest-list row counts are // sufficient to safely narrow local planning for this scan. A row filter can // remove rows from a data file, and delete manifests can remove rows at read @@ -1456,7 +1539,7 @@ type FileScanTask struct { // The purpose for returning the schema up front is to handle the case where there are no // rows returned. The resulting Arrow Schema of the projection will still be known. func (scan *Scan) ToArrowRecords(ctx context.Context) (*arrow.Schema, iter.Seq2[arrow.RecordBatch, error], error) { - tasks, err := scan.PlanFiles(ctx) + tasks, err := scan.planFiles(ctx, true) if err != nil { return nil, nil, err } diff --git a/table/scanner_manifest_projection_test.go b/table/scanner_manifest_projection_test.go new file mode 100644 index 000000000..277e4ad65 --- /dev/null +++ b/table/scanner_manifest_projection_test.go @@ -0,0 +1,94 @@ +// 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" + "testing" + + "github.com/apache/iceberg-go" + iceio "github.com/apache/iceberg-go/io" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestOpenManifestWithProjectionDropsStatsAfterFiltering(t *testing.T) { + spec := partitionedSpec() + schema := simpleSchema() + snapshotID := int64(1) + builder, err := iceberg.NewDataFileBuilder( + spec, + iceberg.EntryContentData, + "mem://default/table/data/file.parquet", + iceberg.ParquetFile, + map[int]any{1000: int32(7)}, + nil, + nil, + 1, + 100, + ) + require.NoError(t, err) + builder. + ValueCounts(map[int]int64{1: 1}). + NullValueCounts(map[int]int64{1: 0}). + NaNValueCounts(map[int]int64{1: 0}). + LowerBoundValues(map[int][]byte{1: {0x01}}). + UpperBoundValues(map[int][]byte{1: {0x01}}) + + manifestPath := "mem://default/table/metadata/manifest.avro" + var manifestBytes bytes.Buffer + manifest, err := iceberg.WriteManifest( + manifestPath, + &manifestBytes, + 2, + spec, + schema, + snapshotID, + []iceberg.ManifestEntry{iceberg.NewManifestEntry( + iceberg.EntryStatusADDED, &snapshotID, nil, nil, builder.Build(), + )}, + ) + require.NoError(t, err) + + fs := iceio.NewMemFS() + require.NoError(t, fs.WriteFile(manifestPath, manifestBytes.Bytes())) + + projection := iceberg.ManifestEntryProjection{IncludeColumnStats: true} + entries, err := openManifestWithProjection( + fs, + manifest, + func(file iceberg.DataFile) (bool, error) { + assert.Equal(t, int32(7), file.Partition()[1000]) + + return true, nil + }, + func(file iceberg.DataFile) (bool, error) { + assert.Equal(t, map[int]int64{1: 1}, file.ValueCounts()) + assert.Equal(t, map[int][]byte{1: {0x01}}, file.LowerBoundValues()) + + return true, nil + }, + &projection, + true, + ) + require.NoError(t, err) + require.Len(t, entries, 1) + assert.Equal(t, "mem://default/table/data/file.parquet", entries[0].DataFile().FilePath()) + assert.Empty(t, entries[0].DataFile().ValueCounts()) + assert.Empty(t, entries[0].DataFile().LowerBoundValues()) +}