This is an automated email from the ASF dual-hosted git repository.
laskoviymishka pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg-go.git
The following commit(s) were added to refs/heads/main by this push:
new a4d6a0949 feat(table): read file-level incremental changelog rows
(#2170)
a4d6a0949 is described below
commit a4d6a0949a3ddc1f1b6f2cc7bec5c479cedb012e
Author: Digvijay <[email protected]>
AuthorDate: Fri Oct 30 17:29:44 2026 -0500
feat(table): read file-level incremental changelog rows (#2170)
---
table/changelog_reader.go | 323 +++++++++++++++++++++
table/changelog_reader_external_test.go | 498 ++++++++++++++++++++++++++++++++
table/changelog_reader_test.go | 88 ++++++
3 files changed, 909 insertions(+)
diff --git a/table/changelog_reader.go b/table/changelog_reader.go
new file mode 100644
index 000000000..79f399096
--- /dev/null
+++ b/table/changelog_reader.go
@@ -0,0 +1,323 @@
+// 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/arrow-go/v18/arrow/scalar"
+ "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. Files are read one at a time, in task order, so the rows
+// stay in that order. WithMaxConcurrency does not read changelog files in
+// parallel. 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 {
+ // Safe because tasks are read sequentially on
one clone.
+ 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)
+ defer source.Release()
+ rows = source.NumRows()
+ }
+ if rows == 0 {
+ return nil, 0, nil
+ }
+
+ stamped, err := appendChangelogColumns(ctx, schema, useLargeTypes,
source, task)
+ if err != nil {
+ return nil, 0, err
+ }
+
+ return stamped, rows, nil
+}
+
+func appendChangelogColumns(ctx context.Context, schema *arrow.Schema,
useLargeTypes bool, rec arrow.RecordBatch, task ChangelogScanTask)
(arrow.RecordBatch, error) {
+ if err := changelogBatchMatches(schema, rec.Schema()); err != nil {
+ return nil, err
+ }
+
+ mem := allocatorFrom(ctx)
+ n := int(rec.NumRows())
+ var changeScalar scalar.Scalar =
scalar.NewStringScalar(string(task.Operation()))
+ if useLargeTypes {
+ changeScalar =
scalar.NewLargeStringScalar(string(task.Operation()))
+ }
+ changeType, err := scalar.MakeArrayFromScalar(changeScalar, n, mem)
+ if err != nil {
+ return nil, err
+ }
+ defer changeType.Release()
+
+ ordinal, err :=
scalar.MakeArrayFromScalar(scalar.NewInt32Scalar(int32(task.ChangeOrdinal())),
n, mem)
+ if err != nil {
+ return nil, err
+ }
+ defer ordinal.Release()
+ snapshotID, err :=
scalar.MakeArrayFromScalar(scalar.NewInt64Scalar(task.CommitSnapshotID()), n,
mem)
+ if err != nil {
+ return nil, err
+ }
+ defer snapshotID.Release()
+
+ cols := make([]arrow.Array, 0, rec.NumCols()+3)
+ cols = append(cols, rec.Columns()...)
+ cols = append(cols, changeType, ordinal, snapshotID)
+
+ return array.NewRecordBatch(schema, cols, rec.NumRows()), nil
+}
+
+func allocatorFrom(ctx context.Context) memory.Allocator {
+ if mem := compute.GetAllocator(ctx); mem != nil {
+ return mem
+ }
+
+ return memory.DefaultAllocator
+}
diff --git a/table/changelog_reader_external_test.go
b/table/changelog_reader_external_test.go
new file mode 100644
index 000000000..7c46f26e9
--- /dev/null
+++ b/table/changelog_reader_external_test.go
@@ -0,0 +1,498 @@
+// 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"
+ "path/filepath"
+ "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"
+ iceio "github.com/apache/iceberg-go/io"
+ "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)
+ for _, rows := range []string{
+ `[{"id": 1, "data": "a"}]`,
+ `[{"id": 2, "data": "b"}, {"id": 3, "data": "c"}]`,
+ } {
+ data, err := array.TableFromJSON(memory.DefaultAllocator,
arrowSchema, []string{rows})
+ 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))
+ require.Equal(t, []int64{4}, changelogIDs(got, table.ChangelogOpInsert,
1))
+}
+
+func TestIncrementalChangelogReadListColumn(t *testing.T) {
+ ctx := t.Context()
+ 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: "tags", Type:
&iceberg.ListType{
+ ElementID: 3, Element: iceberg.PrimitiveTypes.String,
+ }},
+ )
+ meta, err := table.NewMetadata(schema, iceberg.UnpartitionedSpec,
table.UnsortedSortOrder, location,
+ iceberg.Properties{table.PropertyFormatVersion: "2"})
+ require.NoError(t, err)
+ metaLoc := location + "/metadata/v1.metadata.json"
+ fsF := func(context.Context) (iceio.IO, error) { return
iceio.LocalFS{}, nil }
+ tbl := table.New(table.Identifier{"db", "changelog_list"}, meta,
metaLoc, fsF,
+ &concurrentTestCatalog{metadata: meta, location: metaLoc, fsF:
fsF})
+
+ arrowSchema, err := table.SchemaToArrowSchema(schema, nil, false, false)
+ require.NoError(t, err)
+ data, err := array.TableFromJSON(memory.DefaultAllocator, arrowSchema,
[]string{
+ `[{"id": 1, "tags": ["red", "blue"]}]`,
+ })
+ require.NoError(t, err)
+ t.Cleanup(data.Release)
+ tbl, err = tbl.Append(ctx, array.NewTableReader(data, -1), nil)
+ require.NoError(t, err)
+
+ out, records, err :=
tbl.NewIncrementalChangelogScan().ToArrowRecords(ctx)
+ require.NoError(t, err)
+ tags, ok :=
out.Field(out.FieldIndices("tags")[0]).Type.(*arrow.ListType)
+ require.True(t, ok)
+ require.Equal(t, arrow.STRING, tags.Elem().ID())
+
+ var seen int
+ for rec, err := range records {
+ require.NoError(t, err)
+ list := rec.Column(out.FieldIndices("tags")[0]).(*array.List)
+ values := list.ListValues().(*array.String)
+ require.Equal(t, []string{"red", "blue"},
[]string{values.Value(0), values.Value(1)})
+ seen += int(rec.NumRows())
+ rec.Release()
+ }
+ require.Equal(t, 1, seen)
+}
+
+func TestIncrementalChangelogReadLargeTypes(t *testing.T) {
+ ctx := t.Context()
+ tbl := newV2RowLineageTestTable(t)
+ arrowSchema, err :=
table.SchemaToArrowSchema(tbl.Metadata().CurrentSchema(), nil, false, false)
+ require.NoError(t, err)
+ data, err := array.TableFromJSON(memory.DefaultAllocator, arrowSchema,
[]string{
+ `[{"id": 1, "data": "a"}]`,
+ })
+ require.NoError(t, err)
+ t.Cleanup(data.Release)
+ tbl, err = tbl.Append(ctx, array.NewTableReader(data, -1), nil)
+ require.NoError(t, err)
+
+ out, records, err :=
tbl.NewIncrementalChangelogScan(table.WithOptions(iceberg.Properties{
+ table.ScanOptionArrowUseLargeTypes: "true",
+ })).ToArrowRecords(ctx)
+ require.NoError(t, err)
+ require.Equal(t, arrow.LARGE_STRING,
out.Field(out.FieldIndices("data")[0]).Type.ID())
+ require.Equal(t, arrow.LARGE_STRING,
out.Field(out.FieldIndices(table.ChangelogChangeTypeColumn)[0]).Type.ID())
+
+ for rec, err := range records {
+ require.NoError(t, err)
+ require.Equal(t, "a",
rec.Column(out.FieldIndices("data")[0]).(*array.LargeString).Value(0))
+ require.Equal(t, string(table.ChangelogOpInsert),
rec.Column(out.FieldIndices(table.ChangelogChangeTypeColumn)[0]).(*array.LargeString).Value(0))
+ rec.Release()
+ }
+}
+
+func TestIncrementalChangelogReadPreRangeDeleteStaysOutOfFileDelete(t
*testing.T) {
+ ctx := t.Context()
+ tbl := newV2RowLineageTestTable(t)
+ arrowSchema, err :=
table.SchemaToArrowSchema(tbl.Metadata().CurrentSchema(), nil, false, false)
+ require.NoError(t, err)
+ 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)
+
+ files, err := tbl.Scan().PlanFiles(ctx)
+ require.NoError(t, err)
+ require.Len(t, files, 1)
+ dataPath := files[0].File.FilePath()
+ 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).AddDeletes(changelogPosDeleteFile(t, posPath)).Commit(ctx))
+ tbl, err = tx.Commit(ctx)
+ require.NoError(t, err)
+ deleted := tbl.CurrentSnapshot().SnapshotID
+
+ replacement, err := array.TableFromJSON(memory.DefaultAllocator,
arrowSchema, []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)
+
+ got := collectChangelogRows(t, mustReadChangelog(t, ctx,
tbl.NewIncrementalChangelogScan().FromSnapshotExclusive(deleted)))
+ require.Equal(t, []int64{4}, changelogIDs(got, table.ChangelogOpInsert,
0))
+ require.Equal(t, []int64{1, 3}, changelogIDs(got,
table.ChangelogOpDelete, 0))
+}
+
+func TestIncrementalChangelogReadPreRangeSharedDeletionVectorStaysOut(t
*testing.T) {
+ ctx := t.Context()
+ tbl, location, pathA, _, dvA, _ := newTableWithSharedPuffinDVs(t)
+ // Replace one deletion vector in the shared Puffin. The sibling stays.
+ replacementDV := writeDV(t, location, "dv-a2.puffin", pathA, []int64{1})
+ tx := tbl.NewTransaction()
+ require.NoError(t,
tx.NewRowDelta(nil).AddDeletes(replacementDV).RemoveDeletes(dvA).Commit(ctx))
+ tbl, err := tx.Commit(ctx)
+ require.NoError(t, err)
+ removed := tbl.CurrentSnapshot().SnapshotID
+
+ arrowSchema, err :=
table.SchemaToArrowSchema(tbl.Metadata().CurrentSchema(), nil, false, false)
+ require.NoError(t, err)
+ replacement, err := array.TableFromJSON(memory.DefaultAllocator,
arrowSchema, []string{
+ `[{"id": 5, "data": "e"}]`,
+ })
+ 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)
+
+ got := collectChangelogRows(t, mustReadChangelog(t, ctx,
tbl.NewIncrementalChangelogScan().FromSnapshotExclusive(removed)))
+ require.Equal(t, []int64{5}, changelogIDs(got, table.ChangelogOpInsert,
0))
+ // id 2 is hidden by the replacement and id 3 by the sibling that
stayed in
+ // the shared Puffin. Both were committed before the range.
+ require.Equal(t, []int64{1, 4}, changelogIDs(got,
table.ChangelogOpDelete, 0))
+}
+
+func mustReadChangelog(t *testing.T, ctx context.Context, scan
*table.IncrementalChangelogScan) iter.Seq2[arrow.RecordBatch, error] {
+ t.Helper()
+
+ _, records, err := scan.ToArrowRecords(ctx)
+ require.NoError(t, err)
+
+ return records
+}
+
+func changelogOverwriteTable(t *testing.T, ctx context.Context) *table.Table {
+ t.Helper()
+
+ mem := memory.DefaultAllocator
+ 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)
+
+ initial, err := array.TableFromJSON(mem, arrowSchema, []string{
+ `[{"id": 1, "data": "a"}]`,
+ })
+ require.NoError(t, err)
+ t.Cleanup(initial.Release)
+ tbl, err = tbl.Append(ctx, array.NewTableReader(initial, -1), nil)
+ require.NoError(t, err)
+
+ replacement, err := array.TableFromJSON(mem, arrowSchema, []string{
+ `[{"id": 2, "data": "b"}]`,
+ })
+ 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)
+
+ return tbl
+}
+
+type changelogRow struct {
+ id int64
+ data string
+ op string
+ ordinal int64
+ snapshot int64
+}
+
+func collectChangelogRows(t *testing.T, records iter.Seq2[arrow.RecordBatch,
error]) []changelogRow {
+ t.Helper()
+
+ var got []changelogRow
+ for rec, err := range records {
+ require.NoError(t, err)
+
+ got = append(got, changelogRowsFromBatch(rec)...)
+ rec.Release()
+ }
+
+ return got
+}
+
+func changelogRowsFromBatch(rec arrow.RecordBatch) []changelogRow {
+ ids := rec.Column(0).(*array.Int64)
+ opIdx := 1
+ var data *array.String
+ if rec.NumCols() == 5 {
+ data = rec.Column(1).(*array.String)
+ opIdx = 2
+ }
+ ops := rec.Column(opIdx).(*array.String)
+ ordinals := rec.Column(opIdx + 1).(*array.Int32)
+ snapshots := rec.Column(opIdx + 2).(*array.Int64)
+
+ rows := make([]changelogRow, 0, rec.NumRows())
+ for i := range int(rec.NumRows()) {
+ row := changelogRow{
+ id: ids.Value(i),
+ op: ops.Value(i),
+ ordinal: int64(ordinals.Value(i)),
+ snapshot: snapshots.Value(i),
+ }
+ if data != nil && !data.IsNull(i) {
+ row.data = data.Value(i)
+ }
+ rows = append(rows, row)
+ }
+
+ return rows
+}
+
+func schemaFieldNames(schema *arrow.Schema) []string {
+ names := make([]string, schema.NumFields())
+ for i := range names {
+ names[i] = schema.Field(i).Name
+ }
+
+ return names
+}
+
+func clearChangelogSnapshot(rows []changelogRow) []changelogRow {
+ out := make([]changelogRow, len(rows))
+ for i, row := range rows {
+ row.snapshot = 0
+ out[i] = row
+ }
+
+ return out
+}
+
+func changelogIDs(rows []changelogRow, op table.ChangelogOperation, ordinal
int64) []int64 {
+ var ids []int64
+ for _, row := range rows {
+ if row.op == string(op) && row.ordinal == ordinal {
+ ids = append(ids, row.id)
+ }
+ }
+
+ return ids
+}
+
+func changelogDataFile(t *testing.T, path string, records int64)
iceberg.DataFile {
+ t.Helper()
+
+ info, err := os.Stat(path)
+ require.NoError(t, err)
+ b, err := iceberg.NewDataFileBuilder(
+ *iceberg.UnpartitionedSpec, iceberg.EntryContentData,
+ path, iceberg.ParquetFile, nil, nil, nil, records, info.Size())
+ require.NoError(t, err)
+
+ return b.Build()
+}
+
+func changelogPosDeleteFile(t *testing.T, path string) iceberg.DataFile {
+ t.Helper()
+
+ info, err := os.Stat(path)
+ require.NoError(t, err)
+ b, err := iceberg.NewDataFileBuilder(
+ *iceberg.UnpartitionedSpec, iceberg.EntryContentPosDeletes,
+ path, iceberg.ParquetFile, nil, nil, nil, 1, info.Size())
+ require.NoError(t, err)
+
+ return b.Build()
+}
diff --git a/table/changelog_reader_test.go b/table/changelog_reader_test.go
new file mode 100644
index 000000000..4f3545434
--- /dev/null
+++ b/table/changelog_reader_test.go
@@ -0,0 +1,88 @@
+// 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"
+ "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/iceberg-go"
+ "github.com/stretchr/testify/require"
+)
+
+func TestChangelogReadRejectsNilScan(t *testing.T) {
+ var scan *IncrementalChangelogScan
+ _, _, err := scan.Read(context.Background(), nil)
+ require.ErrorIs(t, err, ErrInvalidOperation)
+}
+
+func TestChangelogReadRejectsDeletedRowTasks(t *testing.T) {
+ data := changelogTestDataFile(t, "data/f2.parquet",
iceberg.EntryContentData, iceberg.ParquetFile)
+ added := changelogTestDataFile(t, "deletes/d2.parquet",
iceberg.EntryContentPosDeletes, iceberg.ParquetFile)
+ task, err := NewDeletedRowsScanTask(data, []iceberg.DataFile{added},
nil, 1, 44)
+ require.NoError(t, err)
+
+ scan := &IncrementalChangelogScan{scan: &Scan{limit: ScanNoLimit}}
+ _, _, err = scan.Read(context.Background(), []ChangelogScanTask{task})
+ require.ErrorIs(t, err, iceberg.ErrNotImplemented)
+}
+
+func TestStampChangelogBatchCutsToTheRemainingBudget(t *testing.T) {
+ builder := array.NewInt64Builder(memory.DefaultAllocator)
+ builder.AppendValues([]int64{1, 2, 3}, nil)
+ values := builder.NewArray()
+ defer values.Release()
+ dataSchema := arrow.NewSchema([]arrow.Field{
+ {Name: "id", Type: arrow.PrimitiveTypes.Int64, Nullable: false},
+ }, nil)
+ rec := array.NewRecordBatch(dataSchema, []arrow.Array{values},
int64(values.Len()))
+ defer rec.Release()
+
+ out := dataSchema
+ var err error
+ for _, field := range changelogArrowFields(false) {
+ out, err = out.AddField(out.NumFields(), field)
+ require.NoError(t, err)
+ }
+ file := changelogTestDataFile(t, "data.parquet",
iceberg.EntryContentData, iceberg.ParquetFile)
+ task, err := NewAddedRowsScanTask(file, nil, 0, 7)
+ require.NoError(t, err)
+
+ stamped, rows, err := stampChangelogBatch(context.Background(), out,
false, rec, task, 2, 0)
+ require.NoError(t, err)
+ require.Equal(t, int64(2), rows)
+ require.Equal(t, int64(2), stamped.NumRows())
+ t.Cleanup(stamped.Release)
+ require.Equal(t, []int64{1, 2},
stamped.Column(0).(*array.Int64).Int64Values())
+}
+
+func TestChangelogArrowSchemaRejectsReservedName(t *testing.T) {
+ schema := iceberg.NewSchema(0, iceberg.NestedField{
+ ID: 1, Name: ChangelogChangeTypeColumn, Type:
iceberg.PrimitiveTypes.Int64, Required: true,
+ })
+ meta, err := NewMetadata(schema, iceberg.UnpartitionedSpec,
UnsortedSortOrder, "mem://changelog-reserved", nil)
+ require.NoError(t, err)
+ tbl := New(Identifier{"db", "changelog"}, meta, "metadata.json", nil,
nil)
+
+ _, _, err = changelogArrowSchema(tbl.NewIncrementalChangelogScan().scan)
+ require.ErrorIs(t, err, iceberg.ErrInvalidSchema)
+ require.ErrorContains(t, err, ChangelogChangeTypeColumn)
+}
