laskoviymishka commented on code in PR #2170: URL: https://github.com/apache/iceberg-go/pull/2170#discussion_r4326440181
########## table/changelog_reader.go: ########## @@ -0,0 +1,350 @@ +// 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" + "iter" + "maps" + "strconv" + "sync/atomic" + + "github.com/apache/arrow-go/v18/arrow" + "github.com/apache/arrow-go/v18/arrow/array" + "github.com/apache/arrow-go/v18/arrow/compute" + "github.com/apache/arrow-go/v18/arrow/memory" + "github.com/apache/iceberg-go" +) + +const ( + // ChangelogChangeTypeColumn is INSERT or DELETE. A copy-on-write overwrite + // emits both for rows that did not change. UPDATE_BEFORE and UPDATE_AFTER + // are not produced. + ChangelogChangeTypeColumn = "_change_type" + // ChangelogChangeOrdinalColumn is the zero-based ordinal of the snapshot + // that produced the change, shared by every row from that snapshot. The + // Arrow type is int32, matching Java MetadataColumns.CHANGE_ORDINAL. + ChangelogChangeOrdinalColumn = "_change_ordinal" + // ChangelogCommitSnapshotIDColumn is the snapshot ID that committed the change. + ChangelogCommitSnapshotIDColumn = "_commit_snapshot_id" +) + +// ToArrowRecords plans the changelog and reads it. The result is a raw +// file-level changelog: a copy-on-write overwrite emits a DELETE and an +// INSERT for every row in the rewritten files, including rows that did not +// change, and UPDATE_BEFORE and UPDATE_AFTER are not produced. Added and +// removed data files are read with the deletes already stored on their scan +// task. A plan that includes a row-level delete returns an error instead of +// dropping those rows. +func (s *IncrementalChangelogScan) ToArrowRecords(ctx context.Context) (*arrow.Schema, iter.Seq2[arrow.RecordBatch, error], error) { + tasks, err := s.PlanFiles(ctx) + if err != nil { + return nil, nil, err + } + + return s.Read(ctx, tasks) +} + +// Read materializes changelog rows for tasks. Each batch has the scan +// projection followed by _change_type, _change_ordinal, and +// _commit_snapshot_id. This is a raw file-level changelog: unchanged rows in +// a rewritten file are emitted as DELETE and INSERT, and UPDATE_BEFORE and +// UPDATE_AFTER are not produced. AddedRowsScanTask and DeletedDataFileScanTask +// are read with Scan.ReadTasks, so same-snapshot deletes and deletes that +// already applied to a removed file are applied before the changelog columns +// are added. DeletedRowsScanTask returns an error; those rows are produced by +// a later reader. The returned iterator is single-use. The caller releases +// every batch. +func (s *IncrementalChangelogScan) Read(ctx context.Context, tasks []ChangelogScanTask) (*arrow.Schema, iter.Seq2[arrow.RecordBatch, error], error) { + if s == nil || s.scan == nil { + return nil, nil, fmt.Errorf("%w: incremental changelog scan is not initialized", ErrInvalidOperation) + } + if err := ctx.Err(); err != nil { + return nil, nil, err + } + for _, task := range tasks { + if err := readableChangelogTask(task); err != nil { + return nil, nil, err + } + } + + schema, useLargeTypes, err := changelogArrowSchema(s.scan) + if err != nil { + return nil, nil, err + } + if s.scan.limit == 0 || len(tasks) == 0 { + return schema, func(yield func(arrow.RecordBatch, error) bool) {}, nil + } + + readScan := s.scan.clone() + limit := s.scan.limit + + return schema, func(yield func(arrow.RecordBatch, error) bool) { + var emitted int64 + for _, task := range tasks { + if err := ctx.Err(); err != nil { + yield(nil, err) + + return + } + if limit > 0 { + readScan.limit = limit - emitted + } + + _, records, err := readScan.ReadTasks(ctx, []FileScanTask{task.ScanTask()}) Review Comment: Each task goes through its own single-element `ReadTasks` call, so `WithMaxConcurrency` never applies across files and a changelog over many files reads one file at a time. Serial is probably the right default since consumers expect output in task order. I'd say so on `Read` and open a follow-up for an order-preserving parallel read. ########## table/changelog_reader.go: ########## @@ -0,0 +1,350 @@ +// 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" + "iter" + "maps" + "strconv" + "sync/atomic" + + "github.com/apache/arrow-go/v18/arrow" + "github.com/apache/arrow-go/v18/arrow/array" + "github.com/apache/arrow-go/v18/arrow/compute" + "github.com/apache/arrow-go/v18/arrow/memory" + "github.com/apache/iceberg-go" +) + +const ( + // ChangelogChangeTypeColumn is INSERT or DELETE. A copy-on-write overwrite + // emits both for rows that did not change. UPDATE_BEFORE and UPDATE_AFTER + // are not produced. + ChangelogChangeTypeColumn = "_change_type" + // ChangelogChangeOrdinalColumn is the zero-based ordinal of the snapshot + // that produced the change, shared by every row from that snapshot. The + // Arrow type is int32, matching Java MetadataColumns.CHANGE_ORDINAL. + ChangelogChangeOrdinalColumn = "_change_ordinal" + // ChangelogCommitSnapshotIDColumn is the snapshot ID that committed the change. + ChangelogCommitSnapshotIDColumn = "_commit_snapshot_id" +) + +// ToArrowRecords plans the changelog and reads it. The result is a raw +// file-level changelog: a copy-on-write overwrite emits a DELETE and an +// INSERT for every row in the rewritten files, including rows that did not +// change, and UPDATE_BEFORE and UPDATE_AFTER are not produced. Added and +// removed data files are read with the deletes already stored on their scan +// task. A plan that includes a row-level delete returns an error instead of +// dropping those rows. +func (s *IncrementalChangelogScan) ToArrowRecords(ctx context.Context) (*arrow.Schema, iter.Seq2[arrow.RecordBatch, error], error) { + tasks, err := s.PlanFiles(ctx) + if err != nil { + return nil, nil, err + } + + return s.Read(ctx, tasks) +} + +// Read materializes changelog rows for tasks. Each batch has the scan +// projection followed by _change_type, _change_ordinal, and +// _commit_snapshot_id. This is a raw file-level changelog: unchanged rows in +// a rewritten file are emitted as DELETE and INSERT, and UPDATE_BEFORE and +// UPDATE_AFTER are not produced. AddedRowsScanTask and DeletedDataFileScanTask +// are read with Scan.ReadTasks, so same-snapshot deletes and deletes that +// already applied to a removed file are applied before the changelog columns +// are added. DeletedRowsScanTask returns an error; those rows are produced by +// a later reader. The returned iterator is single-use. The caller releases +// every batch. +func (s *IncrementalChangelogScan) Read(ctx context.Context, tasks []ChangelogScanTask) (*arrow.Schema, iter.Seq2[arrow.RecordBatch, error], error) { + if s == nil || s.scan == nil { + return nil, nil, fmt.Errorf("%w: incremental changelog scan is not initialized", ErrInvalidOperation) + } + if err := ctx.Err(); err != nil { + return nil, nil, err + } + for _, task := range tasks { + if err := readableChangelogTask(task); err != nil { + return nil, nil, err + } + } + + schema, useLargeTypes, err := changelogArrowSchema(s.scan) + if err != nil { + return nil, nil, err + } + if s.scan.limit == 0 || len(tasks) == 0 { + return schema, func(yield func(arrow.RecordBatch, error) bool) {}, nil + } + + readScan := s.scan.clone() + limit := s.scan.limit + + return schema, func(yield func(arrow.RecordBatch, error) bool) { + var emitted int64 + for _, task := range tasks { + if err := ctx.Err(); err != nil { + yield(nil, err) + + return + } + if limit > 0 { + readScan.limit = limit - emitted + } + + _, records, err := readScan.ReadTasks(ctx, []FileScanTask{task.ScanTask()}) + if err != nil { + yield(nil, err) + + return + } + + for rec, err := range records { + if err != nil { + yield(nil, err) + + return + } + + stamped, rows, err := stampChangelogBatch(ctx, schema, useLargeTypes, rec, task, limit, emitted) + rec.Release() + if err != nil { + yield(nil, err) + + return + } + if stamped == nil { + continue + } + if !yield(stamped, nil) { + return + } + emitted += rows + if limit > 0 && emitted >= limit { + return + } + } + } + }, nil +} + +func readableChangelogTask(task ChangelogScanTask) error { + switch task.(type) { + case AddedRowsScanTask, DeletedDataFileScanTask: + return nil + case DeletedRowsScanTask: + return fmt.Errorf("%w: reading deleted rows from added delete files is not supported", iceberg.ErrNotImplemented) + default: + return fmt.Errorf("%w: unsupported changelog scan task %T", iceberg.ErrInvalidArgument, task) + } +} + +func changelogArrowSchema(scan *Scan) (*arrow.Schema, bool, error) { + if atomic.LoadUint32(&scan.closed) != 0 { + return nil, false, fmt.Errorf("%w: scan is closed", ErrInvalidOperation) + } + if scan.selectorErr != nil { + return nil, false, scan.selectorErr + } + + projected, err := scan.Projection() + if err != nil { + return nil, false, err + } + + useLargeTypes, err := strconv.ParseBool(scan.options.Get(ScanOptionArrowUseLargeTypes, "false")) + if err != nil { + useLargeTypes = false + } + + props := scan.metadata.Properties() + if scan.arrowBatchSize > 0 { + props = maps.Clone(props) + if props == nil { + props = iceberg.Properties{} + } + props[ParquetBatchSizeKey] = strconv.Itoa(scan.arrowBatchSize) + } + + dataSchema, err := SchemaToArrowSchemaWithOptions(projected, ArrowSchemaOptions{ + UseLargeTypes: useLargeTypes, + TableProperties: props, + }) + if err != nil { + return nil, false, err + } + + if err := rejectChangelogColumnCollision(dataSchema); err != nil { + return nil, false, err + } + + schema := dataSchema + for _, field := range changelogArrowFields(useLargeTypes) { + schema, err = schema.AddField(schema.NumFields(), field) + if err != nil { + return nil, false, err + } + } + + return schema, useLargeTypes, nil +} + +func rejectChangelogColumnCollision(schema *arrow.Schema) error { + for _, name := range []string{ + ChangelogChangeTypeColumn, + ChangelogChangeOrdinalColumn, + ChangelogCommitSnapshotIDColumn, + } { + if len(schema.FieldIndices(name)) > 0 { + return fmt.Errorf("%w: projected column %q conflicts with a changelog column", + iceberg.ErrInvalidSchema, name) + } + } + + return nil +} + +func changelogBatchMatches(out, data *arrow.Schema) error { + want := out.NumFields() - 3 + if data.NumFields() != want { + return fmt.Errorf("%w: read batch has %d columns, changelog schema expects %d", + iceberg.ErrInvalidSchema, data.NumFields(), want) + } + for i := range want { + if !data.Field(i).Equal(out.Field(i)) { + return fmt.Errorf("%w: read column %q does not match the changelog schema", + iceberg.ErrInvalidSchema, data.Field(i).Name) + } + } + + return nil +} + +func changelogArrowFields(useLargeTypes bool) []arrow.Field { + changeType := arrow.BinaryTypes.String + if useLargeTypes { + changeType = arrow.BinaryTypes.LargeString + } + + return []arrow.Field{ + {Name: ChangelogChangeTypeColumn, Type: changeType, Nullable: false}, + {Name: ChangelogChangeOrdinalColumn, Type: arrow.PrimitiveTypes.Int32, Nullable: false}, + {Name: ChangelogCommitSnapshotIDColumn, Type: arrow.PrimitiveTypes.Int64, Nullable: false}, + } +} + +// stampChangelogBatch appends the changelog columns and applies any remaining +// row limit. A nil batch means the limit was already exhausted. rec is not +// released here. +func stampChangelogBatch(ctx context.Context, schema *arrow.Schema, useLargeTypes bool, rec arrow.RecordBatch, task ChangelogScanTask, limit, emitted int64) (arrow.RecordBatch, int64, error) { + if limit > 0 && emitted >= limit { + return nil, 0, nil + } + + rows := rec.NumRows() + source := rec + if limit > 0 && emitted+rows > limit { + source = rec.NewSlice(0, limit-emitted) Review Comment: `readScan.limit` is already set to `limit - emitted` before `ReadTasks`. If `ReadTasks` truncates to that (I think it does), this slice never runs, and `TestIncrementalChangelogReadLimitCutsABatch` is testing `ReadTasks`' limit, not this branch. I'd either drop the slice or test `stampChangelogBatch` directly with a batch larger than the remaining budget. ########## table/changelog_reader.go: ########## @@ -0,0 +1,350 @@ +// 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" + "iter" + "maps" + "strconv" + "sync/atomic" + + "github.com/apache/arrow-go/v18/arrow" + "github.com/apache/arrow-go/v18/arrow/array" + "github.com/apache/arrow-go/v18/arrow/compute" + "github.com/apache/arrow-go/v18/arrow/memory" + "github.com/apache/iceberg-go" +) + +const ( + // ChangelogChangeTypeColumn is INSERT or DELETE. A copy-on-write overwrite + // emits both for rows that did not change. UPDATE_BEFORE and UPDATE_AFTER + // are not produced. + ChangelogChangeTypeColumn = "_change_type" + // ChangelogChangeOrdinalColumn is the zero-based ordinal of the snapshot + // that produced the change, shared by every row from that snapshot. The + // Arrow type is int32, matching Java MetadataColumns.CHANGE_ORDINAL. + ChangelogChangeOrdinalColumn = "_change_ordinal" + // ChangelogCommitSnapshotIDColumn is the snapshot ID that committed the change. + ChangelogCommitSnapshotIDColumn = "_commit_snapshot_id" +) + +// ToArrowRecords plans the changelog and reads it. The result is a raw +// file-level changelog: a copy-on-write overwrite emits a DELETE and an +// INSERT for every row in the rewritten files, including rows that did not +// change, and UPDATE_BEFORE and UPDATE_AFTER are not produced. Added and +// removed data files are read with the deletes already stored on their scan +// task. A plan that includes a row-level delete returns an error instead of +// dropping those rows. +func (s *IncrementalChangelogScan) ToArrowRecords(ctx context.Context) (*arrow.Schema, iter.Seq2[arrow.RecordBatch, error], error) { + tasks, err := s.PlanFiles(ctx) + if err != nil { + return nil, nil, err + } + + return s.Read(ctx, tasks) +} + +// Read materializes changelog rows for tasks. Each batch has the scan +// projection followed by _change_type, _change_ordinal, and +// _commit_snapshot_id. This is a raw file-level changelog: unchanged rows in +// a rewritten file are emitted as DELETE and INSERT, and UPDATE_BEFORE and +// UPDATE_AFTER are not produced. AddedRowsScanTask and DeletedDataFileScanTask +// are read with Scan.ReadTasks, so same-snapshot deletes and deletes that +// already applied to a removed file are applied before the changelog columns +// are added. DeletedRowsScanTask returns an error; those rows are produced by +// a later reader. The returned iterator is single-use. The caller releases +// every batch. +func (s *IncrementalChangelogScan) Read(ctx context.Context, tasks []ChangelogScanTask) (*arrow.Schema, iter.Seq2[arrow.RecordBatch, error], error) { + if s == nil || s.scan == nil { + return nil, nil, fmt.Errorf("%w: incremental changelog scan is not initialized", ErrInvalidOperation) + } + if err := ctx.Err(); err != nil { + return nil, nil, err + } + for _, task := range tasks { + if err := readableChangelogTask(task); err != nil { + return nil, nil, err + } + } + + schema, useLargeTypes, err := changelogArrowSchema(s.scan) + if err != nil { + return nil, nil, err + } + if s.scan.limit == 0 || len(tasks) == 0 { + return schema, func(yield func(arrow.RecordBatch, error) bool) {}, nil + } + + readScan := s.scan.clone() + limit := s.scan.limit + + return schema, func(yield func(arrow.RecordBatch, error) bool) { + var emitted int64 + for _, task := range tasks { + if err := ctx.Err(); err != nil { + yield(nil, err) + + return + } + if limit > 0 { + readScan.limit = limit - emitted + } + + _, records, err := readScan.ReadTasks(ctx, []FileScanTask{task.ScanTask()}) + if err != nil { + yield(nil, err) + + return + } + + for rec, err := range records { + if err != nil { + yield(nil, err) + + return + } + + stamped, rows, err := stampChangelogBatch(ctx, schema, useLargeTypes, rec, task, limit, emitted) + rec.Release() + if err != nil { + yield(nil, err) + + return + } + if stamped == nil { + continue + } + if !yield(stamped, nil) { + return + } + emitted += rows + if limit > 0 && emitted >= limit { + return + } + } + } + }, nil +} + +func readableChangelogTask(task ChangelogScanTask) error { + switch task.(type) { + case AddedRowsScanTask, DeletedDataFileScanTask: + return nil + case DeletedRowsScanTask: + return fmt.Errorf("%w: reading deleted rows from added delete files is not supported", iceberg.ErrNotImplemented) + default: + return fmt.Errorf("%w: unsupported changelog scan task %T", iceberg.ErrInvalidArgument, task) + } +} + +func changelogArrowSchema(scan *Scan) (*arrow.Schema, bool, error) { + if atomic.LoadUint32(&scan.closed) != 0 { + return nil, false, fmt.Errorf("%w: scan is closed", ErrInvalidOperation) + } + if scan.selectorErr != nil { + return nil, false, scan.selectorErr + } + + projected, err := scan.Projection() + if err != nil { + return nil, false, err + } + + useLargeTypes, err := strconv.ParseBool(scan.options.Get(ScanOptionArrowUseLargeTypes, "false")) + if err != nil { + useLargeTypes = false + } + + props := scan.metadata.Properties() + if scan.arrowBatchSize > 0 { + props = maps.Clone(props) + if props == nil { + props = iceberg.Properties{} + } + props[ParquetBatchSizeKey] = strconv.Itoa(scan.arrowBatchSize) + } + + dataSchema, err := SchemaToArrowSchemaWithOptions(projected, ArrowSchemaOptions{ + UseLargeTypes: useLargeTypes, + TableProperties: props, + }) + if err != nil { + return nil, false, err + } + + if err := rejectChangelogColumnCollision(dataSchema); err != nil { + return nil, false, err + } + + schema := dataSchema + for _, field := range changelogArrowFields(useLargeTypes) { + schema, err = schema.AddField(schema.NumFields(), field) + if err != nil { + return nil, false, err + } + } + + return schema, useLargeTypes, nil +} + +func rejectChangelogColumnCollision(schema *arrow.Schema) error { + for _, name := range []string{ + ChangelogChangeTypeColumn, + ChangelogChangeOrdinalColumn, + ChangelogCommitSnapshotIDColumn, + } { + if len(schema.FieldIndices(name)) > 0 { + return fmt.Errorf("%w: projected column %q conflicts with a changelog column", + iceberg.ErrInvalidSchema, name) + } + } + + return nil +} + +func changelogBatchMatches(out, data *arrow.Schema) error { + want := out.NumFields() - 3 + if data.NumFields() != want { + return fmt.Errorf("%w: read batch has %d columns, changelog schema expects %d", + iceberg.ErrInvalidSchema, data.NumFields(), want) + } + for i := range want { + if !data.Field(i).Equal(out.Field(i)) { Review Comment: `Field.Equal` also compares metadata and nested child fields, and the tests only read `id` and `data`. This schema is built separately from what `arrowScan` produces. I'd add one read with a list or struct column and one with `ScanOptionArrowUseLargeTypes`. Then if the two schemas ever drift apart, a test fails, not a valid read on someone's real table. ########## table/changelog_reader_external_test.go: ########## @@ -0,0 +1,344 @@ +// 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_test + +import ( + "context" + "fmt" + "iter" + "os" + "testing" + + "github.com/apache/arrow-go/v18/arrow" + "github.com/apache/arrow-go/v18/arrow/array" + "github.com/apache/arrow-go/v18/arrow/compute" + "github.com/apache/arrow-go/v18/arrow/memory" + "github.com/apache/iceberg-go" + "github.com/apache/iceberg-go/table" + "github.com/stretchr/testify/require" +) + +func TestIncrementalChangelogReadEmptyTable(t *testing.T) { + schema, records, err := newV2RowLineageTestTable(t).NewIncrementalChangelogScan().ToArrowRecords(t.Context()) + require.NoError(t, err) + require.Equal(t, []string{ + "id", "data", + table.ChangelogChangeTypeColumn, + table.ChangelogChangeOrdinalColumn, + table.ChangelogCommitSnapshotIDColumn, + }, schemaFieldNames(schema)) + + require.Empty(t, collectChangelogRows(t, records)) +} + +func TestIncrementalChangelogReadFileChanges(t *testing.T) { + ctx := t.Context() + tbl := changelogOverwriteTable(t, ctx) + + tasks, err := tbl.NewIncrementalChangelogScan().PlanFiles(ctx) + require.NoError(t, err) + require.Len(t, tasks, 3) + + schema, records, err := tbl.NewIncrementalChangelogScan().Read(ctx, tasks) + require.NoError(t, err) + require.Equal(t, []string{ + "id", "data", + table.ChangelogChangeTypeColumn, + table.ChangelogChangeOrdinalColumn, + table.ChangelogCommitSnapshotIDColumn, + }, schemaFieldNames(schema)) + + got := collectChangelogRows(t, records) + require.Equal(t, []changelogRow{ + {id: 1, data: "a", op: string(table.ChangelogOpInsert), ordinal: int64(tasks[0].ChangeOrdinal()), snapshot: tasks[0].CommitSnapshotID()}, + {id: 1, data: "a", op: string(table.ChangelogOpDelete), ordinal: int64(tasks[1].ChangeOrdinal()), snapshot: tasks[1].CommitSnapshotID()}, + {id: 2, data: "b", op: string(table.ChangelogOpInsert), ordinal: int64(tasks[2].ChangeOrdinal()), snapshot: tasks[2].CommitSnapshotID()}, + }, got) + require.NotEqual(t, tasks[0].CommitSnapshotID(), tasks[1].CommitSnapshotID()) + require.Equal(t, tasks[1].CommitSnapshotID(), tasks[2].CommitSnapshotID()) + require.Equal(t, 0, tasks[0].ChangeOrdinal()) + require.Equal(t, 1, tasks[1].ChangeOrdinal()) + require.Equal(t, 1, tasks[2].ChangeOrdinal()) +} + +func TestIncrementalChangelogReadHonorsProjectionFilterAndLimit(t *testing.T) { + ctx := t.Context() + tbl := changelogOverwriteTable(t, ctx) + + schema, records, err := tbl.NewIncrementalChangelogScan( + table.WithSelectedFields("id"), + table.WithRowFilter(iceberg.EqualTo(iceberg.Reference("id"), int64(2))), + table.WithLimit(1), + ).ToArrowRecords(ctx) + require.NoError(t, err) + require.Equal(t, []string{ + "id", + table.ChangelogChangeTypeColumn, + table.ChangelogChangeOrdinalColumn, + table.ChangelogCommitSnapshotIDColumn, + }, schemaFieldNames(schema)) + + got := collectChangelogRows(t, records) + require.Equal(t, []changelogRow{ + {id: 2, op: string(table.ChangelogOpInsert), ordinal: 1}, + }, clearChangelogSnapshot(got)) +} + +func TestIncrementalChangelogReadLimitCutsABatch(t *testing.T) { + ctx := t.Context() + tbl := newV3RowLineageTestTable(t) + 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"}]`, + }) + require.NoError(t, err) + t.Cleanup(data.Release) + tbl, err = tbl.Append(ctx, array.NewTableReader(data, -1), nil) + require.NoError(t, err) + + _, records, err := tbl.NewIncrementalChangelogScan( + table.WithSelectedFields("id"), + table.WithLimit(2), + ).ToArrowRecords(ctx) + require.NoError(t, err) + got := collectChangelogRows(t, records) + require.Equal(t, []changelogRow{ + {id: 1, op: string(table.ChangelogOpInsert), ordinal: 0}, + {id: 2, op: string(table.ChangelogOpInsert), ordinal: 0}, + }, clearChangelogSnapshot(got)) +} + +func TestIncrementalChangelogReadReleasesCheckedAllocator(t *testing.T) { + tbl := newV3RowLineageTestTable(t) + 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"}]`, + }) + require.NoError(t, err) + t.Cleanup(data.Release) + tbl, err = tbl.Append(t.Context(), array.NewTableReader(data, -1), nil) + require.NoError(t, err) + + read := func(t *testing.T, stopEarly bool) { + t.Helper() + + mem := memory.NewCheckedAllocator(memory.DefaultAllocator) + ctx := compute.WithAllocator(t.Context(), mem) + defer mem.AssertSize(t, 0) + + _, records, err := tbl.NewIncrementalChangelogScan( + table.WithSelectedFields("id"), + ).ToArrowRecords(ctx) + require.NoError(t, err) + for rec, err := range records { + require.NoError(t, err) + rec.Release() + if stopEarly { + break + } + } + } + + t.Run("full drain", func(t *testing.T) { read(t, false) }) + t.Run("early break", func(t *testing.T) { read(t, true) }) +} + +func TestIncrementalChangelogReadAppliesFileDeletes(t *testing.T) { + ctx := t.Context() + tbl := newV2RowLineageTestTable(t) + arrowSchema, err := table.SchemaToArrowSchema(tbl.Metadata().CurrentSchema(), nil, true, false) + require.NoError(t, err) + + dataPath := tbl.Location() + "/data/data-001.parquet" + writeParquetFile(t, dataPath, arrowSchema, `[ + {"id": 1, "data": "a"}, + {"id": 2, "data": "b"}, + {"id": 3, "data": "c"} + ]`) + posPath := tbl.Location() + "/data/pos-del.parquet" + writeParquetFile(t, posPath, table.PositionalDeleteArrowSchema, fmt.Sprintf(`[{"file_path": %q, "pos": 1}]`, dataPath)) + + tx := tbl.NewTransaction() + require.NoError(t, tx.NewRowDelta(nil). + AddRows(changelogDataFile(t, dataPath, 3)). + AddDeletes(changelogPosDeleteFile(t, posPath)). + Commit(ctx)) + tbl, err = tx.Commit(ctx) + require.NoError(t, err) + + replacementSchema := arrow.NewSchema([]arrow.Field{ + {Name: "id", Type: arrow.PrimitiveTypes.Int64, Nullable: false}, + {Name: "data", Type: arrow.BinaryTypes.String, Nullable: true}, + }, nil) + replacement, err := array.TableFromJSON(memory.DefaultAllocator, replacementSchema, []string{ + `[{"id": 4, "data": "d"}]`, + }) + require.NoError(t, err) + t.Cleanup(replacement.Release) + tbl, err = tbl.Overwrite(ctx, array.NewTableReader(replacement, -1), nil, table.WithOverwriteConcurrency(1)) + require.NoError(t, err) + + _, records, err := tbl.NewIncrementalChangelogScan().ToArrowRecords(ctx) + require.NoError(t, err) + got := collectChangelogRows(t, records) + require.Equal(t, []int64{1, 3}, changelogIDs(got, table.ChangelogOpInsert, 0)) + require.Equal(t, []int64{1, 3}, changelogIDs(got, table.ChangelogOpDelete, 1)) Review Comment: This DELETE only drops row 2 because the position delete landed at ordinal 0, inside the range, so the read never goes through `preRange`. I'd add a read where the delete is committed before `FromSnapshotExclusive` and the file is removed in range. I'd also add one with a DV, ideally two in one Puffin with one removed. Both should assert the deleted rows stay out of the DELETE. The DV side of `ScanTask()` isn't read anywhere in this PR yet. ########## table/changelog_reader_external_test.go: ########## @@ -0,0 +1,344 @@ +// 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_test + +import ( + "context" + "fmt" + "iter" + "os" + "testing" + + "github.com/apache/arrow-go/v18/arrow" + "github.com/apache/arrow-go/v18/arrow/array" + "github.com/apache/arrow-go/v18/arrow/compute" + "github.com/apache/arrow-go/v18/arrow/memory" + "github.com/apache/iceberg-go" + "github.com/apache/iceberg-go/table" + "github.com/stretchr/testify/require" +) + +func TestIncrementalChangelogReadEmptyTable(t *testing.T) { + schema, records, err := newV2RowLineageTestTable(t).NewIncrementalChangelogScan().ToArrowRecords(t.Context()) + require.NoError(t, err) + require.Equal(t, []string{ + "id", "data", + table.ChangelogChangeTypeColumn, + table.ChangelogChangeOrdinalColumn, + table.ChangelogCommitSnapshotIDColumn, + }, schemaFieldNames(schema)) + + require.Empty(t, collectChangelogRows(t, records)) +} + +func TestIncrementalChangelogReadFileChanges(t *testing.T) { + ctx := t.Context() + tbl := changelogOverwriteTable(t, ctx) + + tasks, err := tbl.NewIncrementalChangelogScan().PlanFiles(ctx) + require.NoError(t, err) + require.Len(t, tasks, 3) + + schema, records, err := tbl.NewIncrementalChangelogScan().Read(ctx, tasks) + require.NoError(t, err) + require.Equal(t, []string{ + "id", "data", + table.ChangelogChangeTypeColumn, + table.ChangelogChangeOrdinalColumn, + table.ChangelogCommitSnapshotIDColumn, + }, schemaFieldNames(schema)) + + got := collectChangelogRows(t, records) + require.Equal(t, []changelogRow{ + {id: 1, data: "a", op: string(table.ChangelogOpInsert), ordinal: int64(tasks[0].ChangeOrdinal()), snapshot: tasks[0].CommitSnapshotID()}, + {id: 1, data: "a", op: string(table.ChangelogOpDelete), ordinal: int64(tasks[1].ChangeOrdinal()), snapshot: tasks[1].CommitSnapshotID()}, + {id: 2, data: "b", op: string(table.ChangelogOpInsert), ordinal: int64(tasks[2].ChangeOrdinal()), snapshot: tasks[2].CommitSnapshotID()}, + }, got) + require.NotEqual(t, tasks[0].CommitSnapshotID(), tasks[1].CommitSnapshotID()) + require.Equal(t, tasks[1].CommitSnapshotID(), tasks[2].CommitSnapshotID()) + require.Equal(t, 0, tasks[0].ChangeOrdinal()) + require.Equal(t, 1, tasks[1].ChangeOrdinal()) + require.Equal(t, 1, tasks[2].ChangeOrdinal()) +} + +func TestIncrementalChangelogReadHonorsProjectionFilterAndLimit(t *testing.T) { + ctx := t.Context() + tbl := changelogOverwriteTable(t, ctx) + + schema, records, err := tbl.NewIncrementalChangelogScan( + table.WithSelectedFields("id"), + table.WithRowFilter(iceberg.EqualTo(iceberg.Reference("id"), int64(2))), + table.WithLimit(1), + ).ToArrowRecords(ctx) + require.NoError(t, err) + require.Equal(t, []string{ + "id", + table.ChangelogChangeTypeColumn, + table.ChangelogChangeOrdinalColumn, + table.ChangelogCommitSnapshotIDColumn, + }, schemaFieldNames(schema)) + + got := collectChangelogRows(t, records) + require.Equal(t, []changelogRow{ + {id: 2, op: string(table.ChangelogOpInsert), ordinal: 1}, + }, clearChangelogSnapshot(got)) +} + +func TestIncrementalChangelogReadLimitCutsABatch(t *testing.T) { + ctx := t.Context() + tbl := newV3RowLineageTestTable(t) + 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"}]`, + }) + require.NoError(t, err) + t.Cleanup(data.Release) + tbl, err = tbl.Append(ctx, array.NewTableReader(data, -1), nil) + require.NoError(t, err) + + _, records, err := tbl.NewIncrementalChangelogScan( + table.WithSelectedFields("id"), + table.WithLimit(2), + ).ToArrowRecords(ctx) + require.NoError(t, err) + got := collectChangelogRows(t, records) + require.Equal(t, []changelogRow{ + {id: 1, op: string(table.ChangelogOpInsert), ordinal: 0}, + {id: 2, op: string(table.ChangelogOpInsert), ordinal: 0}, + }, clearChangelogSnapshot(got)) +} + +func TestIncrementalChangelogReadReleasesCheckedAllocator(t *testing.T) { + tbl := newV3RowLineageTestTable(t) + 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"}]`, + }) + require.NoError(t, err) + t.Cleanup(data.Release) + tbl, err = tbl.Append(t.Context(), array.NewTableReader(data, -1), nil) + require.NoError(t, err) + + read := func(t *testing.T, stopEarly bool) { + t.Helper() + + mem := memory.NewCheckedAllocator(memory.DefaultAllocator) + ctx := compute.WithAllocator(t.Context(), mem) + defer mem.AssertSize(t, 0) + + _, records, err := tbl.NewIncrementalChangelogScan( + table.WithSelectedFields("id"), + ).ToArrowRecords(ctx) + require.NoError(t, err) + for rec, err := range records { + require.NoError(t, err) + rec.Release() + if stopEarly { Review Comment: The fixture is a single append, so there's one task and one batch, and breaking after it runs the same path as a full drain. I'd build it from two appends, or reuse `changelogOverwriteTable`, so the break leaves a task unread. -- 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]
