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 8ef0a092f feat(table): read rows removed by incremental changelog
deletes (#2171)
8ef0a092f is described below
commit 8ef0a092fcbaf70f99aa8935481bc95071eb41b1
Author: Digvijay <[email protected]>
AuthorDate: Tue Nov 3 08:05:34 2026 -0600
feat(table): read rows removed by incremental changelog deletes (#2171)
---
table/arrow_scanner.go | 52 ++-
table/changelog_deleted_rows.go | 321 ++++++++++++++++++
table/changelog_reader.go | 26 +-
table/changelog_reader_external_test.go | 564 ++++++++++++++++++++++++++++++--
table/changelog_reader_test.go | 6 +-
table/scanner.go | 10 +
website/src/api.md | 20 ++
7 files changed, 960 insertions(+), 39 deletions(-)
diff --git a/table/arrow_scanner.go b/table/arrow_scanner.go
index ad7e68105..f967d1683 100644
--- a/table/arrow_scanner.go
+++ b/table/arrow_scanner.go
@@ -1695,6 +1695,10 @@ type arrowScan struct {
// arrowBatchSize, when positive, overrides the table's
// read.parquet.batch-size property for this scan's reads.
arrowBatchSize int
+
+ // added is nil for a normal scan. A changelog deleted-row read sets it
so
+ // rows matched by that snapshot's new deletes are kept.
+ added *addedDeleteState
}
// preparedFileRead contains the physical schema projection shared by all
@@ -2750,8 +2754,17 @@ func (as *arrowScan) recordsFromTask(ctx
context.Context, task tblutils.Enumerat
// row-dropping position step in the pipeline so cursors never run over
a
// sequence an earlier step already shortened.
applyPosDeletes := positionalDeleteBitmap != nil &&
!positionalDeleteBitmap.IsEmpty() && !hasDV
+ addedPos, addedBitmap, addedEq, err := as.loadAddedDeletes(ctx,
task.Value)
+ if err != nil {
+ return err
+ }
+ addedHasDV := addedBitmap != nil && !addedBitmap.IsEmpty()
+ // An added deletion vector supersedes added position deletes, matching
the
+ // rule used for the deletes already on the task.
+ addedHasPos := addedPos != nil && !addedPos.IsEmpty() && !addedHasDV
+ stampPos := addedHasPos || addedHasDV
var posSource *rowPositionSource
- if synthesizeRowID || applyPosDeletes || hasDV {
+ if synthesizeRowID || applyPosDeletes || hasDV || stampPos {
posSource = &rowPositionSource{}
}
@@ -2772,6 +2785,10 @@ func (as *arrowScan) recordsFromTask(ctx
context.Context, task tblutils.Enumerat
})
}
+ if stampPos {
+ pipeline = append(pipeline, stampChangelogFilePos(ctx,
posSource.cursor()))
+ }
+
if applyPosDeletes {
pipeline = append(pipeline,
processPositionalDeleteBitmap(ctx,
positionalDeleteBitmap, posSource.cursor()))
@@ -2790,6 +2807,13 @@ func (as *arrowScan) recordsFromTask(ctx
context.Context, task tblutils.Enumerat
pipeline = append(pipeline, eqFn)
}
+ if as.added != nil {
+ pipeline = append(pipeline, keepAddedDeleteRows(ctx, addedPos,
addedHasPos, addedBitmap, task.Value.File.Count(), addedEq, iceSchema,
task.Value.File.FilePath(), stampPos))
+ if stampPos {
+ pipeline = append(pipeline, dropChangelogFilePos)
+ }
+ }
+
if filterPlans != nil {
filterFunc, dropFile, err = as.getRecordFilterFromPlan(ctx,
filterPlans.record)
} else {
@@ -3046,9 +3070,13 @@ func newScanTaskChannel(taskCount, numWorkers int) chan
tblutils.Enumerated[File
}
func (as *arrowScan) recordBatchesFromTasksAndDeletes(ctx context.Context,
tasks []FileScanTask, positionDeleteLoader *lazyPositionDeleteLoader, dvLoader
*lazyDeletionVectorLoader, equalityDeleteLoader *lazyEqualityDeleteLoader,
invariants *arrowScanInvariants) iter.Seq2[arrow.RecordBatch, error] {
- var cleanup func()
- if positionDeleteLoader != nil {
- cleanup = positionDeleteLoader.release
+ cleanup := func() {
+ if positionDeleteLoader != nil {
+ positionDeleteLoader.release()
+ }
+ if as.added != nil && as.added.posLoader != nil {
+ as.added.posLoader.release()
+ }
}
return as.recordBatchesWithCleanup(ctx, tasks, 0, positionDeleteLoader,
dvLoader, equalityDeleteLoader, invariants, cleanup)
@@ -3176,6 +3204,11 @@ func (as *arrowScan) GetRecords(ctx context.Context,
tasks []FileScanTask) (*arr
// read concurrently and in any order. release frees the shared position
// deletes and must be called once, after the last stream is done.
func (as *arrowScan) GetRecordsPerTask(ctx context.Context, tasks
[]FileScanTask) (*arrow.Schema, []iter.Seq2[arrow.RecordBatch, error], func(),
error) {
+ // Added-delete reads go through GetRecords only. This path does not
+ // release the added position-delete loader.
+ if as.added != nil {
+ return nil, nil, nil, fmt.Errorf("%w: added-delete reads go
through GetRecords only", ErrInvalidOperation)
+ }
resultSchema, reads, err := as.prepareTaskReads(ctx, tasks)
if err != nil {
return nil, nil, nil, err
@@ -3277,10 +3310,19 @@ func (as *arrowScan) prepareTaskReads(ctx
context.Context, tasks []FileScanTask)
}
equalityDeleteLoader.addFieldIDs(invariants.projectedIDs)
+ positionDeleteLoader := newLazyPositionDeleteLoader(as.fs, tasks)
+ if as.added != nil {
+ if err := as.prepareAddedDeleteLoaders(invariants); err != nil {
+ positionDeleteLoader.release()
+
+ return nil, nil, err
+ }
+ }
+
return resultSchema, &taskReads{
ctx: ctx,
invariants: invariants,
- positionDeletes: newLazyPositionDeleteLoader(as.fs, tasks),
+ positionDeletes: positionDeleteLoader,
deletionVectors: dvLoader,
equalityDeletes: equalityDeleteLoader,
}, nil
diff --git a/table/changelog_deleted_rows.go b/table/changelog_deleted_rows.go
new file mode 100644
index 000000000..b63a982f6
--- /dev/null
+++ b/table/changelog_deleted_rows.go
@@ -0,0 +1,321 @@
+// 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 (
+ "bytes"
+ "context"
+ "fmt"
+ "iter"
+
+ "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/iceberg-go"
+ "github.com/apache/iceberg-go/table/dv"
+)
+
+// changelogFilePosColumn carries the data file position of each row so a
+// later keep step can still see it after existing deletes shorten the batch.
+const changelogFilePosColumn = "_changelog_file_pos"
+
+// addedDeleteState is the delete files added in one changelog snapshot and
+// the loaders that read them. A normal scan leaves arrowScan.added nil.
+type addedDeleteState struct {
+ deletes *FileScanTask
+ posLoader *lazyPositionDeleteLoader
+ dvLoader *lazyDeletionVectorLoader
+ eqLoader *lazyEqualityDeleteLoader
+}
+
+func (scan *Scan) readDeletedRows(ctx context.Context, task
DeletedRowsScanTask) (*arrow.Schema, iter.Seq2[arrow.RecordBatch, error],
error) {
+ base := task.ScanTask()
+ added := base
+ added.DeleteFiles = task.addedDeletes.pos
+ added.EqualityDeleteFiles = task.addedDeletes.eq
+ added.DeletionVectorFiles = task.addedDeletes.dv
+
+ return scan.readFileTasks(ctx, []FileScanTask{base}, &added)
+}
+
+func (as *arrowScan) prepareAddedDeleteLoaders(invariants
*arrowScanInvariants) error {
+ tasks := []FileScanTask{*as.added.deletes}
+ posLoader := newLazyPositionDeleteLoader(as.fs, tasks)
+ releasePos := func() {
+ if posLoader != nil {
+ posLoader.release()
+ }
+ }
+
+ dvLoader, err := newLazyDeletionVectorLoader(as.fs, tasks)
+ if err != nil {
+ releasePos()
+
+ return err
+ }
+
+ eqLoader, err := newLazyEqualityDeleteLoader(
+ as.fs, invariants.tableSchema, nil, invariants.nameMapping,
tasks)
+ if err != nil {
+ releasePos()
+
+ return err
+ }
+ if eqLoader.needsSchemaHistory() {
+ eqLoader.tableSchemas = as.metadata.Schemas()
+ }
+ eqLoader.addFieldIDs(invariants.projectedIDs)
+
+ as.added.posLoader = posLoader
+ as.added.dvLoader = dvLoader
+ as.added.eqLoader = eqLoader
+
+ return nil
+}
+
+func (as *arrowScan) loadAddedDeletes(ctx context.Context, task FileScanTask)
(*dv.RoaringPositionBitmap, *dv.RoaringPositionBitmap, []*equalityDeleteSet,
error) {
+ if as.added == nil {
+ return nil, nil, nil, nil
+ }
+
+ dataPath := ""
+ if task.File != nil {
+ dataPath = task.File.FilePath()
+ }
+ if as.added.deletes.File == nil || as.added.deletes.File.FilePath() !=
dataPath {
+ return nil, nil, nil, fmt.Errorf("%w: added delete files do not
match data file %s",
+ ErrInvalidOperation, dataPath)
+ }
+
+ var pos *dv.RoaringPositionBitmap
+ var err error
+ if as.added.posLoader != nil {
+ pos, err = as.added.posLoader.load(ctx, 0, dataPath)
+ if err != nil {
+ return nil, nil, nil, err
+ }
+ }
+ bitmap, err := as.added.dvLoader.load(ctx, dataPath)
+ if err != nil {
+ return nil, nil, nil, err
+ }
+ eq, err := as.added.eqLoader.load(ctx, *as.added.deletes)
+ if err != nil {
+ return nil, nil, nil, err
+ }
+
+ return pos, bitmap, eq, nil
+}
+
+func stampChangelogFilePos(ctx context.Context, cursor *rowPositionCursor)
recProcessFn {
+ return func(in arrow.RecordBatch) (arrow.RecordBatch, error) {
+ defer in.Release()
+
+ mem := allocatorFrom(ctx)
+ bldr := array.NewInt64Builder(mem)
+ defer bldr.Release()
+ n := int(in.NumRows())
+ bldr.Reserve(n)
+ for range n {
+ bldr.Append(cursor.next())
+ }
+ pos := bldr.NewArray()
+ defer pos.Release()
+
+ schema, err := in.Schema().AddField(int(in.NumCols()),
arrow.Field{
+ Name: changelogFilePosColumn,
+ Type: arrow.PrimitiveTypes.Int64,
+ Nullable: false,
+ })
+ if err != nil {
+ return nil, err
+ }
+
+ cols := make([]arrow.Array, in.NumCols()+1)
+ copy(cols, in.Columns())
+ cols[in.NumCols()] = pos
+
+ return array.NewRecordBatch(schema, cols, in.NumRows()), nil
+ }
+}
+
+func dropChangelogFilePos(in arrow.RecordBatch) (arrow.RecordBatch, error) {
+ defer in.Release()
+
+ idx := in.Schema().FieldIndices(changelogFilePosColumn)
+ if len(idx) != 1 {
+ return nil, fmt.Errorf("%w: missing changelog file position
column %s",
+ iceberg.ErrInvalidSchema, changelogFilePosColumn)
+ }
+ drop := idx[0]
+ fields := make([]arrow.Field, 0, in.NumCols()-1)
+ cols := make([]arrow.Array, 0, in.NumCols()-1)
+ for i, field := range in.Schema().Fields() {
+ if i == drop {
+ continue
+ }
+ fields = append(fields, field)
+ cols = append(cols, in.Column(i))
+ }
+
+ md := in.Schema().Metadata()
+
+ return array.NewRecordBatch(arrow.NewSchema(fields, &md), cols,
in.NumRows()), nil
+}
+
+// keepAddedDeleteRows keeps rows removed by the added deletes. Existing
+// deletes have already dropped rows that were removed earlier. A deletion
+// vector supersedes added position deletes. Equality deletes are combined
+// with that position check: a row is kept when either kind matches.
+func keepAddedDeleteRows(
+ ctx context.Context,
+ positions *dv.RoaringPositionBitmap,
+ hasPos bool,
+ bitmap *dv.RoaringPositionBitmap,
+ rowCount int64,
+ eqSets []*equalityDeleteSet,
+ fileSchema *iceberg.Schema,
+ dataFilePath string,
+ stampPos bool,
+) recProcessFn {
+ var deletedByVector *dv.RoaringPositionBitmap
+ if bitmap != nil && !bitmap.IsEmpty() {
+ deletedByVector = bitmap
+ hasPos = false
+ }
+
+ return func(r arrow.RecordBatch) (arrow.RecordBatch, error) {
+ defer r.Release()
+
+ n := int(r.NumRows())
+ if n == 0 {
+ r.Retain()
+
+ return r, nil
+ }
+
+ var posCol *array.Int64
+ if stampPos {
+ indices :=
r.Schema().FieldIndices(changelogFilePosColumn)
+ if len(indices) != 1 {
+ return nil, fmt.Errorf("%w: missing changelog
file position column %s",
+ iceberg.ErrInvalidSchema,
changelogFilePosColumn)
+ }
+ col, ok := r.Column(indices[0]).(*array.Int64)
+ if !ok {
+ return nil, fmt.Errorf("%w: changelog file
position column is %s",
+ iceberg.ErrInvalidSchema,
r.Column(indices[0]).DataType())
+ }
+ posCol = col
+ }
+
+ keep := make([]bool, n)
+ switch {
+ case deletedByVector != nil:
+ for i := range n {
+ pos := posCol.Value(i)
+ keep[i] =
!deletionVectorKeepsRow(deletedByVector, rowCount, pos)
+ }
+ case hasPos:
+ for i := range n {
+ pos := posCol.Value(i)
+ keep[i] = pos >= 0 &&
positions.Contains(uint64(pos))
+ }
+ }
+
+ if len(eqSets) > 0 {
+ matched, err := equalityRowsMatched(r, eqSets,
fileSchema, dataFilePath)
+ if err != nil {
+ return nil, err
+ }
+ for i := range n {
+ keep[i] = keep[i] || matched[i]
+ }
+ }
+
+ mem := allocatorFrom(ctx)
+ bldr := array.NewBooleanBuilder(mem)
+ defer bldr.Release()
+ bldr.Reserve(n)
+ bldr.AppendValues(keep, nil)
+ mask := bldr.NewBooleanArray()
+ defer mask.Release()
+
+ return compute.FilterRecordBatch(ctx, r, mask,
compute.DefaultFilterOptions())
+ }
+}
+
+// equalityRowsMatched reports rows that match an added equality delete. Keys
use
+// the same encoding as the scan filter that drops equality-deleted rows.
+func equalityRowsMatched(r arrow.RecordBatch, eqDeleteSets
[]*equalityDeleteSet, fileSchema *iceberg.Schema, dataFilePath string) ([]bool,
error) {
+ requestedFieldIDs := make([]int, 0)
+ requestedFieldIDSet := make(map[int]struct{})
+ for _, eqDel := range eqDeleteSets {
+ if len(eqDel.fieldIDs) != len(eqDel.colNames) {
+ return nil, fmt.Errorf("%w: equality delete set has %d
field IDs and %d column names",
+ iceberg.ErrInvalidArgument,
len(eqDel.fieldIDs), len(eqDel.colNames))
+ }
+ for _, fieldID := range eqDel.fieldIDs {
+ if _, ok := requestedFieldIDSet[fieldID]; !ok {
+ requestedFieldIDSet[fieldID] = struct{}{}
+ requestedFieldIDs = append(requestedFieldIDs,
fieldID)
+ }
+ }
+ }
+
+ fieldRefsByID := resolveArrowFieldsByID(fileSchema, requestedFieldIDs)
+ fieldRefs := make([][]arrowFieldRef, len(eqDeleteSets))
+ for i, eqDel := range eqDeleteSets {
+ fieldRefs[i] = make([]arrowFieldRef, len(eqDel.fieldIDs))
+ for fieldIdx, fieldID := range eqDel.fieldIDs {
+ ref, err := resolveArrowField(fieldRefsByID, fieldID,
eqDel.colNames[fieldIdx], dataFilePath)
+ if err != nil {
+ return nil, err
+ }
+ fieldRefs[i][fieldIdx] = ref
+ }
+ }
+
+ matched := make([]bool, r.NumRows())
+ var keyBuf bytes.Buffer
+ for setIdx, eqDel := range eqDeleteSets {
+ encoders := make([]colEncoder, len(eqDel.colNames))
+ for i, name := range eqDel.colNames {
+ enc, err := makeArrowFieldEncoder(r,
fieldRefs[setIdx][i], eqDel.fieldIDs[i], name, dataFilePath)
+ if err != nil {
+ return nil, err
+ }
+ encoders[i] = enc
+ }
+
+ for row := range int(r.NumRows()) {
+ if matched[row] {
+ continue
+ }
+ keyBuf.Reset()
+ for _, enc := range encoders {
+ enc(&keyBuf, row)
+ }
+ if _, ok := eqDel.keys[bufString(&keyBuf)]; ok {
+ matched[row] = true
+ }
+ }
+ }
+
+ return matched, nil
+}
diff --git a/table/changelog_reader.go b/table/changelog_reader.go
index 79f399096..307cc7303 100644
--- a/table/changelog_reader.go
+++ b/table/changelog_reader.go
@@ -51,8 +51,8 @@ const (
// 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.
+// task. Row-level deletes emit the rows removed by delete files added in the
+// range, after deletes that already applied have been excluded.
func (s *IncrementalChangelogScan) ToArrowRecords(ctx context.Context)
(*arrow.Schema, iter.Seq2[arrow.RecordBatch, error], error) {
tasks, err := s.PlanFiles(ctx)
if err != nil {
@@ -69,11 +69,11 @@ func (s *IncrementalChangelogScan) ToArrowRecords(ctx
context.Context) (*arrow.S
// 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.
+// are added. DeletedRowsScanTask emits rows matched by AddedDeletes and skips
+// rows matched by ExistingDeletes. 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)
@@ -111,7 +111,13 @@ func (s *IncrementalChangelogScan) Read(ctx
context.Context, tasks []ChangelogSc
readScan.limit = limit - emitted
}
- _, records, err := readScan.ReadTasks(ctx,
[]FileScanTask{task.ScanTask()})
+ var records iter.Seq2[arrow.RecordBatch, error]
+ var err error
+ if deleted, ok := task.(DeletedRowsScanTask); ok {
+ _, records, err = readScan.readDeletedRows(ctx,
deleted)
+ } else {
+ _, records, err = readScan.ReadTasks(ctx,
[]FileScanTask{task.ScanTask()})
+ }
if err != nil {
yield(nil, err)
@@ -149,10 +155,8 @@ func (s *IncrementalChangelogScan) Read(ctx
context.Context, tasks []ChangelogSc
func readableChangelogTask(task ChangelogScanTask) error {
switch task.(type) {
- case AddedRowsScanTask, DeletedDataFileScanTask:
+ case AddedRowsScanTask, DeletedDataFileScanTask, DeletedRowsScanTask:
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)
}
diff --git a/table/changelog_reader_external_test.go
b/table/changelog_reader_external_test.go
index 7c46f26e9..737b2ab73 100644
--- a/table/changelog_reader_external_test.go
+++ b/table/changelog_reader_external_test.go
@@ -23,6 +23,7 @@ import (
"iter"
"os"
"path/filepath"
+ "strconv"
"testing"
"github.com/apache/arrow-go/v18/arrow"
@@ -169,6 +170,86 @@ func
TestIncrementalChangelogReadReleasesCheckedAllocator(t *testing.T) {
t.Run("early break", func(t *testing.T) { read(t, true) })
}
+func TestIncrementalChangelogReadReleasesCheckedAllocatorForDeletes(t
*testing.T) {
+ ctx := t.Context()
+ for _, tt := range []struct {
+ name string
+ version int
+ }{
+ {name: "position delete", version: 2},
+ {name: "deletion vector", version: 3},
+ } {
+ t.Run(tt.name, func(t *testing.T) {
+ tbl := appendChangelogRows(t, ctx,
newChangelogMORTable(t, tt.version))
+ tbl = deleteChangelogID(t, ctx, tbl, 2)
+ arrowSchema := arrow.NewSchema([]arrow.Field{
+ {Name: "id", Type: arrow.PrimitiveTypes.Int64,
Nullable: false},
+ {Name: "data", Type: arrow.BinaryTypes.String,
Nullable: true},
+ }, nil)
+ extra, err :=
array.TableFromJSON(memory.DefaultAllocator, arrowSchema, []string{
+ `[{"id": 4, "data": "d"}]`,
+ })
+ require.NoError(t, err)
+ t.Cleanup(extra.Release)
+ tbl, err = tbl.Append(ctx, array.NewTableReader(extra,
-1), nil)
+ require.NoError(t, err)
+
+ read := func(t *testing.T, stopEarly bool) {
+ t.Helper()
+
+ mem :=
memory.NewCheckedAllocator(memory.DefaultAllocator)
+ readCtx := compute.WithAllocator(ctx, mem)
+ defer mem.AssertSize(t, 0)
+
+ _, records, err :=
tbl.NewIncrementalChangelogScan().ToArrowRecords(readCtx)
+ require.NoError(t, err)
+ var sawDelete bool
+ for rec, err := range records {
+ require.NoError(t, err)
+ ops :=
rec.Column(rec.Schema().FieldIndices(table.ChangelogChangeTypeColumn)[0]).(*array.String)
+ deleteBatch := rec.NumRows() > 0 &&
ops.Value(0) == string(table.ChangelogOpDelete)
+ rec.Release()
+ if deleteBatch {
+ sawDelete = true
+ if stopEarly {
+ break
+ }
+ }
+ }
+ require.True(t, sawDelete)
+ }
+
+ t.Run("full drain", func(t *testing.T) { read(t, false)
})
+ t.Run("early break", func(t *testing.T) { read(t, true)
})
+ })
+ }
+}
+
+func TestIncrementalChangelogReadExpiredParentEmitsOnlyInRangeDelete(t
*testing.T) {
+ ctx := t.Context()
+ for _, version := range []int{2, 3} {
+ t.Run(fmt.Sprintf("v%d", version), func(t *testing.T) {
+ tbl := appendChangelogRows(t, ctx,
newChangelogMORTable(t, version))
+ tbl = deleteChangelogID(t, ctx, tbl, 1)
+ firstDelete := tbl.CurrentSnapshot().SnapshotID
+ tbl = deleteChangelogID(t, ctx, tbl, 2)
+
+ builder, err :=
table.MetadataBuilderFromBase(tbl.Metadata(), tbl.MetadataLocation())
+ require.NoError(t, err)
+ require.NoError(t,
builder.RemoveSnapshots([]int64{firstDelete}, false))
+ meta, err := builder.Build()
+ require.NoError(t, err)
+ tbl = table.New(tbl.Identifier(), meta,
tbl.MetadataLocation(), func(context.Context) (iceio.IO, error) {
+ return iceio.LocalFS{}, nil
+ }, nil)
+
+ got := readChangelog(t, ctx,
tbl.NewIncrementalChangelogScan().FromSnapshotExclusive(firstDelete))
+ require.Empty(t, changelogIDs(got,
table.ChangelogOpInsert))
+ require.Equal(t, []int64{2}, changelogIDs(got,
table.ChangelogOpDelete))
+ })
+ }
+}
+
func TestIncrementalChangelogReadAppliesFileDeletes(t *testing.T) {
ctx := t.Context()
tbl := newV2RowLineageTestTable(t)
@@ -207,9 +288,20 @@ func TestIncrementalChangelogReadAppliesFileDeletes(t
*testing.T) {
_, 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))
+ var inserted, removed, replaced []int64
+ for _, row := range got {
+ switch {
+ case row.op == string(table.ChangelogOpInsert) && row.ordinal
== 0:
+ inserted = append(inserted, row.id)
+ case row.op == string(table.ChangelogOpDelete) && row.ordinal
== 1:
+ removed = append(removed, row.id)
+ case row.op == string(table.ChangelogOpInsert) && row.ordinal
== 1:
+ replaced = append(replaced, row.id)
+ }
+ }
+ require.Equal(t, []int64{1, 3}, inserted)
+ require.Equal(t, []int64{1, 3}, removed)
+ require.Equal(t, []int64{4}, replaced)
}
func TestIncrementalChangelogReadListColumn(t *testing.T) {
@@ -320,8 +412,8 @@ func
TestIncrementalChangelogReadPreRangeDeleteStaysOutOfFileDelete(t *testing.T
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))
+ require.Equal(t, []int64{4}, changelogIDs(got, table.ChangelogOpInsert))
+ require.Equal(t, []int64{1, 3}, changelogIDs(got,
table.ChangelogOpDelete))
}
func TestIncrementalChangelogReadPreRangeSharedDeletionVectorStaysOut(t
*testing.T) {
@@ -346,10 +438,10 @@ func
TestIncrementalChangelogReadPreRangeSharedDeletionVectorStaysOut(t *testing
require.NoError(t, err)
got := collectChangelogRows(t, mustReadChangelog(t, ctx,
tbl.NewIncrementalChangelogScan().FromSnapshotExclusive(removed)))
- require.Equal(t, []int64{5}, changelogIDs(got, table.ChangelogOpInsert,
0))
+ require.Equal(t, []int64{5}, changelogIDs(got, table.ChangelogOpInsert))
// 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))
+ require.Equal(t, []int64{1, 4}, changelogIDs(got,
table.ChangelogOpDelete))
}
func mustReadChangelog(t *testing.T, ctx context.Context, scan
*table.IncrementalChangelogScan) iter.Seq2[arrow.RecordBatch, error] {
@@ -450,25 +542,409 @@ func schemaFieldNames(schema *arrow.Schema) []string {
return names
}
-func clearChangelogSnapshot(rows []changelogRow) []changelogRow {
- out := make([]changelogRow, len(rows))
- for i, row := range rows {
- row.snapshot = 0
- out[i] = row
+func TestIncrementalChangelogReadPositionDeletes(t *testing.T) {
+ ctx := t.Context()
+ tbl := appendChangelogRows(t, ctx, newChangelogMORTable(t, 2))
+ tbl = deleteChangelogID(t, ctx, tbl, 2)
+ firstDelete := tbl.CurrentSnapshot().SnapshotID
+ tbl = deleteChangelogID(t, ctx, tbl, 3)
+
+ got := readChangelog(t, ctx, tbl.NewIncrementalChangelogScan())
+ require.Equal(t, []int64{1, 2, 3}, changelogIDs(got,
table.ChangelogOpInsert))
+ require.Equal(t, []int64{2, 3}, changelogIDs(got,
table.ChangelogOpDelete))
+
+ later := readChangelog(t, ctx,
tbl.NewIncrementalChangelogScan().FromSnapshotExclusive(firstDelete))
+ require.Equal(t, []int64{3}, changelogIDs(later,
table.ChangelogOpDelete))
+ require.Empty(t, changelogIDs(later, table.ChangelogOpInsert))
+
+ filtered := readChangelog(t, ctx, tbl.NewIncrementalChangelogScan(
+ table.WithRowFilter(iceberg.EqualTo(iceberg.Reference("id"),
int64(2))),
+ ))
+ require.Equal(t, []int64{2}, changelogIDs(filtered,
table.ChangelogOpInsert))
+ require.Equal(t, []int64{2}, changelogIDs(filtered,
table.ChangelogOpDelete))
+}
+
+func TestIncrementalChangelogReadDeletionVectors(t *testing.T) {
+ ctx := t.Context()
+ tbl := appendChangelogRows(t, ctx, newChangelogMORTable(t, 3))
+ tbl = deleteChangelogID(t, ctx, tbl, 2)
+
+ got := readChangelog(t, ctx, tbl.NewIncrementalChangelogScan())
+ require.Equal(t, []int64{1, 2, 3}, changelogIDs(got,
table.ChangelogOpInsert))
+ require.Equal(t, []int64{2}, changelogIDs(got, table.ChangelogOpDelete))
+ require.Equal(t, "b", changelogData(got, table.ChangelogOpDelete)[0])
+}
+
+func TestIncrementalChangelogReadEqualityDeletes(t *testing.T) {
+ ctx := t.Context()
+ tbl := newChangelogMORTable(t, 2)
+ arrowSchema, err :=
table.SchemaToArrowSchema(tbl.Metadata().CurrentSchema(), nil, false, 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"}
+ ]`)
+ tx := tbl.NewTransaction()
+ require.NoError(t, tx.AddDataFilesByPath(ctx, []string{dataPath}, nil,
false))
+ tbl, err = tx.Commit(ctx)
+ require.NoError(t, err)
+
+ eqPath := tbl.Location() + "/data/eq-del.parquet"
+ eqSchema, err := table.SchemaToArrowSchema(
+ iceberg.NewSchema(0, iceberg.NestedField{ID: 1, Name: "id",
Type: iceberg.PrimitiveTypes.Int64, Required: true}),
+ nil, true, false)
+ require.NoError(t, err)
+ writeParquetFile(t, eqPath, eqSchema, `[{"id": 2}]`)
+
+ eqBuilder, err := iceberg.NewDataFileBuilder(
+ *iceberg.UnpartitionedSpec, iceberg.EntryContentEqDeletes,
+ eqPath, iceberg.ParquetFile, nil, nil, nil, 1, 128)
+ require.NoError(t, err)
+ eqBuilder.EqualityFieldIDs([]int{1})
+
+ tx = tbl.NewTransaction()
+ rd := tx.NewRowDelta(nil)
+ rd.AddDeletes(eqBuilder.Build())
+ require.NoError(t, rd.Commit(ctx))
+ tbl, err = tx.Commit(ctx)
+ require.NoError(t, err)
+
+ got := readChangelog(t, ctx, tbl.NewIncrementalChangelogScan())
+ require.Equal(t, []int64{1, 2, 3}, changelogIDs(got,
table.ChangelogOpInsert))
+ require.Equal(t, []int64{2}, changelogIDs(got, table.ChangelogOpDelete))
+
+ // id is not projected. The equality delete still has to match it.
+ schema, records, err :=
tbl.NewIncrementalChangelogScan(table.WithSelectedFields("data")).ToArrowRecords(ctx)
+ require.NoError(t, err)
+ require.Equal(t, []string{
+ "data",
+ table.ChangelogChangeTypeColumn,
+ table.ChangelogChangeOrdinalColumn,
+ table.ChangelogCommitSnapshotIDColumn,
+ }, schemaFieldNames(schema))
+ var deletedData []string
+ for rec, err := range records {
+ require.NoError(t, err)
+ values := rec.Column(0).(*array.String)
+ ops := rec.Column(1).(*array.String)
+ for i := range int(rec.NumRows()) {
+ if ops.Value(i) == string(table.ChangelogOpDelete) {
+ deletedData = append(deletedData,
values.Value(i))
+ }
+ }
+ rec.Release()
}
+ require.Equal(t, []string{"b"}, deletedData)
+}
- return out
+func TestIncrementalChangelogReadSameSnapshotPositionDelete(t *testing.T) {
+ ctx := t.Context()
+ tbl := newChangelogMORTable(t, 2)
+ 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)
+
+ got := readChangelog(t, ctx, tbl.NewIncrementalChangelogScan())
+ require.Equal(t, []int64{1, 3}, changelogIDs(got,
table.ChangelogOpInsert))
+ require.Empty(t, changelogIDs(got, table.ChangelogOpDelete))
}
-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)
+func TestIncrementalChangelogReadReplacedDeletionVector(t *testing.T) {
+ ctx := t.Context()
+ tbl := appendChangelogRows(t, ctx, newChangelogMORTable(t, 3))
+ tbl = deleteChangelogID(t, ctx, tbl, 2)
+ firstDelete := tbl.CurrentSnapshot().SnapshotID
+ tbl = deleteChangelogID(t, ctx, tbl, 3)
+
+ got := readChangelog(t, ctx, tbl.NewIncrementalChangelogScan())
+ require.Equal(t, []int64{1, 2, 3}, changelogIDs(got,
table.ChangelogOpInsert))
+ require.Equal(t, []int64{2, 3}, changelogIDs(got,
table.ChangelogOpDelete))
+
+ later := readChangelog(t, ctx,
tbl.NewIncrementalChangelogScan().FromSnapshotExclusive(firstDelete))
+ require.Empty(t, changelogIDs(later, table.ChangelogOpInsert))
+ require.Equal(t, []int64{3}, changelogIDs(later,
table.ChangelogOpDelete))
+}
+
+func TestIncrementalChangelogReadReplacedSharedPuffinDeletionVector(t
*testing.T) {
+ ctx := t.Context()
+ tbl, location, pathA, pathB, dvA, dvB := newTableWithSharedPuffinDVs(t)
+ shared := tbl.CurrentSnapshot().SnapshotID
+
+ dvA2 := writeDV(t, location, "dv-a2.puffin", pathA, []int64{0, 1})
+ tx := tbl.NewTransaction()
+ require.NoError(t,
tx.NewRowDelta(nil).AddDeletes(dvA2).RemoveDeletes(dvA).Commit(ctx))
+ tbl, err := tx.Commit(ctx)
+ require.NoError(t, err)
+
+ dvB2 := writeDV(t, location, "dv-b2.puffin", pathB, []int64{0, 1})
+ tx = tbl.NewTransaction()
+ require.NoError(t,
tx.NewRowDelta(nil).AddDeletes(dvB2).RemoveDeletes(dvB).Commit(ctx))
+ tbl, err = tx.Commit(ctx)
+ require.NoError(t, err)
+
+ later := readChangelog(t, ctx,
tbl.NewIncrementalChangelogScan().FromSnapshotExclusive(shared))
+ require.ElementsMatch(t, []int64{2, 4}, changelogIDs(later,
table.ChangelogOpDelete))
+}
+
+func TestIncrementalChangelogReadExpiredParentKeepsPreRangeDelete(t
*testing.T) {
+ ctx := t.Context()
+ tbl := appendChangelogRows(t, ctx, newChangelogMORTable(t, 2))
+ tbl = deleteChangelogID(t, ctx, tbl, 1)
+ deletedParent := tbl.CurrentSnapshot().SnapshotID
+
+ 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)
+
+ builder, err := table.MetadataBuilderFromBase(tbl.Metadata(),
tbl.MetadataLocation())
+ require.NoError(t, err)
+ require.NoError(t, builder.RemoveSnapshots([]int64{deletedParent},
false))
+ meta, err := builder.Build()
+ require.NoError(t, err)
+ tbl = table.New(tbl.Identifier(), meta, tbl.MetadataLocation(),
func(context.Context) (iceio.IO, error) {
+ return iceio.LocalFS{}, nil
+ }, nil)
+
+ got := readChangelog(t, ctx,
tbl.NewIncrementalChangelogScan().FromSnapshotExclusive(deletedParent))
+ require.Equal(t, []int64{4}, changelogIDs(got, table.ChangelogOpInsert))
+ require.ElementsMatch(t, []int64{2, 3}, changelogIDs(got,
table.ChangelogOpDelete))
+}
+
+func TestIncrementalChangelogReadSkipsReplaceBeforeRowDelete(t *testing.T) {
+ ctx := t.Context()
+ tbl := newChangelogMORTable(t, 3)
+ arrowSchema, err :=
table.SchemaToArrowSchema(tbl.Metadata().CurrentSchema(), nil, false, false)
+ require.NoError(t, err)
+ 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(ctx, array.NewTableReader(data, -1), nil)
+ require.NoError(t, err)
+ }
+
+ tasks, err := tbl.Scan().PlanFiles(ctx)
+ require.NoError(t, err)
+ require.Len(t, tasks, 2)
+ plan, err := defaultTestCompactionCfg.PlanCompaction(tasks)
+ require.NoError(t, err)
+ require.NotEmpty(t, plan.Groups)
+ _, tbl = runRewriteWithCleanup(t, tbl, toTaskGroups(plan.Groups), false)
+ replaceID := tbl.CurrentSnapshot().SnapshotID
+ require.Equal(t, table.OpReplace,
tbl.CurrentSnapshot().Summary.Operation)
+
+ tbl = deleteChangelogID(t, ctx, tbl, 2)
+ got := readChangelog(t, ctx, tbl.NewIncrementalChangelogScan())
+ require.Equal(t, []int64{1, 2, 3}, changelogIDs(got,
table.ChangelogOpInsert))
+ require.Equal(t, []int64{2}, changelogIDs(got, table.ChangelogOpDelete))
+ for _, row := range got {
+ require.NotEqual(t, replaceID, row.snapshot)
+ }
+}
+
+func TestIncrementalChangelogReadPartitionedRowFilter(t *testing.T) {
+ ctx := t.Context()
+ schema := iceberg.NewSchema(0,
+ iceberg.NestedField{ID: 1, Name: "id", Type:
iceberg.PrimitiveTypes.Int64, Required: true},
+ iceberg.NestedField{ID: 2, Name: "data", Type:
iceberg.PrimitiveTypes.String, Required: false},
+ iceberg.NestedField{ID: 3, Name: "part", Type:
iceberg.PrimitiveTypes.Int64, Required: true},
+ )
+ spec := iceberg.NewPartitionSpec(iceberg.PartitionField{
+ SourceIDs: []int{3}, FieldID: 1000, Name: "part", Transform:
iceberg.IdentityTransform{},
+ })
+ tbl := newChangelogTable(t, 2, schema, &spec)
+
+ 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", "part": 0}, {"id": 2, "data": "b",
"part": 0}, {"id": 3, "data": "c", "part": 1}]`,
+ })
+ 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, 2)
+
+ tbl = deleteChangelogID(t, ctx, tbl, 2)
+ filtered := readChangelog(t, ctx, tbl.NewIncrementalChangelogScan(
+ table.WithSelectedFields("id"),
+ table.WithRowFilter(iceberg.EqualTo(iceberg.Reference("id"),
int64(2))),
+ ))
+ require.Equal(t, []int64{2}, changelogIDs(filtered,
table.ChangelogOpInsert))
+ require.Equal(t, []int64{2}, changelogIDs(filtered,
table.ChangelogOpDelete))
+
+ partZero := readChangelog(t, ctx, tbl.NewIncrementalChangelogScan(
+ table.WithSelectedFields("id"),
+ table.WithRowFilter(iceberg.EqualTo(iceberg.Reference("part"),
int64(0))),
+ ))
+ require.Equal(t, []int64{1, 2}, changelogIDs(partZero,
table.ChangelogOpInsert))
+ require.Equal(t, []int64{2}, changelogIDs(partZero,
table.ChangelogOpDelete))
+
+ partOne := readChangelog(t, ctx, tbl.NewIncrementalChangelogScan(
+ table.WithSelectedFields("id"),
+ table.WithRowFilter(iceberg.EqualTo(iceberg.Reference("part"),
int64(1))),
+ ))
+ require.Equal(t, []int64{3}, changelogIDs(partOne,
table.ChangelogOpInsert))
+ require.Empty(t, changelogIDs(partOne, table.ChangelogOpDelete))
+}
+
+func TestIncrementalChangelogReadDeletedRowLineage(t *testing.T) {
+ ctx := t.Context()
+ tbl := appendChangelogRows(t, ctx, newChangelogMORTable(t, 3))
+ tbl = deleteChangelogID(t, ctx, tbl, 2)
+
+ tasks, err :=
tbl.NewIncrementalChangelogScan(table.WithRowLineage()).PlanFiles(ctx)
+ require.NoError(t, err)
+ var deleted table.DeletedRowsScanTask
+ var found bool
+ for _, task := range tasks {
+ rows, ok := task.(table.DeletedRowsScanTask)
+ if !ok {
+ continue
}
+ deleted = rows
+ found = true
+
+ break
}
+ require.True(t, found)
+ require.NotNil(t, deleted.ScanTask().File.FirstRowID())
+ require.NotNil(t, deleted.ScanTask().DataSequenceNumber)
- return ids
+ schema, records, err :=
tbl.NewIncrementalChangelogScan(table.WithRowLineage()).Read(ctx,
[]table.ChangelogScanTask{deleted})
+ require.NoError(t, err)
+ require.Contains(t, schemaFieldNames(schema), iceberg.RowIDColumnName)
+ require.Contains(t, schemaFieldNames(schema),
iceberg.LastUpdatedSequenceNumberColumnName)
+
+ var gotRowID, gotSeq int64
+ var n int
+ for rec, err := range records {
+ require.NoError(t, err)
+ idIdx := rec.Schema().FieldIndices("id")
+ rowIdx := rec.Schema().FieldIndices(iceberg.RowIDColumnName)
+ seqIdx :=
rec.Schema().FieldIndices(iceberg.LastUpdatedSequenceNumberColumnName)
+ opIdx :=
rec.Schema().FieldIndices(table.ChangelogChangeTypeColumn)
+ require.NotEmpty(t, idIdx)
+ require.NotEmpty(t, rowIdx)
+ require.NotEmpty(t, seqIdx)
+ require.NotEmpty(t, opIdx)
+
+ ids := rec.Column(idIdx[0]).(*array.Int64)
+ rowIDs := rec.Column(rowIdx[0]).(*array.Int64)
+ seqs := rec.Column(seqIdx[0]).(*array.Int64)
+ ops := rec.Column(opIdx[0]).(*array.String)
+ for i := range int(rec.NumRows()) {
+ require.Equal(t, string(table.ChangelogOpDelete),
ops.Value(i))
+ require.Equal(t, int64(2), ids.Value(i))
+ require.False(t, rowIDs.IsNull(i))
+ require.False(t, seqs.IsNull(i))
+ gotRowID = rowIDs.Value(i)
+ gotSeq = seqs.Value(i)
+ n++
+ }
+ rec.Release()
+ }
+ require.Equal(t, 1, n)
+ require.Equal(t, *deleted.ScanTask().File.FirstRowID()+1, gotRowID)
+ require.Equal(t, *deleted.ScanTask().DataSequenceNumber, gotSeq)
+}
+
+func TestIncrementalChangelogReadSchemaEvolution(t *testing.T) {
+ ctx := t.Context()
+ tbl := appendChangelogRows(t, ctx, newChangelogMORTable(t, 3))
+ tx := tbl.NewTransaction()
+ require.NoError(t, tx.UpdateSchema(true, false).
+ AddColumn([]string{"extra"}, iceberg.PrimitiveTypes.String, "",
false, iceberg.StringLiteral("x")).
+ Commit())
+ tbl, err := tx.Commit(ctx)
+ require.NoError(t, err)
+ tbl = deleteChangelogID(t, ctx, tbl, 2)
+
+ schema, records, err :=
tbl.NewIncrementalChangelogScan(table.WithSelectedFields("id",
"extra")).ToArrowRecords(ctx)
+ require.NoError(t, err)
+ require.Equal(t, []string{"id", "extra",
table.ChangelogChangeTypeColumn, table.ChangelogChangeOrdinalColumn,
table.ChangelogCommitSnapshotIDColumn}, schemaFieldNames(schema))
+
+ got := collectChangelogRows(t, records)
+ require.Equal(t, []int64{2}, changelogIDs(got, table.ChangelogOpDelete))
+ require.Equal(t, []string{"x"}, changelogData(got,
table.ChangelogOpDelete))
+}
+
+func newChangelogMORTable(t *testing.T, version int) *table.Table {
+ t.Helper()
+
+ schema := iceberg.NewSchema(0,
+ iceberg.NestedField{ID: 1, Name: "id", Type:
iceberg.PrimitiveTypes.Int64, Required: true},
+ iceberg.NestedField{ID: 2, Name: "data", Type:
iceberg.PrimitiveTypes.String, Required: false},
+ )
+
+ return newChangelogTable(t, version, schema, iceberg.UnpartitionedSpec)
+}
+
+func newChangelogTable(t *testing.T, version int, schema *iceberg.Schema, spec
*iceberg.PartitionSpec) *table.Table {
+ t.Helper()
+
+ location := filepath.ToSlash(t.TempDir())
+ meta, err := table.NewMetadata(schema, spec, table.UnsortedSortOrder,
location,
+ iceberg.Properties{
+ table.PropertyFormatVersion: strconv.Itoa(version),
+ table.WriteDeleteModeKey: table.WriteModeMergeOnRead,
+ })
+ require.NoError(t, err)
+
+ metaLoc := location + "/metadata/v1.metadata.json"
+ fsF := func(context.Context) (iceio.IO, error) { return
iceio.LocalFS{}, nil }
+ cat := &concurrentTestCatalog{metadata: meta, location: metaLoc, fsF:
fsF}
+
+ return table.New(table.Identifier{"db", "changelog_read"}, meta,
metaLoc, fsF, cat)
+}
+
+func appendChangelogRows(t *testing.T, ctx context.Context, tbl *table.Table)
*table.Table {
+ t.Helper()
+
+ 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)
+
+ return tbl
}
func changelogDataFile(t *testing.T, path string, records int64)
iceberg.DataFile {
@@ -496,3 +972,53 @@ func changelogPosDeleteFile(t *testing.T, path string)
iceberg.DataFile {
return b.Build()
}
+
+func deleteChangelogID(t *testing.T, ctx context.Context, tbl *table.Table, id
int64) *table.Table {
+ t.Helper()
+
+ tbl, err := tbl.Delete(ctx, iceberg.EqualTo(iceberg.Reference("id"),
id), nil)
+ require.NoError(t, err)
+
+ return tbl
+}
+
+func readChangelog(t *testing.T, ctx context.Context, scan
*table.IncrementalChangelogScan) []changelogRow {
+ t.Helper()
+
+ _, records, err := scan.ToArrowRecords(ctx)
+ require.NoError(t, err)
+
+ return collectChangelogRows(t, records)
+}
+
+func changelogIDs(rows []changelogRow, op table.ChangelogOperation) []int64 {
+ var ids []int64
+ for _, row := range rows {
+ if row.op == string(op) {
+ ids = append(ids, row.id)
+ }
+ }
+
+ return ids
+}
+
+func changelogData(rows []changelogRow, op table.ChangelogOperation) []string {
+ var data []string
+ for _, row := range rows {
+ if row.op == string(op) {
+ data = append(data, row.data)
+ }
+ }
+
+ return data
+}
+
+func clearChangelogSnapshot(rows []changelogRow) []changelogRow {
+ out := make([]changelogRow, len(rows))
+ for i, row := range rows {
+ row.snapshot = 0
+ out[i] = row
+ }
+
+ return out
+}
diff --git a/table/changelog_reader_test.go b/table/changelog_reader_test.go
index 4f3545434..3661ad70f 100644
--- a/table/changelog_reader_test.go
+++ b/table/changelog_reader_test.go
@@ -34,15 +34,13 @@ func TestChangelogReadRejectsNilScan(t *testing.T) {
require.ErrorIs(t, err, ErrInvalidOperation)
}
-func TestChangelogReadRejectsDeletedRowTasks(t *testing.T) {
+func TestChangelogReadAcceptsDeletedRowTasks(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)
+ require.NoError(t, readableChangelogTask(task))
}
func TestStampChangelogBatchCutsToTheRemainingBudget(t *testing.T) {
diff --git a/table/scanner.go b/table/scanner.go
index f9c1f860a..c8ad9effd 100644
--- a/table/scanner.go
+++ b/table/scanner.go
@@ -2186,10 +2186,20 @@ func bindReadTasksResiduals(
// therefore still hold its whole decoded contents while it waits. The bound
for
// a compaction pipeline is stated on [WithCompactionArrowBatchSize].
func (scan *Scan) ReadTasks(ctx context.Context, tasks []FileScanTask)
(*arrow.Schema, iter.Seq2[arrow.RecordBatch, error], error) {
+ return scan.readFileTasks(ctx, tasks, nil)
+}
+
+// readFileTasks reads tasks. addedDeletes, when non-nil, is paired with the
+// single data file in tasks: rows that survive that task's own deletes are
+// kept only when they match addedDeletes. Normal scans pass nil.
+func (scan *Scan) readFileTasks(ctx context.Context, tasks []FileScanTask,
addedDeletes *FileScanTask) (*arrow.Schema, iter.Seq2[arrow.RecordBatch,
error], error) {
as, readTasks, releasePlanIO, err := scan.arrowScanForTasks(ctx, tasks)
if err != nil {
return nil, nil, err
}
+ if addedDeletes != nil {
+ as.added = &addedDeleteState{deletes: addedDeletes}
+ }
readRecords := scan.readRecords
if readRecords == nil {
diff --git a/website/src/api.md b/website/src/api.md
index 6486ff5f4..e6110ee4b 100644
--- a/website/src/api.md
+++ b/website/src/api.md
@@ -303,6 +303,26 @@ if err != nil { /* ... */ }
arrowSchema, batches, err := scan.ReadTasks(ctx, tasks)
```
+### Incremental changelog scans
+
+`NewIncrementalChangelogScan` returns rows added or deleted between snapshots.
`PlanFiles` returns the tasks. `Read` reads a planned task list, and
`ToArrowRecords` plans and reads. Remote planning is not supported.
+
+```go
+schema, batches, err := tbl.NewIncrementalChangelogScan(
+ table.WithSelectedFields("id", "name"),
+).FromSnapshotExclusive(startSnapshotID).ToArrowRecords(ctx)
+```
+
+`FromSnapshotInclusive`, `FromSnapshotExclusive`, and `ToSnapshot` bound the
snapshot range. Projection, row filters, and limits apply while the rows are
read. If the start snapshot has been expired, changes in expired snapshots
before the oldest retained one are not reported. Each snapshot that adds
deletes is planned against every live data file, narrowed only by the partition
filter, so the cost follows the table size rather than the size of the snapshot
range.
+
+Each batch is the scan projection followed by three columns:
+
+- `_change_type` — `INSERT` or `DELETE`
+- `_change_ordinal` — int32 ordinal of the snapshot that produced the change
+- `_commit_snapshot_id` — snapshot ID that committed the change
+
+Added data files are inserts. Removed data files are deletes of the rows in
that file. Delete files added in the range emit the rows they newly remove. A
rewritten file emits a DELETE and an INSERT for every row in that file,
including rows that did not change. `UPDATE` rows are not produced. Delete-file
handling is iceberg-go behavior.
+
## Writing data
The shortcut methods on `Table` open a transaction, perform the write, and
commit. Use `NewTransaction` directly when you need to combine multiple
operations.
