feat(table): add incremental changelog scan

manifest.go

@@ -3318,7 +3318,8 @@ type DataFile interface {
 // ManifestEntry is an interface for both v1 and v2 manifest entries.
 type ManifestEntry interface {
 	// Status returns the type of the file tracked by this entry.
-	// Deletes are informational only and not used in scans.
+	// Whether entries with EntryStatusDELETED are returned depends on the
+	// caller's discardDeleted option.
 	Status() ManifestEntryStatus
 	// SnapshotID is the id where the file was added, or deleted,
 	// if null it is inherited from the manifest list.

table/changelog_scan_task.go

@@ -35,6 +35,11 @@ const (
 
 // ChangelogScanTask is a unit of work that produces changelog rows.
 type ChangelogScanTask interface {
+	// Implementations are intentionally limited to the changelog task types
+	// provided by this package.
+	isChangelogScanTask()
+	// ScanTask returns the underlying file scan task for reading the data file.
+	ScanTask() FileScanTask
 	Operation() ChangelogOperation
 	ChangeOrdinal() int
 	CommitSnapshotID() int64
@@ -91,6 +96,8 @@ func NewAddedRowsScanTask(dataFile iceberg.DataFile, deletes []iceberg.DataFile,
 func (t AddedRowsScanTask) Operation() ChangelogOperation { return ChangelogOpInsert }
 func (t AddedRowsScanTask) ChangeOrdinal() int            { return t.changeOrdinal }
 func (t AddedRowsScanTask) CommitSnapshotID() int64       { return t.commitSnapshotID }
+func (t AddedRowsScanTask) ScanTask() FileScanTask        { return t.FileScanTask }
+func (AddedRowsScanTask) isChangelogScanTask()            {}
 
 // Deletes returns every delete file applied while reading the added data
 // file: position deletes, then equality deletes, then deletion vectors.
@@ -125,6 +132,8 @@ func NewDeletedDataFileScanTask(dataFile iceberg.DataFile, existingDeletes []ice
 func (t DeletedDataFileScanTask) Operation() ChangelogOperation { return ChangelogOpDelete }
 func (t DeletedDataFileScanTask) ChangeOrdinal() int            { return t.changeOrdinal }
 func (t DeletedDataFileScanTask) CommitSnapshotID() int64       { return t.commitSnapshotID }
+func (t DeletedDataFileScanTask) ScanTask() FileScanTask        { return t.FileScanTask }
+func (DeletedDataFileScanTask) isChangelogScanTask()            {}
 
 // ExistingDeletes returns delete files that applied before the data file was
 // removed.
@@ -168,6 +177,8 @@ func NewDeletedRowsScanTask(dataFile iceberg.DataFile, addedDeletes, existingDel
 func (t DeletedRowsScanTask) Operation() ChangelogOperation { return ChangelogOpDelete }
 func (t DeletedRowsScanTask) ChangeOrdinal() int            { return t.changeOrdinal }
 func (t DeletedRowsScanTask) CommitSnapshotID() int64       { return t.commitSnapshotID }
+func (t DeletedRowsScanTask) ScanTask() FileScanTask        { return t.FileScanTask }
+func (DeletedRowsScanTask) isChangelogScanTask()            {}
 
 // AddedDeletes returns delete files whose removals should appear in the
 // changelog.

table/changelog_scan_task_test.go

@@ -98,6 +98,9 @@ func TestChangelogScanTaskInterface(t *testing.T) {
 	require.Equal(t, ChangelogOpInsert, tasks[0].Operation())
 	require.Equal(t, ChangelogOpDelete, tasks[1].Operation())
 	require.Equal(t, ChangelogOpDelete, tasks[2].Operation())
+	require.Equal(t, data.FilePath(), tasks[0].ScanTask().File.FilePath())
+	require.Equal(t, data.FilePath(), tasks[1].ScanTask().File.FilePath())
+	require.Equal(t, data.FilePath(), tasks[2].ScanTask().File.FilePath())
 }
 
 func TestClassifyDeleteFiles(t *testing.T) {

table/incremental_append_scan.go

@@ -253,48 +253,13 @@ func (s *IncrementalAppendScan) toSnapshot() (*Snapshot, error) {
 }
 
 func (s *IncrementalAppendScan) snapshotsBetween(toSnapshotID int64) ([]Snapshot, error) {
-	ancestors := AncestorsOf(toSnapshotID, s.scan.metadata.SnapshotByID)
-	if len(ancestors) == 0 {
-		return nil, fmt.Errorf("%w: ending snapshot not found: %d", iceberg.ErrInvalidArgument, toSnapshotID)
-	}
-
-	if s.fromSnapshotID == nil {
-		slices.Reverse(ancestors)
-
-		return appendOnlySnapshots(ancestors)
-	}
-
-	fromID := *s.fromSnapshotID
-	if !s.fromInclusive {
-		if fromID == toSnapshotID {
-			return nil, fmt.Errorf("%w: starting snapshot %d must be a parent ancestor of ending snapshot %d for an exclusive scan",
-				iceberg.ErrInvalidArgument, fromID, toSnapshotID)
-		}
-		between, found := AncestorsBetween(toSnapshotID, fromID, s.scan.metadata.SnapshotByID)
-		if !found {
-			return nil, fmt.Errorf("%w: starting snapshot %d is not an ancestor of ending snapshot %d", iceberg.ErrInvalidArgument, fromID, toSnapshotID)
-		}
-		slices.Reverse(between)
-
-		return appendOnlySnapshots(between)
-	}
-
-	if s.scan.metadata.SnapshotByID(fromID) == nil {
-		return nil, fmt.Errorf("%w: starting snapshot not found: %d", iceberg.ErrInvalidArgument, fromID)
-	}
-	if !IsAncestorOf(toSnapshotID, fromID, s.scan.metadata.SnapshotByID) {
-		return nil, fmt.Errorf("%w: starting snapshot %d is not an ancestor of ending snapshot %d", iceberg.ErrInvalidArgument, fromID, toSnapshotID)
-	}
-	selected := make([]Snapshot, 0, len(ancestors))
-	for _, snapshot := range ancestors {
-		selected = append(selected, snapshot)
-		if snapshot.SnapshotID == fromID {
-			break
-		}
+	snapshots, err := incrementalSnapshotsBetween(
+		s.scan.metadata, s.fromSnapshotID, s.fromInclusive, toSnapshotID)
+	if err != nil {
+		return nil, err
 	}
-	slices.Reverse(selected)
 
-	return appendOnlySnapshots(selected)
+	return appendOnlySnapshots(snapshots)
 }
 
 func appendOnlySnapshots(snapshots []Snapshot) ([]Snapshot, error) {

table/incremental_changelog_row_lineage_test.go

@@ -0,0 +1,83 @@
+// 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"
+	"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/table"
+	"github.com/stretchr/testify/require"
+)
+
+func TestIncrementalChangelogScanPreservesV3RowLineageMetadata(t *testing.T) {
+	ctx := context.Background()
+	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)
+
+	initialData, err := array.TableFromJSON(mem, arrowSchema, []string{
+		`[{"id": 1, "data": "a"}]`,
+	})
+	require.NoError(t, err)
+	t.Cleanup(initialData.Release)
+
+	tbl, err = tbl.Append(ctx, array.NewTableReader(initialData, -1), nil)
+	require.NoError(t, err)
+
+	replacementData, err := array.TableFromJSON(mem, arrowSchema, []string{
+		`[{"id": 2, "data": "b"}]`,
+	})
+	require.NoError(t, err)
+	t.Cleanup(replacementData.Release)
+
+	tbl, err = tbl.Overwrite(ctx, array.NewTableReader(replacementData, -1), nil,
+		table.WithOverwriteConcurrency(1))
+	require.NoError(t, err)
+
+	tasks, err := tbl.NewIncrementalChangelogScan().PlanFiles(ctx)
+	require.NoError(t, err)
+	require.Len(t, tasks, 3)
+
+	insertTask := tasks[0].ScanTask()
+	deleteTask := tasks[1].ScanTask()
+	laterInsertTask := tasks[2].ScanTask()
+	require.Equal(t, table.ChangelogOpInsert, tasks[0].Operation())
+	require.Equal(t, table.ChangelogOpDelete, tasks[1].Operation())
+	require.Equal(t, table.ChangelogOpInsert, tasks[2].Operation())
+	require.Equal(t, insertTask.File.FilePath(), deleteTask.File.FilePath())
+	require.NotEqual(t, insertTask.File.FilePath(), laterInsertTask.File.FilePath())
+
+	for _, task := range []table.FileScanTask{insertTask, deleteTask, laterInsertTask} {
+		require.NotNil(t, task.FirstRowID)
+		require.NotNil(t, task.DataSequenceNumber)
+	}
+	require.Equal(t, int64(0), *insertTask.FirstRowID)
+	require.Equal(t, int64(1), *laterInsertTask.FirstRowID)
+	require.Equal(t, int64(0), *deleteTask.FirstRowID)
+	require.Equal(t, int64(1), *insertTask.DataSequenceNumber)
+	require.Equal(t, int64(2), *laterInsertTask.DataSequenceNumber)
+	require.Equal(t, int64(1), *deleteTask.DataSequenceNumber)
+}