dgvj-work commented on code in PR #2170: URL: https://github.com/apache/iceberg-go/pull/2170#discussion_r4420175845
########## 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: `Read` now says files are read one at a time, in task order, and that `WithMaxConcurrency` does not span changelog files. Order-preserving parallel reads are filed as #2371. ########## 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: There is now a changelog read of a list column, and one with `ScanOptionArrowUseLargeTypes`. Both go through the schema check, including the nested list field and the large string type on `_change_type`. ########## 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: `ReadTasks` does truncate to the remaining budget, so I left the slice in place and tested `stampChangelogBatch` directly with a 3-row batch and a budget of 2. ########## 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 checked-allocator fixture is two appends now, so the early break leaves the second 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]
