laskoviymishka commented on code in PR #1939:
URL: https://github.com/apache/iceberg-go/pull/1939#discussion_r3895079997
##########
table/internal/parquet_files.go:
##########
@@ -1844,8 +1898,18 @@ func (w wrapPqArrowReader) GetRecords(ctx
context.Context, cols []int, tester an
}
}
+ rangeSet := rowGroupTester != nil &&
Review Comment:
The `RangeSet` field you added is exactly the right disambiguation, but this
line then undercuts it: `RangeSet || Start != 0 || Length != 0` means any
caller that sets `Length` without `RangeSet` silently turns on range filtering
— the opposite of what the field's own doc comment promises ("the zero value
keeps the historical full-file behavior"). `processRecordsWithPlans` only ever
sets the fields together, so it's masked today, but `ParquetRowGroupTester` is
exported, so this is a live footgun for the next caller.
I'd gate purely on the explicit flag:
```go
rangeSet := rowGroupTester != nil && rowGroupTester.RangeSet
```
wdyt?
##########
table/internal/parquet_files.go:
##########
@@ -1833,6 +1841,52 @@ func (w wrapPqArrowReader) PrunedSchema(projectedIDs
map[int]struct{}, mapping i
return pruneParquetColumns(w.Manifest, projectedIDs, false, mapping)
}
+// parquetRowGroupSplitOffset returns the same first-page offset used when
+// DataFileStatistics.SplitOffsets is built. RowGroup.file_offset is optional
+// in Parquet and Arrow returns zero when it is absent, so relying on it can
+// discard every row group in a split task from an otherwise valid file.
+func parquetRowGroupSplitOffset(rgMeta *metadata.RowGroupMetaData) int64 {
Review Comment:
This is probably the silent-row-loss case zeroshade already flagged, but let
me pin the exact mechanism in case it helps: task boundaries come from
`file.SplitOffsets()` (whatever the writer stored), while row-group membership
here is re-derived from column 0's first page. For iceberg-go-written files
those agree. For a file written by Java or iceberg-rust, the stored offsets
come from `RowGroup.file_offset` / `getStartingPos()`, which can differ from
column 0's first page when columns aren't laid out in field order. When they
differ, a row group can fall outside every task's `[start, start+length)` and
vanish from the scan with no error.
The durable fix is to stop re-deriving: carry the per-row-group offset that
planning actually used (the stored `SplitOffsets` entry) into the task and
filter against that same value in `GetRecords`, rather than recomputing from
column 0 here. wdyt?
##########
table/scanner.go:
##########
@@ -1133,6 +1133,8 @@ func (scan *Scan) planFilesLocal(ctx context.Context, acc
*scanMetricsAccumulato
}
results = make([]FileScanTask, 0, len(entries.dataEntries))
+ splitTargetSize := scan.metadata.Properties().GetInt64(
Review Comment:
Splitting only happens here in `planFilesLocal`, so REST-catalog scans
(which go through `planFilesRemote` and hand back server-provided tasks) never
split, even with `read.split.target-size` set. Reasonable scope for a first
cut, but the property reads as global table config, so I'd call the local-only
limitation out in the godoc and in `configuration.md` — otherwise someone
benchmarks it on a local catalog and then wonders where the speedup went on
REST. wdyt?
##########
table/scan_splits_test.go:
##########
@@ -0,0 +1,466 @@
+// 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 (
+ "context"
+ "fmt"
+ "path/filepath"
+ "strings"
+ "testing"
+
+ "github.com/apache/arrow-go/v18/arrow"
+ "github.com/apache/arrow-go/v18/arrow/array"
+ "github.com/apache/arrow-go/v18/arrow/memory"
+ "github.com/apache/arrow-go/v18/parquet"
+ "github.com/apache/arrow-go/v18/parquet/pqarrow"
+ "github.com/apache/iceberg-go"
+ iceio "github.com/apache/iceberg-go/io"
+ "github.com/apache/iceberg-go/metrics"
+ "github.com/apache/iceberg-go/table/dv"
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+)
+
+func writeSplitParquetFile(t testing.TB, path string, sc *arrow.Schema,
jsonData string) {
+ t.Helper()
+
+ rec, _, err := array.RecordFromJSON(memory.DefaultAllocator, sc,
strings.NewReader(jsonData))
+ require.NoError(t, err)
+ defer rec.Release()
+
+ fs := iceio.LocalFS{}
+ fw, err := fs.Create(path)
+ require.NoError(t, err)
+ defer fw.Close()
+
+ data := array.NewTableFromRecords(sc, []arrow.RecordBatch{rec})
+ defer data.Release()
+
+ props := parquet.NewWriterProperties(parquet.WithStats(true))
+ require.NoError(t, pqarrow.WriteTable(data, fw, rec.NumRows(), props,
pqarrow.DefaultWriterProps()))
+}
+
+func splitTestDataFile(t *testing.T, format iceberg.FileFormat, size int64,
offsets []int64) iceberg.DataFile {
+ t.Helper()
+
+ builder, err := iceberg.NewDataFileBuilder(
+ *iceberg.UnpartitionedSpec,
+ iceberg.EntryContentData,
+ "mem://table/data.parquet",
+ format,
+ nil,
+ nil,
+ nil,
+ 100,
+ size,
+ )
+ require.NoError(t, err)
+ if offsets != nil {
+ builder.SplitOffsets(offsets)
+ }
+
+ return builder.Build()
+}
+
+func dataFileWithSplitOffsets(t *testing.T, source iceberg.DataFile, offsets
[]int64) iceberg.DataFile {
+ t.Helper()
+
+ builder, err := iceberg.NewDataFileBuilder(
+ *iceberg.UnpartitionedSpec,
+ source.ContentType(),
+ source.FilePath(),
+ source.FileFormat(),
+ nil,
+ nil,
+ nil,
+ source.Count(),
+ source.FileSizeBytes(),
+ )
+ require.NoError(t, err)
+
+ return builder.SplitOffsets(offsets).Build()
+}
+
+func TestSplitParquetScanTask(t *testing.T) {
+ firstRowID := int64(10)
+ file := splitTestDataFile(t, iceberg.ParquetFile, 100, []int64{8, 40,
70})
+ task := FileScanTask{
+ File: file,
+ DeleteFiles: []iceberg.DataFile{file},
+ Start: 0,
+ Length: 100,
+ Residual: iceberg.AlwaysTrue{},
+ FirstRowID: &firstRowID,
+ }
+
+ got, split := splitParquetScanTask(task, 50)
+ require.True(t, split)
+ require.Len(t, got, 3)
+
+ assert.Equal(t, []int64{40, 30, 30}, []int64{got[0].Length,
got[1].Length, got[2].Length})
+ assert.Equal(t, []int64{0, 40, 70}, []int64{got[0].Start, got[1].Start,
got[2].Start})
+ for _, split := range got {
+ assert.Equal(t, task.File, split.File)
+ assert.Equal(t, task.DeleteFiles, split.DeleteFiles)
+ assert.Equal(t, task.Residual, split.Residual)
+ assert.Equal(t, task.FirstRowID, split.FirstRowID)
+ }
+}
+
+func TestSplitParquetScanTaskCoversSparseOffsets(t *testing.T) {
+ tests := []struct {
+ name string
+ offsets []int64
+ starts []int64
+ lengths []int64
+ }{
+ {
+ name: "omitted leading row group",
+ offsets: []int64{40, 70},
+ starts: []int64{0, 70},
+ lengths: []int64{70, 30},
+ },
+ {
+ name: "omitted interior row group",
+ offsets: []int64{8, 70},
+ starts: []int64{0, 70},
+ lengths: []int64{70, 30},
+ },
+ }
+
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ file := splitTestDataFile(t, iceberg.ParquetFile, 100,
tt.offsets)
+ task := FileScanTask{File: file, Start: 0, Length:
file.FileSizeBytes()}
+
+ got, split := splitParquetScanTask(task, 50)
+ require.True(t, split)
+ require.Len(t, got, len(tt.starts))
+ assert.Equal(t, tt.starts, []int64{got[0].Start,
got[1].Start})
+ assert.Equal(t, tt.lengths, []int64{got[0].Length,
got[1].Length})
+ assert.Equal(t, file.FileSizeBytes(),
got[len(got)-1].Start+got[len(got)-1].Length)
+ })
+ }
+}
+
+func TestSplitParquetScanTaskCoalescesRangesToTarget(t *testing.T) {
+ file := splitTestDataFile(t, iceberg.ParquetFile, 80, []int64{10, 20,
30, 40, 50, 60, 70})
+ task := FileScanTask{File: file, Start: 0, Length: file.FileSizeBytes()}
+
+ got, split := splitParquetScanTask(task, 25)
+ require.True(t, split)
+ require.Len(t, got, 4)
+ assert.Equal(t, []int64{0, 20, 40, 60}, []int64{got[0].Start,
got[1].Start, got[2].Start, got[3].Start})
+ assert.Equal(t, []int64{20, 20, 20, 20}, []int64{got[0].Length,
got[1].Length, got[2].Length, got[3].Length})
+}
+
+func TestSplitParquetScanTaskKeepsUnsafeTasksIntact(t *testing.T) {
+ baseFile := splitTestDataFile(t, iceberg.ParquetFile, 100, []int64{8,
40, 70})
+
+ tests := []struct {
+ name string
+ file iceberg.DataFile
+ task FileScanTask
+ target int64
+ }{
+ {
+ name: "small file",
+ file: splitTestDataFile(t, iceberg.ParquetFile, 100,
[]int64{8, 40, 70}),
+ target: 100,
+ },
+ {
+ name: "no split offsets",
+ file: splitTestDataFile(t, iceberg.ParquetFile, 100,
nil),
+ target: 50,
+ },
+ {
+ name: "negative offset",
+ file: splitTestDataFile(t, iceberg.ParquetFile, 100,
[]int64{-1, 40}),
+ target: 50,
+ },
+ {
+ name: "non increasing offsets",
+ file: splitTestDataFile(t, iceberg.ParquetFile, 100,
[]int64{8, 40, 40}),
+ target: 50,
+ },
+ {
+ name: "offset at file end",
+ file: splitTestDataFile(t, iceberg.ParquetFile, 100,
[]int64{8, 100}),
+ target: 50,
+ },
+ {
+ name: "partial task",
+ file: baseFile,
+ task: FileScanTask{File: baseFile, Start: 8, Length:
92},
+ target: 50,
+ },
+ {
+ name: "non parquet file",
+ file: splitTestDataFile(t, iceberg.AvroFile, 100,
[]int64{8, 40, 70}),
+ target: 50,
+ },
+ {
+ name: "invalid target",
+ file: baseFile,
+ target: 0,
+ },
+ }
+
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ task := tt.task
+ if task.File == nil {
+ task = FileScanTask{File: tt.file, Start: 0,
Length: tt.file.FileSizeBytes()}
+ }
+
+ got, split := splitParquetScanTask(task, tt.target)
+ assert.False(t, split)
+ assert.Nil(t, got)
+ })
+ }
+}
+
+func TestPlanFilesSplitsLargeParquetFileAndReadsEachRowOnce(t *testing.T) {
+ ctx := context.Background()
+ location := filepath.ToSlash(t.TempDir())
+ schema := iceberg.NewSchema(0,
+ iceberg.NestedField{ID: 1, Name: "id", Type:
iceberg.PrimitiveTypes.Int64, Required: true},
+ iceberg.NestedField{ID: 2, Name: "data", Type:
iceberg.PrimitiveTypes.String},
+ )
+ meta, err := NewMetadata(schema, iceberg.UnpartitionedSpec,
UnsortedSortOrder, location,
+ iceberg.Properties{
+ PropertyFormatVersion: "3",
+ ParquetRowGroupLimitKey: "2",
+ ReadSplitTargetSizeKey: "1",
+ })
+ require.NoError(t, err)
+
+ tbl := New(Identifier{"db", "split"}, meta,
location+"/metadata/v1.metadata.json",
+ func(context.Context) (iceio.IO, error) { return
iceio.LocalFS{}, nil },
+ &countingCatalog{metadata: meta})
+ arrowSchema := arrow.NewSchema([]arrow.Field{
+ {Name: "id", Type: arrow.PrimitiveTypes.Int64, Nullable: false},
+ {Name: "data", Type: arrow.BinaryTypes.String, Nullable: true},
+ }, nil)
+ data, err := array.TableFromJSON(memory.DefaultAllocator, arrowSchema,
[]string{
+
`[{"id":1,"data":"a"},{"id":2,"data":"b"},{"id":3,"data":"c"},{"id":4,"data":"d"},`
+
+
`{"id":5,"data":"e"},{"id":6,"data":"f"},{"id":7,"data":"g"},{"id":8,"data":"h"}]`,
+ })
+ require.NoError(t, err)
+ defer data.Release()
+
+ tbl, err = tbl.Append(ctx, array.NewTableReader(data, -1), nil)
+ require.NoError(t, err)
+
+ reporter := &metrics.InMemoryReporter{}
+ tasks, err := tbl.Scan(WithReporter(reporter)).PlanFiles(ctx)
+ require.NoError(t, err)
+ require.Len(t, tasks, 4, "one task should be planned per row group")
+ reports := reporter.Reports()
+ require.Len(t, reports, 1)
+ report, ok := reports[0].(metrics.ScanReport)
+ require.True(t, ok)
+ assert.Equal(t, int64(1), report.Metrics.ResultDataFiles.Value,
+ "split tasks must still report one data file")
+ assert.Equal(t, tasks[0].File.FileSizeBytes(),
report.Metrics.TotalFileSizeInBytes.Value,
+ "split tasks must report the original file size once")
+ for i, task := range tasks {
+ assert.Equal(t, tasks[0].File.FilePath(), task.File.FilePath())
+ assert.Positive(t, task.Length)
+ if i > 0 {
+ assert.Equal(t, tasks[i-1].Start+tasks[i-1].Length,
task.Start)
+ }
+ }
+
+ fullOffsets := tasks[0].File.SplitOffsets()
+ require.Len(t, fullOffsets, 4)
+ sparseFile := dataFileWithSplitOffsets(t, tasks[0].File,
[]int64{fullOffsets[1], fullOffsets[3]})
+ sparseTask := FileScanTask{File: sparseFile, Start: 0, Length:
sparseFile.FileSizeBytes()}
+ sparseTasks, split := splitParquetScanTask(sparseTask, 1)
+ require.True(t, split)
+ require.Len(t, sparseTasks, 2)
+
+ _, sparseRecords, err := tbl.Scan().ReadTasks(ctx, sparseTasks)
+ require.NoError(t, err)
+ seen := make(map[int64]int)
+ for record, readErr := range sparseRecords {
+ require.NoError(t, readErr)
+ ids :=
record.Column(record.Schema().FieldIndices("id")[0]).(*array.Int64)
+ for i := range ids.Len() {
+ seen[ids.Value(i)]++
+ }
+ record.Release()
+ }
+ wantSeen := map[int64]int{1: 1, 2: 1, 3: 1, 4: 1, 5: 1, 6: 1, 7: 1, 8:
1}
+ assert.Equal(t, wantSeen, seen,
+ "sparse split offsets must retain leading and interior row
groups")
+
+ result, err := tbl.Scan(WithRowLineage()).ToArrowTable(ctx)
+ require.NoError(t, err)
+ defer result.Release()
+
+ assert.EqualValues(t, 8, result.NumRows())
+ idIdx := result.Schema().FieldIndices("id")
+ rowIDIdx := result.Schema().FieldIndices(iceberg.RowIDColumnName)
+ require.Len(t, idIdx, 1)
+ require.Len(t, rowIDIdx, 1)
+ idChunks := result.Column(idIdx[0]).Data().Chunks()
+ rowIDChunks := result.Column(rowIDIdx[0]).Data().Chunks()
+ require.Equal(t, len(idChunks), len(rowIDChunks))
+
+ gotRowIDs := make(map[int64]int64, 8)
+ for chunk := range idChunks {
+ ids := idChunks[chunk].(*array.Int64)
+ rowIDs := rowIDChunks[chunk].(*array.Int64)
+ require.Equal(t, ids.Len(), rowIDs.Len())
+ for i := range ids.Len() {
+ gotRowIDs[ids.Value(i)] = rowIDs.Value(i)
+ }
+ }
+ assert.Equal(t, map[int64]int64{1: 0, 2: 1, 3: 2, 4: 3, 5: 4, 6: 5, 7:
6, 8: 7}, gotRowIDs)
+
+ posDelPath := tbl.Location() + "/data/pos-del.parquet"
+ posSc, err := SchemaToArrowSchema(iceberg.PositionalDeleteSchema, nil,
true, false)
+ require.NoError(t, err)
+ writeSplitParquetFile(t, posDelPath, posSc, fmt.Sprintf(
+ `[{"file_path":%q,"pos":1},{"file_path":%q,"pos":6}]`,
+ tasks[0].File.FilePath(), tasks[0].File.FilePath()))
+ posDelBuilder, err := iceberg.NewDataFileBuilder(
+ *iceberg.UnpartitionedSpec,
+ iceberg.EntryContentPosDeletes,
+ posDelPath,
+ iceberg.ParquetFile,
+ nil,
+ nil,
+ nil,
+ 2,
+ 256,
+ )
+ require.NoError(t, err)
+
+ tasksWithDeletes := make([]FileScanTask, len(tasks))
+ for i, task := range tasks {
+ tasksWithDeletes[i] = task
+ tasksWithDeletes[i].DeleteFiles =
[]iceberg.DataFile{posDelBuilder.Build()}
+ }
+ _, records, err := tbl.Scan(WithRowLineage()).ReadTasks(ctx,
tasksWithDeletes)
+ require.NoError(t, err)
+ gotRowIDs = make(map[int64]int64, 6)
+ for record, err := range records {
+ require.NoError(t, err)
+ ids :=
record.Column(record.Schema().FieldIndices("id")[0]).(*array.Int64)
+ rowIDs :=
record.Column(record.Schema().FieldIndices(iceberg.RowIDColumnName)[0]).(*array.Int64)
+ for i := range int(record.NumRows()) {
+ gotRowIDs[ids.Value(i)] = rowIDs.Value(i)
+ }
+ record.Release()
+ }
+ assert.Equal(t, map[int64]int64{1: 0, 3: 2, 4: 3, 5: 4, 6: 5, 8: 7},
gotRowIDs,
+ "position deletes must use original positions across split
tasks")
+
+ dvWriter := dv.NewDVWriter(iceio.LocalFS{}, func(specID int32)
*iceberg.PartitionSpec {
+ if specID == 0 {
+ return iceberg.UnpartitionedSpec
+ }
+
+ return nil
+ })
+ require.NoError(t, dvWriter.Add(tasks[0].File.FilePath(), []int64{2,
5}, 0, nil))
+ dvFiles, err := dvWriter.Flush(ctx,
tbl.Location()+"/data/scan-split.puffin")
+ require.NoError(t, err)
+
+ tasksWithDVs := make([]FileScanTask, len(tasks))
+ for i, task := range tasks {
+ tasksWithDVs[i] = task
+ tasksWithDVs[i].DeletionVectorFiles = dvFiles
+ }
+ _, records, err = tbl.Scan(WithRowLineage()).ReadTasks(ctx,
tasksWithDVs)
+ require.NoError(t, err)
+ gotRowIDs = make(map[int64]int64, 6)
+ for record, err := range records {
+ require.NoError(t, err)
+ ids :=
record.Column(record.Schema().FieldIndices("id")[0]).(*array.Int64)
+ rowIDs :=
record.Column(record.Schema().FieldIndices(iceberg.RowIDColumnName)[0]).(*array.Int64)
+ for i := range int(record.NumRows()) {
+ gotRowIDs[ids.Value(i)] = rowIDs.Value(i)
+ }
+ record.Release()
+ }
+ assert.Equal(t, map[int64]int64{1: 0, 2: 1, 4: 3, 5: 4, 7: 6, 8: 7},
gotRowIDs,
+ "deletion vectors must use original positions across split
tasks")
+}
+
+type prepareReadTrackingIO struct {
+ iceio.IO
+ closed int
+}
+
+func (fs *prepareReadTrackingIO) Open(path string) (iceio.File, error) {
+ file, err := fs.IO.Open(path)
+ if err != nil {
+ return nil, err
+ }
+
+ return &prepareReadTrackingFile{File: file, fs: fs}, nil
+}
+
+type prepareReadTrackingFile struct {
+ iceio.File
+ fs *prepareReadTrackingIO
+}
+
+func (f *prepareReadTrackingFile) Close() error {
+ f.fs.closed++
+
+ return f.File.Close()
+}
+
+func TestPrepareToReadClosesReaderOnSchemaError(t *testing.T) {
+ for _, cache := range []bool{false, true} {
+ t.Run(fmt.Sprintf("cache=%t", cache), func(t *testing.T) {
Review Comment:
nit while you're here: `fmt.Sprintf("cache=%t", cache)` (and `field_ids=%t`
below) trips `perfsprint` — `"cache=" + strconv.FormatBool(cache)` is the
lint-clean form. Separately, in `TestSplitParquetScanTask` the `for _, split :=
range got` shadows the outer `split` bool; renaming the loop var reads cleaner
and keeps `govet` shadow quiet.
##########
table/properties.go:
##########
@@ -34,6 +34,9 @@ const (
ObjectStoreEnabledKey = "write.object-storage.enabled"
ObjectStoreEnabledDefault = false
+ ReadSplitTargetSizeKey = "read.split.target-size"
Review Comment:
`read.split.target-size` is one of a three-property group in Java —
`read.split.planning-lookback` (10) and `read.split.open-file-cost` (4 MiB) go
with it and shape the coalescing. Exporting only this one means a table
carrying all three (written by a Java engine) has two silently ignored, with no
discoverable constant for them. Even without wiring the behavior now, I'd
reserve the two constants with their Java defaults so the group stays coherent
and you avoid an API bump later.
(Tiny: the `// 128 MB` comment is really 128 MiB / 134217728 — the value's
correct and matches Java's own loose comment, so only worth touching if you're
already in here.)
##########
table/internal/parquet_files.go:
##########
@@ -1844,8 +1898,18 @@ func (w wrapPqArrowReader) GetRecords(ctx
context.Context, cols []int, tester an
}
}
+ rangeSet := rowGroupTester != nil &&
+ (rowGroupTester.RangeSet || rowGroupTester.Start != 0 ||
rowGroupTester.Length != 0)
+ if rangeSet {
+ if err := validateParquetRowGroupRange(
+ w.SourceFileSize(), rowGroupTester.Start,
rowGroupTester.Length); err != nil {
Review Comment:
One subtle mismatch: `splitParquetScanTask` builds task ranges from the
manifest's `FileSizeBytes()`, but `validateParquetRowGroupRange` here checks
against `w.SourceFileSize()` (the actual opened file). If those ever disagree —
manifest written before the file was grown/truncated — the last split task's
`length` can exceed `SourceFileSize() - start` and this returns a hard
`ErrInvalidArgument` on otherwise-healthy metadata. Clamping the last range to
`min(FileSizeBytes(), SourceFileSize())`, or validating locally-planned splits
against `FileSizeBytes()`, avoids the surprise.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
