From 72b676417ebacd55364d85bc7bdf2a987778278a Mon Sep 17 00:00:00 2001 From: Minh Vu Date: Fri, 4 Sep 2026 23:51:08 +0200 Subject: [PATCH] feat(table): add incremental changelog scan Signed-off-by: Minh Vu --- manifest.go | 3 +- table/changelog_scan_task.go | 11 + table/changelog_scan_task_test.go | 3 + table/incremental_append_scan.go | 45 +- .../incremental_changelog_row_lineage_test.go | 83 ++ table/incremental_changelog_scan.go | 386 +++++++++ table/incremental_changelog_scan_test.go | 784 ++++++++++++++++++ table/incremental_scan.go | 83 ++ table/scanner.go | 122 ++- 9 files changed, 1468 insertions(+), 52 deletions(-) create mode 100644 table/incremental_changelog_row_lineage_test.go create mode 100644 table/incremental_changelog_scan.go create mode 100644 table/incremental_changelog_scan_test.go create mode 100644 table/incremental_scan.go diff --git a/manifest.go b/manifest.go index f6bb9849f..67705f498 100644 --- a/manifest.go +++ b/manifest.go @@ -3353,7 +3353,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. diff --git a/table/changelog_scan_task.go b/table/changelog_scan_task.go index bfa5a4106..b43782849 100644 --- a/table/changelog_scan_task.go +++ b/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. diff --git a/table/changelog_scan_task_test.go b/table/changelog_scan_task_test.go index e5af7f5ab..387589cdd 100644 --- a/table/changelog_scan_task_test.go +++ b/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) { diff --git a/table/incremental_append_scan.go b/table/incremental_append_scan.go index 98a1f6415..837eda6c6 100644 --- a/table/incremental_append_scan.go +++ b/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) { diff --git a/table/incremental_changelog_row_lineage_test.go b/table/incremental_changelog_row_lineage_test.go new file mode 100644 index 000000000..3764e6d15 --- /dev/null +++ b/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) +} diff --git a/table/incremental_changelog_scan.go b/table/incremental_changelog_scan.go new file mode 100644 index 000000000..dcb5f2620 --- /dev/null +++ b/table/incremental_changelog_scan.go @@ -0,0 +1,386 @@ +// 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 ( + "cmp" + "context" + "fmt" + "maps" + "slices" + "time" + + "github.com/apache/iceberg-go" + "github.com/apache/iceberg-go/metrics" +) + +// IncrementalChangelogScan plans data-file changes between snapshots. It +// emits insert and delete tasks for data-manifest entries and skips replace +// snapshots. PlanFiles returns an error if any in-range snapshot's manifest +// list references a delete manifest, including ones carried forward from +// earlier snapshots. +type IncrementalChangelogScan struct { + scan *Scan + fromSnapshotID *int64 + fromInclusive bool + toSnapshotID *int64 +} + +type plannedChangelogTask struct { + task ChangelogScanTask + file FileScanTask +} + +// NewIncrementalChangelogScan creates an incremental changelog planner. +// Projection and row limits are not applied to returned tasks. Auto planning +// falls back to local planning, while remote planning is not supported. Use +// ChangelogScanTask.ScanTask with Scan.ReadTasks to read the returned files. +// Row filters are attached to each task as residuals without partition-specific +// simplification, matching the existing incremental append scan behavior. +func (t Table) NewIncrementalChangelogScan(opts ...ScanOption) *IncrementalChangelogScan { + return &IncrementalChangelogScan{scan: t.Scan(opts...)} +} + +// FromSnapshotInclusive includes changes committed by the starting snapshot. +// The snapshot is validated when files are planned. +func (s *IncrementalChangelogScan) FromSnapshotInclusive(snapshotID int64) *IncrementalChangelogScan { + out := *s + out.fromSnapshotID = &snapshotID + out.fromInclusive = true + + return &out +} + +// FromSnapshotExclusive starts after the given snapshot. The starting +// snapshot must be a parent ancestor of the ending snapshot when planning. +// The snapshot is validated when files are planned. +func (s *IncrementalChangelogScan) FromSnapshotExclusive(snapshotID int64) *IncrementalChangelogScan { + out := *s + out.fromSnapshotID = &snapshotID + out.fromInclusive = false + + return &out +} + +// ToSnapshot sets the inclusive ending snapshot. +// The snapshot is validated when files are planned. +func (s *IncrementalChangelogScan) ToSnapshot(snapshotID int64) *IncrementalChangelogScan { + out := *s + out.toSnapshotID = &snapshotID + + return &out +} + +// PlanFiles returns one task for each added or deleted data-file entry. A +// cancelled context returns its cancellation error before planning starts. +// Tasks are ordered by change ordinal, then by DELETE before INSERT within an +// ordinal, and finally by data-file path. When an ending snapshot is +// available, it emits a ScanReport through the configured reporter on +// successful planning. Changelog reports count every returned task in +// ResultDataFiles and TotalFileSizeInBytes, so a file inserted and deleted +// within the range is counted twice. +func (s *IncrementalChangelogScan) PlanFiles(ctx context.Context) ([]ChangelogScanTask, error) { + if s == nil || s.scan == nil { + return nil, fmt.Errorf("%w: incremental changelog scan is not initialized", ErrInvalidOperation) + } + if err := ctx.Err(); err != nil { + return nil, err + } + + switch s.scan.planningMode { + case ScanPlanningLocal, ScanPlanningAuto: + case ScanPlanningRemote: + return nil, fmt.Errorf("%w: incremental changelog scans do not support remote planning", ErrInvalidOperation) + default: + return nil, fmt.Errorf("%w: unknown scan planning mode %q", iceberg.ErrInvalidArgument, s.scan.planningMode) + } + start := time.Now() + + toSnapshot, err := s.toSnapshot() + if err != nil { + return nil, err + } + if toSnapshot == nil { + if s.fromSnapshotID != nil { + return nil, fmt.Errorf("%w: no ending snapshot found for incremental changelog scan from %d", + iceberg.ErrInvalidArgument, *s.fromSnapshotID) + } + + return nil, nil + } + + planningScan := *s.scan + planningScan.identifier = slices.Clone(s.scan.identifier) + planningScan.selectedFields = slices.Clone(s.scan.selectedFields) + planningScan.options = maps.Clone(s.scan.options) + if s.toSnapshotID != nil { + planningScan.snapshotID = &toSnapshot.SnapshotID + planningScan.asOfTimestamp = nil + } + schema, err := planningScan.effectiveSchema() + if err != nil { + return nil, err + } + residual, err := bindTaskFilter(schema, planningScan.rowFilter, planningScan.caseSensitive) + if err != nil { + return nil, fmt.Errorf("bind incremental changelog scan residual: %w", err) + } + var acc scanMetricsAccumulator + finish := func(plannedTasks []plannedChangelogTask) []ChangelogScanTask { + acc.resultDataFiles = int64(len(plannedTasks)) + tasks := make([]ChangelogScanTask, len(plannedTasks)) + fileTasks := make([]FileScanTask, len(plannedTasks)) + for i, planned := range plannedTasks { + tasks[i] = planned.task + fileTasks[i] = planned.file + acc.totalFileSize += planned.file.File.FileSizeBytes() + } + acc.applyResultDeleteMetrics(fileTasks) + planningDuration := time.Since(start) + + if rep := planningScan.Reporter(); !metrics.IsNop(rep) { + projected, _ := planningScan.Projection() + safeReport(ctx, rep, planningScan.buildScanReport(&acc, schema, projected, planningDuration)) + } + + return tasks + } + + snapshotRange, err := incrementalSnapshotsBetween( + s.scan.metadata, s.fromSnapshotID, s.fromInclusive, toSnapshot.SnapshotID) + if err != nil { + return nil, err + } + snapshots, err := changelogSnapshots(snapshotRange) + if err != nil { + return nil, err + } + if len(snapshots) == 0 { + return finish(nil), nil + } + + changelogSnapshotIDs := make(map[int64]struct{}, len(snapshots)) + snapshotOrdinals := make(map[int64]int, len(snapshots)) + for ordinal, snapshot := range snapshots { + changelogSnapshotIDs[snapshot.SnapshotID] = struct{}{} + snapshotOrdinals[snapshot.SnapshotID] = ordinal + } + + if s.scan.ioF == nil { + return nil, fmt.Errorf("%w: table file IO is not configured", ErrInvalidOperation) + } + fs, err := s.scan.ioF(ctx) + if err != nil { + return nil, err + } + + manifestsByPath := make(map[string]iceberg.ManifestFile) + for _, snapshot := range snapshots { + if err := ctx.Err(); err != nil { + return nil, err + } + manifests, err := snapshot.Manifests(fs) + if err != nil { + return nil, err + } + for _, manifest := range manifests { + if manifest.ManifestContent() == iceberg.ManifestContentDeletes { + return nil, fmt.Errorf("%w: incremental changelog scan range references a delete manifest originating in snapshot %d", + ErrInvalidOperation, manifest.SnapshotID()) + } + if manifest.ManifestContent() != iceberg.ManifestContentData { + continue + } + if _, ok := changelogSnapshotIDs[manifest.SnapshotID()]; !ok { + continue + } + manifestsByPath[manifest.FilePath()] = manifest + } + } + + paths := make([]string, 0, len(manifestsByPath)) + for path := range manifestsByPath { + paths = append(paths, path) + } + slices.Sort(paths) + manifestList := make([]iceberg.ManifestFile, 0, len(paths)) + for _, path := range paths { + manifestList = append(manifestList, manifestsByPath[path]) + } + + // Changelog metrics intentionally count only manifests that contain added or + // deleted data files; no-change manifests are removed before the scan metric + // accumulator sees them. + manifestList = slices.DeleteFunc(manifestList, func(manifest iceberg.ManifestFile) bool { + return !manifestHasChangelogEntries(manifest) + }) + partitionFilters := planningScan.partitionFiltersForSchema(schema) + manifestList, err = planningScan.filterManifestsWithSchemaOptions( + manifestList, schema, &acc, partitionFilters, + /* includeDeleted= */ true) + if err != nil { + return nil, err + } + if len(manifestList) == 0 { + return finish(nil), nil + } + entries, err := planningScan.collectManifestEntriesWithSchemaOptions( + ctx, manifestList, schema, + partitionFilters, + /* discardDeleted= */ false, + /* discardExisting= */ true, + ) + if err != nil { + return nil, err + } + + plannedTasks := make([]plannedChangelogTask, 0, len(entries.dataEntries)) + for _, entry := range entries.dataEntries { + ordinal, ok := snapshotOrdinals[entry.SnapshotID()] + if !ok { + continue + } + + task, err := newChangelogScanTask(entry, ordinal, residual) + if err != nil { + return nil, fmt.Errorf("incremental changelog scan snapshot %d: %w", entry.SnapshotID(), err) + } + plannedTasks = append(plannedTasks, plannedChangelogTask{ + task: task, + file: task.ScanTask(), + }) + } + slices.SortFunc(plannedTasks, func(left, right plannedChangelogTask) int { + if ordinal := cmp.Compare(left.task.ChangeOrdinal(), right.task.ChangeOrdinal()); ordinal != 0 { + return ordinal + } + if operation := cmp.Compare(changelogOperationOrder(left.task.Operation()), changelogOperationOrder(right.task.Operation())); operation != 0 { + return operation + } + + return cmp.Compare(left.file.File.FilePath(), right.file.File.FilePath()) + }) + + return finish(plannedTasks), nil +} + +func manifestHasChangelogEntries(manifest iceberg.ManifestFile) bool { + // V1 manifest lists use -1 for unknown counts, so only zero means the + // manifest is known not to contain added or deleted entries. + return manifest.AddedDataFiles() != 0 || manifest.DeletedDataFiles() != 0 +} + +func changelogOperationOrder(operation ChangelogOperation) int { + switch operation { + // Deletes must be replayed before inserts within one change ordinal. Keep + // this explicit instead of relying on the string values' lexical order. + case ChangelogOpDelete: + return 0 + case ChangelogOpInsert: + return 1 + case ChangelogOpUpdateBefore: + return 2 + case ChangelogOpUpdateAfter: + return 3 + default: + return 4 + } +} + +func changelogOperation(status iceberg.ManifestEntryStatus) (ChangelogOperation, error) { + switch status { + case iceberg.EntryStatusADDED: + return ChangelogOpInsert, nil + case iceberg.EntryStatusDELETED: + return ChangelogOpDelete, nil + default: + return "", fmt.Errorf("%w: unknown manifest entry status %d", ErrInvalidMetadata, status) + } +} + +func newChangelogScanTask(entry iceberg.ManifestEntry, ordinal int, residual iceberg.BooleanExpression) (ChangelogScanTask, error) { + operation, err := changelogOperation(entry.Status()) + if err != nil { + return nil, err + } + + file := entry.DataFile() + configureFileScanTask := func(task *FileScanTask) { + task.Start = 0 + task.Length = file.FileSizeBytes() + task.Residual = residual + task.FirstRowID = file.FirstRowID() + if sequenceNumber := entry.SequenceNum(); sequenceNumber >= 0 { + task.DataSequenceNumber = &sequenceNumber + } + } + + switch operation { + case ChangelogOpInsert: + task, err := NewAddedRowsScanTask(file, nil, ordinal, entry.SnapshotID()) + if err != nil { + return nil, err + } + configureFileScanTask(&task.FileScanTask) + + return task, nil + case ChangelogOpDelete: + task, err := NewDeletedDataFileScanTask(file, nil, ordinal, entry.SnapshotID()) + if err != nil { + return nil, err + } + configureFileScanTask(&task.FileScanTask) + + return task, nil + default: + return nil, fmt.Errorf("%w: unsupported changelog operation %q", ErrInvalidOperation, operation) + } +} + +func (s *IncrementalChangelogScan) toSnapshot() (*Snapshot, error) { + if s.toSnapshotID != nil { + snapshot := s.scan.metadata.SnapshotByID(*s.toSnapshotID) + if snapshot == nil { + return nil, fmt.Errorf("%w: ending snapshot not found: %d", iceberg.ErrInvalidArgument, *s.toSnapshotID) + } + + return snapshot, nil + } + + return s.scan.ResolveSnapshot() +} + +// changelogSnapshots retains every snapshot with a non-empty operation except +// replace snapshots. Keeping unknown future operations matches Java's +// incremental changelog scan behavior; only a missing operation is rejected. +func changelogSnapshots(snapshots []Snapshot) ([]Snapshot, error) { + result := make([]Snapshot, 0, len(snapshots)) + for _, snapshot := range snapshots { + if snapshot.Summary == nil || snapshot.Summary.Operation == "" { + return nil, fmt.Errorf("%w: cannot determine operation for snapshot %d", + ErrMissingOperation, snapshot.SnapshotID) + } + + if snapshot.Summary.Operation == OpReplace { + continue + } + result = append(result, snapshot) + } + + return result, nil +} diff --git a/table/incremental_changelog_scan_test.go b/table/incremental_changelog_scan_test.go new file mode 100644 index 000000000..75c960e38 --- /dev/null +++ b/table/incremental_changelog_scan_test.go @@ -0,0 +1,784 @@ +// 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" + "sync/atomic" + "testing" + + "github.com/apache/iceberg-go" + iceio "github.com/apache/iceberg-go/io" + "github.com/apache/iceberg-go/metrics" + "github.com/stretchr/testify/require" +) + +func TestIncrementalChangelogScanPlansAddedAndDeletedEntries(t *testing.T) { + tbl := incrementalChangelogTestTable(t) + + tasks, err := tbl.NewIncrementalChangelogScan().PlanFiles(context.Background()) + require.NoError(t, err) + require.Len(t, tasks, 4) + + expected := []struct { + path string + operation ChangelogOperation + ordinal int + commitSnapshotID int64 + }{ + {"mem://default/changelog/data-old.parquet", ChangelogOpInsert, 0, 1}, + {"mem://default/changelog/data-old.parquet", ChangelogOpDelete, 1, 2}, + {"mem://default/changelog/data-new.parquet", ChangelogOpInsert, 1, 2}, + {"mem://default/changelog/data-later.parquet", ChangelogOpInsert, 2, 4}, + } + for i, want := range expected { + fileTask := tasks[i].ScanTask() + require.Equal(t, want.path, fileTask.File.FilePath()) + require.Equal(t, want.operation, tasks[i].Operation()) + require.Equal(t, want.ordinal, tasks[i].ChangeOrdinal()) + require.Equal(t, want.commitSnapshotID, tasks[i].CommitSnapshotID()) + require.Zero(t, fileTask.DeleteFiles) + require.Zero(t, fileTask.EqualityDeleteFiles) + require.Zero(t, fileTask.DeletionVectorFiles) + require.NotNil(t, fileTask.DataSequenceNumber) + } +} + +func TestIncrementalChangelogScanSkipsManifestsWithoutChanges(t *testing.T) { + reporter := &metrics.InMemoryReporter{} + tbl := incrementalChangelogTestTable(t) + + tasks, err := tbl.NewIncrementalChangelogScan( + WithReporter(reporter), + ).PlanFiles(context.Background()) + require.NoError(t, err) + require.Len(t, tasks, 4) + + reports := reporter.Reports() + require.Len(t, reports, 1) + report, ok := reports[0].(metrics.ScanReport) + require.True(t, ok) + require.Equal(t, int64(3), report.Metrics.TotalDataManifests.Value) + require.Equal(t, int64(3), report.Metrics.ScannedDataManifests.Value) +} + +func TestOpenManifestWithOptionsCanDiscardExistingEntries(t *testing.T) { + spec := partitionedSpec() + schema := simpleSchema() + _, fs := createTestTransactionWithMemIO(t, spec) + + existingFile := newTestDataFile(t, spec, + "mem://default/changelog/existing.parquet", map[int]any{1000: int32(1)}) + addedFile := newTestDataFile(t, spec, + "mem://default/changelog/added.parquet", map[int]any{1000: int32(2)}) + snapshotID := int64(1) + sequenceNumber := int64(1) + entries := []iceberg.ManifestEntry{ + iceberg.NewManifestEntry(iceberg.EntryStatusEXISTING, &snapshotID, &sequenceNumber, &sequenceNumber, existingFile), + iceberg.NewManifestEntry(iceberg.EntryStatusADDED, &snapshotID, &sequenceNumber, &sequenceNumber, addedFile), + } + manifestPath := "mem://default/changelog/metadata/mixed-manifest.avro" + var buf bytes.Buffer + manifest, err := iceberg.WriteManifest(manifestPath, &buf, 2, spec, schema, snapshotID, entries) + require.NoError(t, err) + require.NoError(t, fs.WriteFile(manifestPath, buf.Bytes())) + + partitionCalls := 0 + metricsCalls := 0 + got, err := openManifestWithOptions( + fs, + manifest, + func(iceberg.DataFile) (bool, error) { + partitionCalls++ + + return true, nil + }, + func(iceberg.DataFile) (bool, error) { + metricsCalls++ + + return true, nil + }, + false, + true, + ) + require.NoError(t, err) + require.Len(t, got, 1) + require.Equal(t, iceberg.EntryStatusADDED, got[0].Status()) + require.Equal(t, 1, partitionCalls) + require.Equal(t, 1, metricsCalls) +} + +func TestIncrementalChangelogScanHonorsSnapshotBoundaries(t *testing.T) { + tbl := incrementalChangelogTestTable(t) + + tasks, err := tbl.NewIncrementalChangelogScan(). + FromSnapshotExclusive(1). + ToSnapshot(4). + PlanFiles(context.Background()) + require.NoError(t, err) + require.Len(t, tasks, 3) + require.Equal(t, "mem://default/changelog/data-old.parquet", tasks[0].ScanTask().File.FilePath()) + require.Equal(t, ChangelogOpDelete, tasks[0].Operation()) + require.Equal(t, 0, tasks[0].ChangeOrdinal()) + require.Equal(t, "mem://default/changelog/data-new.parquet", tasks[1].ScanTask().File.FilePath()) + require.Equal(t, ChangelogOpInsert, tasks[1].Operation()) + require.Equal(t, 0, tasks[1].ChangeOrdinal()) + require.Equal(t, "mem://default/changelog/data-later.parquet", tasks[2].ScanTask().File.FilePath()) + require.Equal(t, 1, tasks[2].ChangeOrdinal()) + + tasks, err = tbl.NewIncrementalChangelogScan(). + FromSnapshotInclusive(2). + ToSnapshot(4). + PlanFiles(context.Background()) + require.NoError(t, err) + require.Len(t, tasks, 3) + for _, task := range tasks { + require.GreaterOrEqual(t, task.CommitSnapshotID(), int64(2)) + } +} + +func TestIncrementalChangelogScanSkipsReplaceSnapshots(t *testing.T) { + tbl := incrementalChangelogTestTable(t) + + tasks, err := tbl.NewIncrementalChangelogScan(). + FromSnapshotInclusive(3). + ToSnapshot(3). + PlanFiles(context.Background()) + require.NoError(t, err) + require.Empty(t, tasks) +} + +func TestIncrementalChangelogScanPreservesChangesAndSortsByFilePath(t *testing.T) { + tbl := incrementalChangelogManifestRewriteTable(t) + + tasks, err := tbl.NewIncrementalChangelogScan().PlanFiles(context.Background()) + require.NoError(t, err) + require.Len(t, tasks, 5) + + require.Equal(t, "mem://default/changelog-rewrite/data-a.parquet", tasks[0].ScanTask().File.FilePath()) + require.Equal(t, ChangelogOpInsert, tasks[0].Operation()) + require.Equal(t, 0, tasks[0].ChangeOrdinal()) + require.Equal(t, int64(1), tasks[0].CommitSnapshotID()) + + require.Equal(t, "mem://default/changelog-rewrite/data-b.parquet", tasks[1].ScanTask().File.FilePath()) + require.Equal(t, ChangelogOpInsert, tasks[1].Operation()) + require.Equal(t, 1, tasks[1].ChangeOrdinal()) + require.Equal(t, int64(2), tasks[1].CommitSnapshotID()) + + require.Equal(t, "mem://default/changelog-rewrite/data-m.parquet", tasks[2].ScanTask().File.FilePath()) + require.Equal(t, ChangelogOpInsert, tasks[2].Operation()) + require.Equal(t, 1, tasks[2].ChangeOrdinal()) + require.Equal(t, int64(2), tasks[2].CommitSnapshotID()) + + require.Equal(t, "mem://default/changelog-rewrite/data-z.parquet", tasks[3].ScanTask().File.FilePath()) + require.Equal(t, ChangelogOpInsert, tasks[3].Operation()) + require.Equal(t, 1, tasks[3].ChangeOrdinal()) + require.Equal(t, int64(2), tasks[3].CommitSnapshotID()) + + require.Equal(t, "mem://default/changelog-rewrite/data-c.parquet", tasks[4].ScanTask().File.FilePath()) + require.Equal(t, ChangelogOpInsert, tasks[4].Operation()) + require.Equal(t, 2, tasks[4].ChangeOrdinal()) + require.Equal(t, int64(4), tasks[4].CommitSnapshotID()) +} + +func TestIncrementalChangelogScanUsesLiveSchemaForImplicitCurrent(t *testing.T) { + tbl := incrementalAppendSchemaEvolutionTable(t) + filter := iceberg.EqualTo(iceberg.Reference("category"), "new") + + normalTasks, err := tbl.Scan(WithRowFilter(filter)).PlanFiles(context.Background()) + require.NoError(t, err) + require.Len(t, normalTasks, 2) + + incrementalTasks, err := tbl.NewIncrementalChangelogScan(WithRowFilter(filter)).PlanFiles(context.Background()) + require.NoError(t, err) + require.Len(t, incrementalTasks, 2) +} + +func TestIncrementalChangelogScanUsesSnapshotSchemaForExplicitEnd(t *testing.T) { + tbl := incrementalAppendSchemaEvolutionTable(t) + filter := iceberg.EqualTo(iceberg.Reference("category"), "new") + + scan := tbl.NewIncrementalChangelogScan(WithRowFilter(filter)).ToSnapshot(2) + _, err := scan.PlanFiles(context.Background()) + require.Error(t, err) + require.ErrorContains(t, err, "category") +} + +func TestIncrementalChangelogScanAppliesRowFilters(t *testing.T) { + tbl := incrementalChangelogTestTable(t) + filter := iceberg.EqualTo(iceberg.Reference("id"), int32(2)) + + tasks, err := tbl.NewIncrementalChangelogScan( + WithRowFilter(filter), + ).ToSnapshot(4).PlanFiles(context.Background()) + require.NoError(t, err) + require.Len(t, tasks, 1) + require.Equal(t, "mem://default/changelog/data-new.parquet", tasks[0].ScanTask().File.FilePath()) + require.Equal(t, ChangelogOpInsert, tasks[0].Operation()) + require.NotNil(t, tasks[0].ScanTask().Residual) +} + +func TestIncrementalChangelogScanEmitsScanReport(t *testing.T) { + reporter := &metrics.InMemoryReporter{} + tbl := incrementalChangelogTestTable(t) + + tasks, err := tbl.NewIncrementalChangelogScan( + WithSelectedFields("id"), + WithReporter(reporter), + ).PlanFiles(context.Background()) + require.NoError(t, err) + require.Len(t, tasks, 4) + + reports := reporter.Reports() + require.Len(t, reports, 1) + report, ok := reports[0].(metrics.ScanReport) + require.True(t, ok) + require.Equal(t, int64(4), report.SnapshotID) + require.Equal(t, []string{"id"}, report.ProjectedFieldNames) + require.Equal(t, int64(4), report.Metrics.ResultDataFiles.Value) + require.Equal(t, int64(3), report.Metrics.TotalDataManifests.Value) + require.Equal(t, int64(3), report.Metrics.ScannedDataManifests.Value) +} + +func TestIncrementalChangelogScanRejectsDeleteManifests(t *testing.T) { + tbl := incrementalChangelogDeleteManifestTable(t) + + tasks, err := tbl.NewIncrementalChangelogScan().PlanFiles(context.Background()) + require.ErrorIs(t, err, ErrInvalidOperation) + require.ErrorContains(t, err, "scan range references a delete manifest") + require.Nil(t, tasks) +} + +func TestIncrementalChangelogScanRejectsCarriedForwardDeleteManifests(t *testing.T) { + tbl := incrementalChangelogDeleteManifestTable(t) + + tasks, err := tbl.NewIncrementalChangelogScan(). + FromSnapshotExclusive(2). + ToSnapshot(3). + PlanFiles(context.Background()) + require.ErrorIs(t, err, ErrInvalidOperation) + require.ErrorContains(t, err, "scan range references a delete manifest") + require.ErrorContains(t, err, "snapshot 2") + require.Nil(t, tasks) +} + +func TestIncrementalChangelogScanRejectsMissingSnapshotOperation(t *testing.T) { + tbl := incrementalChangelogTestTable(t) + setIncrementalSnapshotSummary(t, tbl, 1, nil) + + _, err := tbl.NewIncrementalChangelogScan().ToSnapshot(1).PlanFiles(context.Background()) + require.ErrorIs(t, err, ErrMissingOperation) + require.ErrorContains(t, err, "cannot determine operation for snapshot 1") +} + +func TestIncrementalChangelogScanAllowsUnknownSnapshotOperation(t *testing.T) { + tbl := incrementalChangelogTestTable(t) + setIncrementalSnapshotSummary(t, tbl, 2, &Summary{Operation: Operation("unknown")}) + + tasks, err := tbl.NewIncrementalChangelogScan().ToSnapshot(4).PlanFiles(context.Background()) + require.NoError(t, err) + require.Len(t, tasks, 4) +} + +func TestIncrementalChangelogScanRejectsUnknownManifestEntryStatus(t *testing.T) { + _, err := changelogOperation(iceberg.ManifestEntryStatus(99)) + require.ErrorIs(t, err, ErrInvalidMetadata) + require.ErrorContains(t, err, "unknown manifest entry status 99") +} + +func TestIncrementalChangelogScanRejectsUnknownStart(t *testing.T) { + tbl := incrementalChangelogTestTable(t) + + _, err := tbl.NewIncrementalChangelogScan(). + FromSnapshotInclusive(999). + ToSnapshot(4). + PlanFiles(context.Background()) + require.ErrorIs(t, err, iceberg.ErrInvalidArgument) + require.ErrorContains(t, err, "starting snapshot not found") +} + +func TestIncrementalChangelogScanRejectsUnknownEnd(t *testing.T) { + tbl := incrementalChangelogTestTable(t) + + _, err := tbl.NewIncrementalChangelogScan().ToSnapshot(999).PlanFiles(context.Background()) + require.ErrorIs(t, err, iceberg.ErrInvalidArgument) + require.ErrorContains(t, err, "ending snapshot not found") +} + +func TestIncrementalChangelogScanAllowsExpiredExclusiveParent(t *testing.T) { + tbl := incrementalAppendExpiredExclusiveTable(t) + + tasks, err := tbl.NewIncrementalChangelogScan(). + FromSnapshotExclusive(1). + ToSnapshot(3). + PlanFiles(context.Background()) + require.NoError(t, err) + require.Len(t, tasks, 2) + require.Equal(t, "mem://default/table-location/data-b.parquet", tasks[0].ScanTask().File.FilePath()) + require.Equal(t, "mem://default/table-location/data-c.parquet", tasks[1].ScanTask().File.FilePath()) +} + +func TestIncrementalChangelogScanRejectsDivergentStart(t *testing.T) { + tbl := incrementalChangelogDivergentTable(t) + + inclusive := tbl.NewIncrementalChangelogScan(). + FromSnapshotInclusive(3). + ToSnapshot(2) + tasks, err := inclusive.PlanFiles(context.Background()) + require.ErrorIs(t, err, iceberg.ErrInvalidArgument) + require.ErrorContains(t, err, "starting snapshot 3 is not an ancestor of ending snapshot 2") + require.Nil(t, tasks) + + exclusive := tbl.NewIncrementalChangelogScan(). + FromSnapshotExclusive(3). + ToSnapshot(2) + tasks, err = exclusive.PlanFiles(context.Background()) + require.ErrorIs(t, err, iceberg.ErrInvalidArgument) + require.ErrorContains(t, err, "starting snapshot 3 is not an ancestor of ending snapshot 2") + require.Nil(t, tasks) +} + +func TestChangelogOperationOrderPlacesDeletesBeforeInserts(t *testing.T) { + operations := []ChangelogOperation{ + ChangelogOpDelete, + ChangelogOpInsert, + ChangelogOpUpdateBefore, + ChangelogOpUpdateAfter, + } + for expected, operation := range operations { + require.Equal(t, expected, changelogOperationOrder(operation)) + } +} + +func TestIncrementalChangelogScanAutoFallsBackToLocal(t *testing.T) { + tbl := incrementalChangelogTestTable(t) + + tasks, err := tbl.NewIncrementalChangelogScan( + WithScanPlanningMode(ScanPlanningAuto), + ).PlanFiles(context.Background()) + require.NoError(t, err) + require.Len(t, tasks, 4) +} + +func TestIncrementalChangelogScanHonorsSnapshotOptions(t *testing.T) { + tbl := incrementalChangelogTestTable(t) + + tasks, err := tbl.NewIncrementalChangelogScan(WithSnapshotID(2)).PlanFiles(context.Background()) + require.NoError(t, err) + require.Len(t, tasks, 3) + for _, task := range tasks { + require.LessOrEqual(t, task.CommitSnapshotID(), int64(2)) + } +} + +func TestIncrementalChangelogScanHonorsContextCancellation(t *testing.T) { + tbl := incrementalChangelogTestTable(t) + var manifestListOpens atomic.Int64 + originalFSF := tbl.fsF + tbl.fsF = func(ctx context.Context) (iceio.IO, error) { + fs, err := originalFSF(ctx) + if err != nil { + return nil, err + } + + return &countingOpenIO{IO: fs, opens: &manifestListOpens}, nil + } + ctx, cancel := context.WithCancel(context.Background()) + cancel() + + _, err := tbl.NewIncrementalChangelogScan().PlanFiles(ctx) + require.ErrorIs(t, err, context.Canceled) + require.Zero(t, manifestListOpens.Load()) +} + +func TestIncrementalChangelogScanChecksContextBetweenSnapshots(t *testing.T) { + tbl := incrementalChangelogTestTable(t) + var manifestListOpens atomic.Int64 + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + originalFSF := tbl.fsF + tbl.fsF = func(ctx context.Context) (iceio.IO, error) { + fs, err := originalFSF(ctx) + if err != nil { + return nil, err + } + + return &countingOpenIO{ + IO: fs, + opens: &manifestListOpens, + afterOpen: cancel, + }, nil + } + + _, err := tbl.NewIncrementalChangelogScan().PlanFiles(ctx) + require.ErrorIs(t, err, context.Canceled) + require.Equal(t, int64(1), manifestListOpens.Load()) +} + +func TestIncrementalChangelogScanZeroValueReturnsError(t *testing.T) { + var scan IncrementalChangelogScan + + tasks, err := scan.PlanFiles(context.Background()) + require.ErrorIs(t, err, ErrInvalidOperation) + require.ErrorContains(t, err, "not initialized") + require.Nil(t, tasks) +} + +func TestIncrementalChangelogScanRejectsUnsupportedPlanningModes(t *testing.T) { + tbl := incrementalChangelogTestTable(t) + + tasks, err := tbl.NewIncrementalChangelogScan( + WithScanPlanningMode(ScanPlanningRemote), + ).PlanFiles(context.Background()) + require.ErrorIs(t, err, ErrInvalidOperation) + require.ErrorContains(t, err, "do not support remote planning") + require.Nil(t, tasks) +} + +func incrementalChangelogDivergentTable(t *testing.T) *Table { + t.Helper() + + base, err := NewMetadata(simpleSchema(), iceberg.UnpartitionedSpec, UnsortedSortOrder, + "mem://default/changelog-divergent", nil) + require.NoError(t, err) + builder, err := MetadataBuilderFromBase(base, "") + require.NoError(t, err) + + baseTimestamp := base.LastUpdatedMillis() + snapshots := []*Snapshot{ + {SnapshotID: 1, TimestampMs: baseTimestamp + 1, SequenceNumber: 1, Summary: &Summary{Operation: OpAppend}}, + {SnapshotID: 2, ParentSnapshotID: int64Ptr(1), TimestampMs: baseTimestamp + 2, SequenceNumber: 2, Summary: &Summary{Operation: OpAppend}}, + {SnapshotID: 3, ParentSnapshotID: int64Ptr(1), TimestampMs: baseTimestamp + 3, SequenceNumber: 3, Summary: &Summary{Operation: OpAppend}}, + } + for _, snapshot := range snapshots { + require.NoError(t, builder.AddSnapshot(snapshot)) + } + require.NoError(t, builder.SetSnapshotRef(MainBranch, 2, BranchRef)) + require.NoError(t, builder.SetSnapshotRef("feature", 3, BranchRef)) + meta, err := builder.Build() + require.NoError(t, err) + + return New(Identifier{"incremental-changelog-divergent"}, meta, "metadata.json", nil, nil) +} + +func incrementalChangelogTestTable(t *testing.T) *Table { + t.Helper() + + spec := partitionedSpec() + txn, fs := createTestTransactionWithMemIO(t, spec) + schema := simpleSchema() + + oldFile := newTestDataFile(t, spec, + "mem://default/changelog/data-old.parquet", map[int]any{1000: int32(1)}) + newFile := newTestDataFile(t, spec, + "mem://default/changelog/data-new.parquet", map[int]any{1000: int32(2)}) + replaceFile := newTestDataFile(t, spec, + "mem://default/changelog/data-replace.parquet", map[int]any{1000: int32(3)}) + laterFile := newTestDataFile(t, spec, + "mem://default/changelog/data-later.parquet", map[int]any{1000: int32(4)}) + + entry := func(status iceberg.ManifestEntryStatus, snapshotID, sequenceNumber int64, file iceberg.DataFile) iceberg.ManifestEntry { + return iceberg.NewManifestEntry(status, &snapshotID, &sequenceNumber, &sequenceNumber, file) + } + writeManifest := func(path string, snapshotID int64, entries []iceberg.ManifestEntry) iceberg.ManifestFile { + var buf bytes.Buffer + manifest, err := iceberg.WriteManifest(path, &buf, 2, spec, schema, snapshotID, entries) + require.NoError(t, err) + require.NoError(t, fs.WriteFile(path, buf.Bytes())) + + return manifest + } + writeManifestList := func(path string, snapshotID int64, manifests []iceberg.ManifestFile) []iceberg.ManifestFile { + var buf bytes.Buffer + sequenceNumber := snapshotID + require.NoError(t, iceberg.WriteManifestList(2, &buf, snapshotID, nil, + &sequenceNumber, 0, manifests)) + require.NoError(t, fs.WriteFile(path, buf.Bytes())) + + listFile, err := fs.Open(path) + require.NoError(t, err) + list, err := iceberg.ReadManifestList(listFile) + require.NoError(t, err) + require.NoError(t, listFile.Close()) + + return list + } + + manifestOne := writeManifest( + "mem://default/changelog/metadata/manifest-1.avro", 1, + []iceberg.ManifestEntry{entry(iceberg.EntryStatusADDED, 1, 1, oldFile)}) + listOnePath := "mem://default/changelog/metadata/snap-1.avro" + listOne := writeManifestList(listOnePath, 1, []iceberg.ManifestFile{manifestOne}) + + manifestTwo := writeManifest( + "mem://default/changelog/metadata/manifest-2.avro", 2, + []iceberg.ManifestEntry{ + entry(iceberg.EntryStatusDELETED, 2, 1, oldFile), + entry(iceberg.EntryStatusADDED, 2, 2, newFile), + }) + listTwoPath := "mem://default/changelog/metadata/snap-2.avro" + listTwo := writeManifestList(listTwoPath, 2, append(listOne, manifestTwo)) + + manifestThree := writeManifest( + "mem://default/changelog/metadata/manifest-3.avro", 3, + []iceberg.ManifestEntry{entry(iceberg.EntryStatusADDED, 3, 3, replaceFile)}) + listThreePath := "mem://default/changelog/metadata/snap-3.avro" + listThree := writeManifestList(listThreePath, 3, append(listTwo, manifestThree)) + + manifestFour := writeManifest( + "mem://default/changelog/metadata/manifest-4.avro", 4, + []iceberg.ManifestEntry{entry(iceberg.EntryStatusADDED, 4, 4, laterFile)}) + existingOnlyFile := newTestDataFile(t, spec, + "mem://default/changelog/existing-only.parquet", map[int]any{1000: int32(5)}) + existingOnlyManifest := writeManifest( + "mem://default/changelog/metadata/manifest-existing-only.avro", 4, + []iceberg.ManifestEntry{entry(iceberg.EntryStatusEXISTING, 1, 1, existingOnlyFile)}, + ) + listFourPath := "mem://default/changelog/metadata/snap-4.avro" + writeManifestList(listFourPath, 4, append(listThree, manifestFour, existingOnlyManifest)) + + txn.meta.snapshotList = []Snapshot{ + {SnapshotID: 1, TimestampMs: 1000, ManifestList: listOnePath, SequenceNumber: 1, SchemaID: &schema.ID, Summary: &Summary{Operation: OpAppend}}, + {SnapshotID: 2, ParentSnapshotID: int64Ptr(1), TimestampMs: 2000, ManifestList: listTwoPath, SequenceNumber: 2, SchemaID: &schema.ID, Summary: &Summary{Operation: OpOverwrite}}, + {SnapshotID: 3, ParentSnapshotID: int64Ptr(2), TimestampMs: 3000, ManifestList: listThreePath, SequenceNumber: 3, SchemaID: &schema.ID, Summary: &Summary{Operation: OpReplace}}, + {SnapshotID: 4, ParentSnapshotID: int64Ptr(3), TimestampMs: 4000, ManifestList: listFourPath, SequenceNumber: 4, SchemaID: &schema.ID, Summary: &Summary{Operation: OpAppend}}, + } + txn.meta.snapshotLog = []SnapshotLogEntry{ + {SnapshotID: 1, TimestampMs: 1000}, + {SnapshotID: 2, TimestampMs: 2000}, + {SnapshotID: 3, TimestampMs: 3000}, + {SnapshotID: 4, TimestampMs: 4000}, + } + currentSnapshotID := int64(4) + txn.meta.currentSnapshotID = ¤tSnapshotID + meta, err := txn.meta.Build() + require.NoError(t, err) + + return New(Identifier{"incremental-changelog"}, meta, "metadata.json", func(context.Context) (iceio.IO, error) { + return fs, nil + }, nil) +} + +func incrementalChangelogManifestRewriteTable(t *testing.T) *Table { + t.Helper() + + spec := partitionedSpec() + txn, fs := createTestTransactionWithMemIO(t, spec) + schema := simpleSchema() + + fileA := newTestDataFile(t, spec, + "mem://default/changelog-rewrite/data-a.parquet", map[int]any{1000: int32(1)}) + fileB := newTestDataFile(t, spec, + "mem://default/changelog-rewrite/data-b.parquet", map[int]any{1000: int32(2)}) + fileM := newTestDataFile(t, spec, + "mem://default/changelog-rewrite/data-m.parquet", map[int]any{1000: int32(13)}) + fileZ := newTestDataFile(t, spec, + "mem://default/changelog-rewrite/data-z.parquet", map[int]any{1000: int32(26)}) + fileC := newTestDataFile(t, spec, + "mem://default/changelog-rewrite/data-c.parquet", map[int]any{1000: int32(3)}) + + entry := func(status iceberg.ManifestEntryStatus, snapshotID, sequenceNumber int64, file iceberg.DataFile) iceberg.ManifestEntry { + return iceberg.NewManifestEntry(status, &snapshotID, &sequenceNumber, &sequenceNumber, file) + } + writeManifest := func(path string, snapshotID int64, entries []iceberg.ManifestEntry) iceberg.ManifestFile { + var buf bytes.Buffer + manifest, err := iceberg.WriteManifest(path, &buf, 2, spec, schema, snapshotID, entries) + require.NoError(t, err) + require.NoError(t, fs.WriteFile(path, buf.Bytes())) + + return manifest + } + writeManifestList := func(path string, snapshotID int64, manifests []iceberg.ManifestFile) []iceberg.ManifestFile { + var buf bytes.Buffer + sequenceNumber := snapshotID + require.NoError(t, iceberg.WriteManifestList(2, &buf, snapshotID, nil, + &sequenceNumber, 0, manifests)) + require.NoError(t, fs.WriteFile(path, buf.Bytes())) + + listFile, err := fs.Open(path) + require.NoError(t, err) + list, err := iceberg.ReadManifestList(listFile) + require.NoError(t, err) + require.NoError(t, listFile.Close()) + + return list + } + + manifestA := writeManifest( + "mem://default/changelog-rewrite/metadata/manifest-a.avro", 1, + []iceberg.ManifestEntry{entry(iceberg.EntryStatusADDED, 1, 1, fileA)}) + listOnePath := "mem://default/changelog-rewrite/metadata/snap-1.avro" + listOne := writeManifestList(listOnePath, 1, []iceberg.ManifestFile{manifestA}) + + manifestB := writeManifest( + "mem://default/changelog-rewrite/metadata/manifest-b.avro", 2, + []iceberg.ManifestEntry{ + // This is the merged-manifest shape produced by append commits: + // an EXISTING entry from an earlier in-range snapshot followed by + // entries added by the manifest's owning snapshot. The added entries + // are deliberately reverse-sorted to exercise the task path tiebreak. + entry(iceberg.EntryStatusEXISTING, 1, 1, fileA), + entry(iceberg.EntryStatusADDED, 2, 2, fileZ), + entry(iceberg.EntryStatusADDED, 2, 2, fileM), + entry(iceberg.EntryStatusADDED, 2, 2, fileB), + }) + listTwoPath := "mem://default/changelog-rewrite/metadata/snap-2.avro" + writeManifestList(listTwoPath, 2, append(listOne, manifestB)) + + manifestRewrite := writeManifest( + "mem://default/changelog-rewrite/metadata/manifest-rewrite.avro", 3, + []iceberg.ManifestEntry{ + entry(iceberg.EntryStatusEXISTING, 1, 1, fileA), + entry(iceberg.EntryStatusEXISTING, 2, 2, fileB), + entry(iceberg.EntryStatusEXISTING, 2, 2, fileZ), + }) + listThreePath := "mem://default/changelog-rewrite/metadata/snap-3.avro" + listThree := writeManifestList(listThreePath, 3, []iceberg.ManifestFile{manifestRewrite}) + + manifestC := writeManifest( + "mem://default/changelog-rewrite/metadata/manifest-c.avro", 4, + []iceberg.ManifestEntry{entry(iceberg.EntryStatusADDED, 4, 4, fileC)}) + listFourPath := "mem://default/changelog-rewrite/metadata/snap-4.avro" + writeManifestList(listFourPath, 4, append(listThree, manifestC)) + + txn.meta.snapshotList = []Snapshot{ + {SnapshotID: 1, TimestampMs: 1000, ManifestList: listOnePath, SequenceNumber: 1, SchemaID: &schema.ID, Summary: &Summary{Operation: OpAppend}}, + {SnapshotID: 2, ParentSnapshotID: int64Ptr(1), TimestampMs: 2000, ManifestList: listTwoPath, SequenceNumber: 2, SchemaID: &schema.ID, Summary: &Summary{Operation: OpAppend}}, + {SnapshotID: 3, ParentSnapshotID: int64Ptr(2), TimestampMs: 3000, ManifestList: listThreePath, SequenceNumber: 3, SchemaID: &schema.ID, Summary: &Summary{Operation: OpReplace}}, + {SnapshotID: 4, ParentSnapshotID: int64Ptr(3), TimestampMs: 4000, ManifestList: listFourPath, SequenceNumber: 4, SchemaID: &schema.ID, Summary: &Summary{Operation: OpAppend}}, + } + txn.meta.snapshotLog = []SnapshotLogEntry{ + {SnapshotID: 1, TimestampMs: 1000}, + {SnapshotID: 2, TimestampMs: 2000}, + {SnapshotID: 3, TimestampMs: 3000}, + {SnapshotID: 4, TimestampMs: 4000}, + } + currentSnapshotID := int64(4) + txn.meta.currentSnapshotID = ¤tSnapshotID + meta, err := txn.meta.Build() + require.NoError(t, err) + + return New(Identifier{"incremental-changelog-rewrite"}, meta, "metadata.json", func(context.Context) (iceio.IO, error) { + return fs, nil + }, nil) +} + +func incrementalChangelogDeleteManifestTable(t *testing.T) *Table { + t.Helper() + + spec := partitionedSpec() + txn, fs := createTestTransactionWithMemIO(t, spec) + schema := simpleSchema() + dataFileOne := newTestDataFile(t, spec, + "mem://default/changelog-delete/data-1.parquet", map[int]any{1000: int32(1)}) + dataFileThree := newTestDataFile(t, spec, + "mem://default/changelog-delete/data-3.parquet", map[int]any{1000: int32(3)}) + deletePath := "mem://default/changelog-delete/delete.parquet" + + deleteFile := newTestPosDeleteFileForSpec(t, spec, deletePath, map[int]any{1000: int32(1)}, dataFileOne.FilePath()) + entry := func(status iceberg.ManifestEntryStatus, snapshotID, sequenceNumber int64, file iceberg.DataFile) iceberg.ManifestEntry { + return iceberg.NewManifestEntry(status, &snapshotID, &sequenceNumber, &sequenceNumber, file) + } + writeDataManifest := func(path string, snapshotID int64, file iceberg.DataFile) iceberg.ManifestFile { + var buf bytes.Buffer + manifest, err := iceberg.WriteManifest(path, &buf, 2, spec, schema, snapshotID, + []iceberg.ManifestEntry{entry(iceberg.EntryStatusADDED, snapshotID, snapshotID, file)}) + require.NoError(t, err) + require.NoError(t, fs.WriteFile(path, buf.Bytes())) + + return manifest + } + writeManifestList := func(path string, snapshotID int64, manifests []iceberg.ManifestFile) []iceberg.ManifestFile { + var buf bytes.Buffer + sequenceNumber := snapshotID + require.NoError(t, iceberg.WriteManifestList(2, &buf, snapshotID, nil, + &sequenceNumber, 0, manifests)) + require.NoError(t, fs.WriteFile(path, buf.Bytes())) + + listFile, err := fs.Open(path) + require.NoError(t, err) + list, err := iceberg.ReadManifestList(listFile) + require.NoError(t, err) + require.NoError(t, listFile.Close()) + + return list + } + + manifestOnePath := "mem://default/changelog-delete/metadata/manifest-1.avro" + manifestOne := writeDataManifest(manifestOnePath, 1, dataFileOne) + listOnePath := "mem://default/changelog-delete/metadata/snap-1.avro" + listOne := writeManifestList(listOnePath, 1, []iceberg.ManifestFile{manifestOne}) + + deleteManifestPath := "mem://default/changelog-delete/metadata/delete-manifest.avro" + deleteSnapshotID := int64(2) + deleteSequenceNumber := int64(2) + deleteEntry := entry(iceberg.EntryStatusADDED, deleteSnapshotID, deleteSequenceNumber, deleteFile) + var deleteManifestBuf bytes.Buffer + writer, err := iceberg.NewManifestWriter(2, &deleteManifestBuf, spec, schema, deleteSnapshotID, + iceberg.WithManifestWriterContent(iceberg.ManifestContentDeletes)) + require.NoError(t, err) + require.NoError(t, writer.Add(deleteEntry)) + require.NoError(t, writer.Close()) + deleteManifest, err := writer.ToManifestFile(deleteManifestPath, int64(deleteManifestBuf.Len()), + iceberg.WithManifestFileContent(iceberg.ManifestContentDeletes)) + require.NoError(t, err) + require.NoError(t, fs.WriteFile(deleteManifestPath, deleteManifestBuf.Bytes())) + + listTwoPath := "mem://default/changelog-delete/metadata/snap-2.avro" + listTwo := writeManifestList(listTwoPath, 2, append(listOne, deleteManifest)) + + manifestThreePath := "mem://default/changelog-delete/metadata/manifest-3.avro" + manifestThree := writeDataManifest(manifestThreePath, 3, dataFileThree) + listThreePath := "mem://default/changelog-delete/metadata/snap-3.avro" + writeManifestList(listThreePath, 3, append(listTwo, manifestThree)) + + txn.meta.snapshotList = []Snapshot{ + {SnapshotID: 1, TimestampMs: 1000, ManifestList: listOnePath, SequenceNumber: 1, SchemaID: &schema.ID, Summary: &Summary{Operation: OpAppend}}, + {SnapshotID: 2, ParentSnapshotID: int64Ptr(1), TimestampMs: 2000, ManifestList: listTwoPath, SequenceNumber: 2, SchemaID: &schema.ID, Summary: &Summary{Operation: OpDelete}}, + {SnapshotID: 3, ParentSnapshotID: int64Ptr(2), TimestampMs: 3000, ManifestList: listThreePath, SequenceNumber: 3, SchemaID: &schema.ID, Summary: &Summary{Operation: OpAppend}}, + } + txn.meta.snapshotLog = []SnapshotLogEntry{ + {SnapshotID: 1, TimestampMs: 1000}, + {SnapshotID: 2, TimestampMs: 2000}, + {SnapshotID: 3, TimestampMs: 3000}, + } + currentSnapshotID := int64(3) + txn.meta.currentSnapshotID = ¤tSnapshotID + meta, err := txn.meta.Build() + require.NoError(t, err) + + return New(Identifier{"incremental-changelog-delete"}, meta, "metadata.json", func(context.Context) (iceio.IO, error) { + return fs, nil + }, nil) +} + +type countingOpenIO struct { + iceio.IO + opens *atomic.Int64 + afterOpen func() +} + +func (io *countingOpenIO) Open(name string) (iceio.File, error) { + io.opens.Add(1) + if io.afterOpen != nil { + io.afterOpen() + } + + return io.IO.Open(name) +} diff --git a/table/incremental_scan.go b/table/incremental_scan.go new file mode 100644 index 000000000..60cbac467 --- /dev/null +++ b/table/incremental_scan.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 + +import ( + "fmt" + "slices" + + "github.com/apache/iceberg-go" +) + +// incrementalSnapshotsBetween returns the snapshots in one ancestry chain +// from oldest to newest, applying the same inclusive/exclusive boundary rules +// to all incremental scan types. An exclusive starting snapshot may have +// expired because AncestorsBetween can validate it by parent ID alone. With no +// explicit starting snapshot, an expired intermediate snapshot makes +// AncestorsOf truncate the chain and the caller receives the remaining +// snapshots with ordinals starting at zero; this preserves the existing +// incremental-scan behavior. +func incrementalSnapshotsBetween( + metadata Metadata, + fromSnapshotID *int64, + fromInclusive bool, + toSnapshotID int64, +) ([]Snapshot, error) { + ancestors := AncestorsOf(toSnapshotID, metadata.SnapshotByID) + if len(ancestors) == 0 { + return nil, fmt.Errorf("%w: ending snapshot not found: %d", iceberg.ErrInvalidArgument, toSnapshotID) + } + + if fromSnapshotID == nil { + slices.Reverse(ancestors) + + return ancestors, nil + } + + fromID := *fromSnapshotID + if !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, 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 between, nil + } + + if metadata.SnapshotByID(fromID) == nil { + return nil, fmt.Errorf("%w: starting snapshot not found: %d", iceberg.ErrInvalidArgument, fromID) + } + if !IsAncestorOf(toSnapshotID, fromID, 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 + } + } + slices.Reverse(selected) + + return selected, nil +} diff --git a/table/scanner.go b/table/scanner.go index 1cd40efb8..70f84e084 100644 --- a/table/scanner.go +++ b/table/scanner.go @@ -382,14 +382,48 @@ func openManifestWithProjection( partitionFilter, metricsEval func(iceberg.DataFile) (bool, error), projection *iceberg.ManifestEntryProjection, dropColumnStats bool, +) ([]iceberg.ManifestEntry, error) { + return openManifestWithReadOptions( + io, manifest, partitionFilter, metricsEval, projection, dropColumnStats, true, false) +} + +func openManifestWithOptions(io io.IO, manifest iceberg.ManifestFile, + partitionFilter, metricsEval func(iceberg.DataFile) (bool, error), discardDeleted, discardExisting bool, +) ([]iceberg.ManifestEntry, error) { + return openManifestWithReadOptions( + io, manifest, partitionFilter, metricsEval, nil, false, discardDeleted, discardExisting) +} + +func openManifestWithReadOptions( + io io.IO, + manifest iceberg.ManifestFile, + partitionFilter, metricsEval func(iceberg.DataFile) (bool, error), + projection *iceberg.ManifestEntryProjection, + dropColumnStats, discardDeleted, discardExisting bool, ) ([]iceberg.ManifestEntry, error) { // Counts may be -1 (unset) on V1 manifests, so clamp before allocating. - out := make([]iceberg.ManifestEntry, 0, max(0, int(manifest.AddedDataFiles())+int(manifest.ExistingDataFiles()))) - if err := streamManifest(io, manifest, partitionFilter, metricsEval, projection, dropColumnStats, func(entry iceberg.ManifestEntry) error { - out = append(out, entry) + capacity := 0 + if added := manifest.AddedDataFiles(); added > 0 { + capacity += int(added) + } + if !discardExisting { + if existing := manifest.ExistingDataFiles(); existing > 0 { + capacity += int(existing) + } + } + if !discardDeleted { + if deleted := manifest.DeletedDataFiles(); deleted > 0 { + capacity += int(deleted) + } + } + out := make([]iceberg.ManifestEntry, 0, capacity) + if err := streamManifestWithReadOptions( + io, manifest, partitionFilter, metricsEval, projection, dropColumnStats, + discardDeleted, discardExisting, func(entry iceberg.ManifestEntry) error { + out = append(out, entry) - return nil - }); err != nil { + return nil + }); err != nil { return nil, err } @@ -406,14 +440,37 @@ func streamManifest(manifestIO io.IO, manifest iceberg.ManifestFile, dropColumnStats bool, visit func(iceberg.ManifestEntry) error, ) error { - entries := manifest.Entries(manifestIO, true) + return streamManifestWithReadOptions( + manifestIO, manifest, partitionFilter, metricsEval, projection, dropColumnStats, + true, false, visit) +} + +func streamManifestWithOptions(manifestIO io.IO, manifest iceberg.ManifestFile, + partitionFilter, metricsEval func(iceberg.DataFile) (bool, error), discardDeleted, discardExisting bool, + visit func(iceberg.ManifestEntry) error, +) error { + return streamManifestWithReadOptions( + manifestIO, manifest, partitionFilter, metricsEval, nil, false, + discardDeleted, discardExisting, visit) +} + +func streamManifestWithReadOptions(manifestIO io.IO, manifest iceberg.ManifestFile, + partitionFilter, metricsEval func(iceberg.DataFile) (bool, error), + projection *iceberg.ManifestEntryProjection, + dropColumnStats, discardDeleted, discardExisting bool, + visit func(iceberg.ManifestEntry) error, +) error { + entries := manifest.Entries(manifestIO, discardDeleted) if projection != nil { - entries = iceberg.EntriesWithProjection(manifestIO, manifest, true, *projection) + entries = iceberg.EntriesWithProjection(manifestIO, manifest, discardDeleted, *projection) } for entry, err := range entries { if err != nil { return err } + if discardExisting && entry.Status() == iceberg.EntryStatusEXISTING { + continue + } dataFile := entry.DataFile() use, err := partitionFilter(dataFile) @@ -1005,6 +1062,20 @@ func (scan *Scan) filterManifestsWithSchema( schema *iceberg.Schema, acc *scanMetricsAccumulator, partitionFilters *keyDefaultMapErr[int, iceberg.BooleanExpression], +) ([]iceberg.ManifestFile, error) { + return scan.filterManifestsWithSchemaOptions( + manifestList, schema, acc, partitionFilters, false) +} + +// filterManifestsWithSchemaOptions is filterManifestsWithSchema with an +// option for changelog scans, which must retain data manifests containing +// deleted entries even when they have no live entries. +func (scan *Scan) filterManifestsWithSchemaOptions( + manifestList []iceberg.ManifestFile, + schema *iceberg.Schema, + acc *scanMetricsAccumulator, + partitionFilters *keyDefaultMapErr[int, iceberg.BooleanExpression], + includeDeleted bool, ) ([]iceberg.ManifestFile, error) { // Build per-spec manifest evaluators and filter out irrelevant manifests. manifestEvaluators := newKeyDefaultMapWrapErr(func(specID int) (func(iceberg.ManifestFile) (bool, error), error) { @@ -1028,8 +1099,12 @@ func (scan *Scan) filterManifestsWithSchema( return nil, fmt.Errorf("failed to evaluate manifest %s: %w", mf.FilePath(), err) } // Has*Files returns true for unknown counts, so this only skips manifests - // known to contain no added or existing (live) entries. - if use && !mf.HasAddedFiles() && !mf.HasExistingFiles() { + // known to contain no added or existing (live) entries. The deleted-file + // count follows the same rule: V1's -1 means unknown, while zero is the + // only known empty value. Changelog scans also retain manifests known to + // contain deleted entries. + if use && !mf.HasAddedFiles() && !mf.HasExistingFiles() && + (!includeDeleted || mf.DeletedDataFiles() == 0) { if isDelete { acc.skippedDeleteManifests++ } else { @@ -1127,6 +1202,31 @@ func (scan *Scan) collectManifestEntriesWithSchemaMinSequenceNum( partitionFilters *keyDefaultMapErr[int, iceberg.BooleanExpression], minSeqNum int64, projectScanColumns bool, +) (*manifestEntries, error) { + return scan.collectManifestEntriesWithSchemaOptionsAndMinSequenceNum( + ctx, manifestList, schema, partitionFilters, minSeqNum, projectScanColumns, true, false) +} + +func (scan *Scan) collectManifestEntriesWithSchemaOptions( + ctx context.Context, + manifestList []iceberg.ManifestFile, + schema *iceberg.Schema, + partitionFilters *keyDefaultMapErr[int, iceberg.BooleanExpression], + discardDeleted, discardExisting bool, +) (*manifestEntries, error) { + return scan.collectManifestEntriesWithSchemaOptionsAndMinSequenceNum( + ctx, manifestList, schema, partitionFilters, + minSequenceNum(manifestList), false, discardDeleted, discardExisting) +} + +func (scan *Scan) collectManifestEntriesWithSchemaOptionsAndMinSequenceNum( + ctx context.Context, + manifestList []iceberg.ManifestFile, + schema *iceberg.Schema, + partitionFilters *keyDefaultMapErr[int, iceberg.BooleanExpression], + minSeqNum int64, + projectScanColumns bool, + discardDeleted, discardExisting bool, ) (*manifestEntries, error) { metricsEval, err := newInclusiveMetricsEvaluator( schema, @@ -1169,8 +1269,8 @@ func (scan *Scan) collectManifestEntriesWithSchemaMinSequenceNum( // Keep pruning stats until equality and positional deletes are indexed. projection = &iceberg.ManifestEntryProjection{IncludePruningStats: true} } - manifestEntries, err := openManifestWithProjection( - fs, mf, partEval, metricsEval, projection, false) + manifestEntries, err := openManifestWithReadOptions( + fs, mf, partEval, metricsEval, projection, false, discardDeleted, discardExisting) if err != nil { return err }