laskoviymishka commented on code in PR #2106: URL: https://github.com/apache/iceberg-go/pull/2106#discussion_r4208252701
########## table/readtasks_residual_binding_internal_test.go: ########## @@ -0,0 +1,340 @@ +// 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" + "os" + "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/stretchr/testify/require" +) + +func residualBindingTestScan(t *testing.T) (*Scan, *iceberg.Schema) { + t.Helper() + + schema := iceberg.NewSchema(0, + iceberg.NestedField{ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int64}, + ) + metadata, err := NewMetadata( + schema, iceberg.UnpartitionedSpec, UnsortedSortOrder, "mem://mixed-residuals", nil, + ) + require.NoError(t, err) + memFS := iceio.NewMemFS() + tbl := New( + Identifier{"db", "tbl"}, metadata, "metadata.json", + func(context.Context) (iceio.IO, error) { return memFS, nil }, nil, + ) + + return tbl.Scan(), schema +} + +func writeResidualBindingParquetFile(t testing.TB, path string, schema *arrow.Schema, jsonData string) { + t.Helper() + + record, _, err := array.RecordFromJSON(memory.DefaultAllocator, schema, strings.NewReader(jsonData)) + require.NoError(t, err) + defer record.Release() + + fs := iceio.LocalFS{} + writer, err := fs.Create(path) + require.NoError(t, err) + defer writer.Close() + + tbl := array.NewTableFromRecords(schema, []arrow.RecordBatch{record}) + defer tbl.Release() + + props := parquet.NewWriterProperties(parquet.WithStats(true)) + require.NoError(t, pqarrow.WriteTable( + tbl, writer, record.NumRows(), props, pqarrow.DefaultWriterProps(), + )) +} + +func TestBindReadTasksResidualsCopyOnWrite(t *testing.T) { + _, schema := residualBindingTestScan(t) + unbound := iceberg.GreaterThan(iceberg.Reference("id"), int64(1)) + bound, err := iceberg.BindExpr(schema, unbound, true) + require.NoError(t, err) + + tests := []struct { + name string + residuals []iceberg.BooleanExpression + wantAlias bool + wantSame []bool + }{ + { + name: "all bound and nil", + residuals: []iceberg.BooleanExpression{bound, nil, bound}, + wantAlias: true, + wantSame: []bool{true, false, true}, + }, + { + name: "all nil", + residuals: []iceberg.BooleanExpression{nil, nil, nil}, + wantAlias: true, + wantSame: []bool{false, false, false}, + }, + { + name: "mixed", + residuals: []iceberg.BooleanExpression{bound, nil, unbound, bound}, + wantSame: []bool{true, false, false, true}, + }, + { + name: "first task unbound", + residuals: []iceberg.BooleanExpression{unbound, bound}, + wantSame: []bool{false, true}, + }, + { + name: "all unbound", + residuals: []iceberg.BooleanExpression{unbound, unbound}, + wantSame: []bool{false, false}, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + tasks := make([]FileScanTask, len(tt.residuals)) + for i, residual := range tt.residuals { + tasks[i].Residual = residual + } + + got, err := bindReadTasksResiduals(schema, tasks, true) + require.NoError(t, err) + require.Len(t, got, len(tasks)) + if len(tasks) > 0 { + require.Equal(t, tt.wantAlias, &got[0] == &tasks[0]) + } + + for i, original := range tt.residuals { + if original == nil { + require.Nil(t, got[i].Residual) + + continue + } + + // The input plan is never rewritten, even when the output needs binding. + require.Same(t, original, tasks[i].Residual) + if tt.wantSame[i] { + require.Same(t, original, got[i].Residual) + } else { + require.NotSame(t, original, got[i].Residual) + } + state, visitErr := iceberg.VisitExpr(got[i].Residual, filterBindingVisitor{}) + require.NoError(t, visitErr) + require.True(t, state.hasBound) + require.False(t, state.hasUnbound) + } + }) + } +} + +func TestReadTasksResidualPlanIsReusable(t *testing.T) { + scan, schema := residualBindingTestScan(t) + unbound := iceberg.GreaterThan(iceberg.Reference("id"), int64(1)) + bound, err := iceberg.BindExpr(schema, unbound, true) + require.NoError(t, err) + + wrongSchema := iceberg.NewSchema(0, + iceberg.NestedField{ID: 2, Name: "other", Type: iceberg.PrimitiveTypes.Int64}, + ) + wrongBound, err := iceberg.BindExpr( + wrongSchema, iceberg.EqualTo(iceberg.Reference("other"), int64(1)), true, + ) + require.NoError(t, err) + + tests := []struct { + name string + tasks []FileScanTask + wantErr bool + }{ + { + name: "mixed plan", + tasks: []FileScanTask{{Residual: bound}, {}, {Residual: unbound}, {Residual: bound}}, + }, + { + name: "invalid bound residual", + tasks: []FileScanTask{{Residual: bound}, {Residual: unbound}, {Residual: wrongBound}}, + wantErr: true, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + originals := make([]iceberg.BooleanExpression, len(tt.tasks)) + for i := range tt.tasks { + originals[i] = tt.tasks[i].Residual + } + + // Run twice to prove the caller-owned plan remains reusable. + for range 2 { + _, _, err := scan.ReadTasks(t.Context(), tt.tasks) + if tt.wantErr { + require.ErrorIs(t, err, iceberg.ErrInvalidArgument) + require.ErrorContains(t, err, "field ID 2") + } else { + require.NoError(t, err) + } + for i, original := range originals { + if original == nil { + require.Nil(t, tt.tasks[i].Residual) + } else { + require.Same(t, original, tt.tasks[i].Residual) + } + } + } + }) + } +} + +func TestReadTasksConcurrentScansPreserveInputTasks(t *testing.T) { + schema := iceberg.NewSchema(0, + iceberg.NestedField{ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int64}, + ) + location := t.TempDir() + metadata, err := NewMetadata( + schema, iceberg.UnpartitionedSpec, UnsortedSortOrder, location, nil, + ) + require.NoError(t, err) + tbl := New( + Identifier{"db", "tbl"}, metadata, filepath.Join(location, "metadata.json"), + func(context.Context) (iceio.IO, error) { return iceio.LocalFS{}, nil }, nil, + ) + + unbound := iceberg.GreaterThan(iceberg.Reference("id"), int64(1)) + bound, err := iceberg.BindExpr(schema, unbound, true) + require.NoError(t, err) + arrowSchema, err := SchemaToArrowSchema(schema, nil, false, false) + require.NoError(t, err) + + paths := []string{ + filepath.Join(location, "data-1.parquet"), + filepath.Join(location, "data-2.parquet"), + } + rows := []string{ + `[{"id":2}]`, + `[{"id":3}]`, + } + tasks := make([]FileScanTask, len(paths)) + for i, path := range paths { + writeResidualBindingParquetFile(t, path, arrowSchema, rows[i]) + info, statErr := os.Stat(path) + require.NoError(t, statErr) + + builder, buildErr := iceberg.NewDataFileBuilder( + *iceberg.UnpartitionedSpec, + iceberg.EntryContentData, + path, + iceberg.ParquetFile, + nil, + nil, + nil, + 1, + info.Size(), + ) + require.NoError(t, buildErr) + tasks[i] = FileScanTask{File: builder.Build(), Residual: bound} + } + snapshot := append([]FileScanTask(nil), tasks...) + + errCh := make(chan error, 2) + for range 2 { + go func() { + scan := tbl.Scan(WithMaxConcurrency(4)) + _, records, readErr := scan.ReadTasks(t.Context(), tasks) + if readErr != nil { + errCh <- readErr + + return + } + for record, iterErr := range records { + if iterErr != nil { + errCh <- iterErr + + return + } + record.Release() + } + errCh <- nil + }() + } + + for range 2 { + require.NoError(t, <-errCh) + } + require.Equal(t, snapshot, tasks) +} + +func TestReadTasksPassesBoundResidualsToGetRecords(t *testing.T) { + schema := iceberg.NewSchema(0, + iceberg.NestedField{ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int64}, + ) + location := t.TempDir() + metadata, err := NewMetadata( + schema, iceberg.UnpartitionedSpec, UnsortedSortOrder, location, nil, + ) + require.NoError(t, err) + tbl := New( + Identifier{"db", "tbl"}, metadata, filepath.Join(location, "metadata.json"), + func(context.Context) (iceio.IO, error) { return iceio.LocalFS{}, nil }, nil, + ) + + arrowSchema, err := SchemaToArrowSchema(schema, nil, false, false) + require.NoError(t, err) + path := filepath.Join(location, "data.parquet") + writeResidualBindingParquetFile(t, path, arrowSchema, `[{"id":1},{"id":2},{"id":3}]`) + + info, err := os.Stat(path) + require.NoError(t, err) + builder, err := iceberg.NewDataFileBuilder( + *iceberg.UnpartitionedSpec, + iceberg.EntryContentData, + path, + iceberg.ParquetFile, + nil, + nil, + nil, + 3, + info.Size(), + ) + require.NoError(t, err) + + unbound := iceberg.GreaterThan(iceberg.Reference("id"), int64(1)) + tasks := []FileScanTask{{File: builder.Build(), Residual: unbound}} + + _, records, err := tbl.Scan().ReadTasks(t.Context(), tasks) + require.NoError(t, err) + + var rows int64 + for record, readErr := range records { + require.NoError(t, readErr) + rows += record.NumRows() + record.Release() + } + + require.EqualValues(t, 2, rows) Review Comment: Once the fixture's fixed this is the test that's meant to bite, but with one task and a bare row total it still can't. `require.EqualValues(2, rows)` passes for the right reason today, but a reverted clone or a bind that passed `tasks` instead of `readTasks` would stay green, because nothing shifts a single file's count. To make that swap actually fail here I'd go to 2-3 tasks with distinct per-file rows so a wrong index moves the total, or add a negative control that calls `GetRecords` with `taskResidualsBound: true` and an unbound residual and expects an error. That's the exact boundary from rounds 1 and 3. The concurrency test has the same blind spot, `require.Equal(snapshot, tasks)` also holds under the old unconditional clone, so a mixed-residual case is what would exercise the clone path concurrently (and only under `-race`). ########## table/readtasks_residual_binding_internal_test.go: ########## @@ -0,0 +1,340 @@ +// 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" + "os" + "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/stretchr/testify/require" +) + +func residualBindingTestScan(t *testing.T) (*Scan, *iceberg.Schema) { + t.Helper() + + schema := iceberg.NewSchema(0, + iceberg.NestedField{ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int64}, + ) + metadata, err := NewMetadata( + schema, iceberg.UnpartitionedSpec, UnsortedSortOrder, "mem://mixed-residuals", nil, + ) + require.NoError(t, err) + memFS := iceio.NewMemFS() + tbl := New( + Identifier{"db", "tbl"}, metadata, "metadata.json", + func(context.Context) (iceio.IO, error) { return memFS, nil }, nil, + ) + + return tbl.Scan(), schema +} + +func writeResidualBindingParquetFile(t testing.TB, path string, schema *arrow.Schema, jsonData string) { + t.Helper() + + record, _, err := array.RecordFromJSON(memory.DefaultAllocator, schema, strings.NewReader(jsonData)) + require.NoError(t, err) + defer record.Release() + + fs := iceio.LocalFS{} + writer, err := fs.Create(path) + require.NoError(t, err) + defer writer.Close() + + tbl := array.NewTableFromRecords(schema, []arrow.RecordBatch{record}) + defer tbl.Release() + + props := parquet.NewWriterProperties(parquet.WithStats(true)) + require.NoError(t, pqarrow.WriteTable( + tbl, writer, record.NumRows(), props, pqarrow.DefaultWriterProps(), + )) +} + +func TestBindReadTasksResidualsCopyOnWrite(t *testing.T) { + _, schema := residualBindingTestScan(t) + unbound := iceberg.GreaterThan(iceberg.Reference("id"), int64(1)) + bound, err := iceberg.BindExpr(schema, unbound, true) + require.NoError(t, err) + + tests := []struct { + name string + residuals []iceberg.BooleanExpression + wantAlias bool + wantSame []bool + }{ + { + name: "all bound and nil", + residuals: []iceberg.BooleanExpression{bound, nil, bound}, + wantAlias: true, + wantSame: []bool{true, false, true}, + }, + { + name: "all nil", + residuals: []iceberg.BooleanExpression{nil, nil, nil}, + wantAlias: true, + wantSame: []bool{false, false, false}, + }, + { + name: "mixed", + residuals: []iceberg.BooleanExpression{bound, nil, unbound, bound}, + wantSame: []bool{true, false, false, true}, + }, + { + name: "first task unbound", + residuals: []iceberg.BooleanExpression{unbound, bound}, + wantSame: []bool{false, true}, + }, + { + name: "all unbound", + residuals: []iceberg.BooleanExpression{unbound, unbound}, + wantSame: []bool{false, false}, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + tasks := make([]FileScanTask, len(tt.residuals)) + for i, residual := range tt.residuals { + tasks[i].Residual = residual + } + + got, err := bindReadTasksResiduals(schema, tasks, true) + require.NoError(t, err) + require.Len(t, got, len(tasks)) + if len(tasks) > 0 { + require.Equal(t, tt.wantAlias, &got[0] == &tasks[0]) + } + + for i, original := range tt.residuals { + if original == nil { + require.Nil(t, got[i].Residual) + + continue + } + + // The input plan is never rewritten, even when the output needs binding. + require.Same(t, original, tasks[i].Residual) + if tt.wantSame[i] { + require.Same(t, original, got[i].Residual) + } else { + require.NotSame(t, original, got[i].Residual) + } + state, visitErr := iceberg.VisitExpr(got[i].Residual, filterBindingVisitor{}) + require.NoError(t, visitErr) + require.True(t, state.hasBound) + require.False(t, state.hasUnbound) + } + }) + } +} + +func TestReadTasksResidualPlanIsReusable(t *testing.T) { + scan, schema := residualBindingTestScan(t) + unbound := iceberg.GreaterThan(iceberg.Reference("id"), int64(1)) + bound, err := iceberg.BindExpr(schema, unbound, true) + require.NoError(t, err) + + wrongSchema := iceberg.NewSchema(0, + iceberg.NestedField{ID: 2, Name: "other", Type: iceberg.PrimitiveTypes.Int64}, + ) + wrongBound, err := iceberg.BindExpr( + wrongSchema, iceberg.EqualTo(iceberg.Reference("other"), int64(1)), true, + ) + require.NoError(t, err) + + tests := []struct { + name string + tasks []FileScanTask + wantErr bool + }{ + { + name: "mixed plan", + tasks: []FileScanTask{{Residual: bound}, {}, {Residual: unbound}, {Residual: bound}}, + }, + { + name: "invalid bound residual", + tasks: []FileScanTask{{Residual: bound}, {Residual: unbound}, {Residual: wrongBound}}, + wantErr: true, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + originals := make([]iceberg.BooleanExpression, len(tt.tasks)) + for i := range tt.tasks { + originals[i] = tt.tasks[i].Residual + } + + // Run twice to prove the caller-owned plan remains reusable. + for range 2 { + _, _, err := scan.ReadTasks(t.Context(), tt.tasks) + if tt.wantErr { + require.ErrorIs(t, err, iceberg.ErrInvalidArgument) + require.ErrorContains(t, err, "field ID 2") + } else { + require.NoError(t, err) + } + for i, original := range originals { + if original == nil { + require.Nil(t, tt.tasks[i].Residual) + } else { + require.Same(t, original, tt.tasks[i].Residual) + } + } + } + }) + } +} + +func TestReadTasksConcurrentScansPreserveInputTasks(t *testing.T) { + schema := iceberg.NewSchema(0, + iceberg.NestedField{ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int64}, + ) + location := t.TempDir() + metadata, err := NewMetadata( + schema, iceberg.UnpartitionedSpec, UnsortedSortOrder, location, nil, + ) + require.NoError(t, err) + tbl := New( + Identifier{"db", "tbl"}, metadata, filepath.Join(location, "metadata.json"), + func(context.Context) (iceio.IO, error) { return iceio.LocalFS{}, nil }, nil, + ) + + unbound := iceberg.GreaterThan(iceberg.Reference("id"), int64(1)) + bound, err := iceberg.BindExpr(schema, unbound, true) + require.NoError(t, err) + arrowSchema, err := SchemaToArrowSchema(schema, nil, false, false) Review Comment: This has to be `true`, it's what's turning CI red on both new tests. The third arg to `SchemaToArrowSchema` is `includeFieldIDs`, so `false` writes the parquet without `PARQUET:field_id` and the reader bails at file open with `missing field_id`, before any residual is ever evaluated. Same call at line 305. ```suggestion arrowSchema, err := SchemaToArrowSchema(schema, nil, true, false) ``` Might be worth folding the arrow-schema conversion into `writeResidualBindingParquetFile` so neither call site can set this wrong again. ########## table/scanner.go: ########## @@ -2161,6 +2195,12 @@ func (scan *Scan) ToArrowRecords(ctx context.Context) (*arrow.Schema, iter.Seq2[ // reached; if no such task is processed, the file is not read and its error is not // returned. The returned iterator is single-use. // +// The caller must treat tasks and every task element as immutable until a range +// over the returned iterator has returned. If the iterator is never ranged, +// tasks must remain immutable while it may still be used. When no residual needs Review Comment: The nested-slices note landed, good. The middle sentence still reads circular the same way "abandoned" did last round: "while it may still be used" isn't an event the caller can pin down. I'd tie it to concrete caller actions instead: > The caller must not mutate tasks or any task element while the returned iterator can still be used, that is, until a range over it has returned or the iterator is otherwise dropped. ReadTasks may retain the caller's backing array when no residual needs binding, and any clone is shallow, so nested slices (for example DeleteFiles) are always shared with the caller. (This rewrites the whole paragraph rather than a single line, so it's not a one-line suggestion.) -- 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]
