laskoviymishka commented on code in PR #2171: URL: https://github.com/apache/iceberg-go/pull/2171#discussion_r4241786984
########## table/incremental_changelog_deletes.go: ########## @@ -0,0 +1,658 @@ +// 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" + "slices" + + "github.com/apache/iceberg-go" + "github.com/apache/iceberg-go/io" +) + +// changelogTaskPlan is the snapshot range and the data-file changes already +// read for one changelog plan. +type changelogTaskPlan struct { + scan *Scan + fs io.IO + snapshots []Snapshot + manifestsBySnapshot map[int64][]iceberg.ManifestFile + deleteManifests map[int64][]iceberg.ManifestFile + dataEntries []iceberg.ManifestEntry + snapshotOrdinals map[int64]int + schema *iceberg.Schema + partitionFilters *keyDefaultMapErr[int, iceberg.BooleanExpression] + residual iceberg.BooleanExpression + metrics *scanMetricsAccumulator +} + +// changelogDeleteIndex matches delete files to a data file with the same +// sequence rules as a normal scan. +type changelogDeleteIndex struct { + pos *positionalDeleteIndex + dv map[string]iceberg.ManifestEntry + eq *equalityDeleteIndex +} + +// changelogDeleteState tracks delete files added and removed inside the +// changelog range, plus delete files that were already live before it. +type changelogDeleteState struct { + active bool + snapshots []Snapshot + preRange []iceberg.ManifestEntry + added map[int64][]iceberg.ManifestEntry + removed map[int64]map[string]struct{} + specs partitionSpecLookup + schema *iceberg.Schema + + addedCache map[int64]*changelogDeleteIndex + beforeCache map[int64]*changelogDeleteIndex +} + +func (s *IncrementalChangelogScan) planChangelogTasks(ctx context.Context, plan changelogTaskPlan) ([]plannedChangelogTask, error) { + state, err := s.loadChangelogDeleteState(ctx, plan) + if err != nil { + return nil, err + } + + changedPaths := make(map[int64]map[string]struct{}) + tasks := make([]plannedChangelogTask, 0, len(plan.dataEntries)) + for _, entry := range plan.dataEntries { + ordinal, ok := plan.snapshotOrdinals[entry.SnapshotID()] + if !ok { + continue + } + paths := changedPaths[entry.SnapshotID()] + if paths == nil { + paths = make(map[string]struct{}) + changedPaths[entry.SnapshotID()] = paths + } + paths[entry.DataFile().FilePath()] = struct{}{} + + deletes, err := state.deletesFor(entry) + if err != nil { + return nil, fmt.Errorf("incremental changelog scan snapshot %d: %w", entry.SnapshotID(), err) + } + task, err := newChangelogScanTask(entry, ordinal, plan.residual, deletes) + if err != nil { + return nil, fmt.Errorf("incremental changelog scan snapshot %d: %w", entry.SnapshotID(), err) + } + tasks = append(tasks, plannedChangelogTask{task: task}) + } + + rowTasks, err := s.planDeletedRowTasks(ctx, plan, state, changedPaths) + if err != nil { + return nil, err + } + tasks = append(tasks, rowTasks...) + sortChangelogTasks(tasks) + + return tasks, nil +} + +func (s *IncrementalChangelogScan) loadChangelogDeleteState(ctx context.Context, plan changelogTaskPlan) (*changelogDeleteState, error) { + state := &changelogDeleteState{ + snapshots: plan.snapshots, + added: make(map[int64][]iceberg.ManifestEntry), + removed: make(map[int64]map[string]struct{}), + specs: plan.scan.metadata, + schema: plan.schema, + } + for _, snapshot := range plan.snapshots { + if err := ctx.Err(); err != nil { + return nil, err + } + manifests := plan.deleteManifests[snapshot.SnapshotID] + if len(manifests) == 0 { + continue + } + filtered, err := plan.scan.filterManifestsWithSchemaOptions( + manifests, plan.schema, plan.metrics, plan.partitionFilters, true) + if err != nil { + return nil, err + } + entries, err := readManifestEntries(ctx, plan.fs, filtered, false, false) + if err != nil { + return nil, err + } + for _, entry := range entries { + if entry.SnapshotID() != snapshot.SnapshotID { + continue + } + if entry.Status() == iceberg.EntryStatusDELETED { + if state.removed[snapshot.SnapshotID] == nil { + state.removed[snapshot.SnapshotID] = make(map[string]struct{}) + } + state.removed[snapshot.SnapshotID][entry.DataFile().FilePath()] = struct{}{} Review Comment: We key delete removals by `DataFile().FilePath()`, but for a deletion vector that's the Puffin path, not a per-DV identity. One Puffin file can hold DV blobs for many data files (that's what Spark's `BaseDVFileWriter` does), so a path alone can't tell them apart. Concretely: Puffin P holds DVs for data files A and B. A snapshot rewrites A's DV into Puffin Q and marks the P entry DELETED. `removed[S]` now contains P, so `entriesBefore` drops B's still-live DV out of the existing set too. When a later snapshot adds a delete on B, B's earlier-deleted rows are no longer in `existing` and get re-emitted as DELETE. That's the double-emission this PR exists to prevent, and it's silent. I'd key DV identity by `(path, referenced_data_file)` (or content offset) and keep the plain path key only for position and equality delete files. `keepNewestDeletionVector` in this same file already keys by referenced data file, so there's a pattern to follow. ########## table/incremental_changelog_deletes.go: ########## @@ -0,0 +1,658 @@ +// 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" + "slices" + + "github.com/apache/iceberg-go" + "github.com/apache/iceberg-go/io" +) + +// changelogTaskPlan is the snapshot range and the data-file changes already +// read for one changelog plan. +type changelogTaskPlan struct { + scan *Scan + fs io.IO + snapshots []Snapshot + manifestsBySnapshot map[int64][]iceberg.ManifestFile + deleteManifests map[int64][]iceberg.ManifestFile + dataEntries []iceberg.ManifestEntry + snapshotOrdinals map[int64]int + schema *iceberg.Schema + partitionFilters *keyDefaultMapErr[int, iceberg.BooleanExpression] + residual iceberg.BooleanExpression + metrics *scanMetricsAccumulator +} + +// changelogDeleteIndex matches delete files to a data file with the same +// sequence rules as a normal scan. +type changelogDeleteIndex struct { + pos *positionalDeleteIndex + dv map[string]iceberg.ManifestEntry + eq *equalityDeleteIndex +} + +// changelogDeleteState tracks delete files added and removed inside the +// changelog range, plus delete files that were already live before it. +type changelogDeleteState struct { + active bool + snapshots []Snapshot + preRange []iceberg.ManifestEntry + added map[int64][]iceberg.ManifestEntry + removed map[int64]map[string]struct{} + specs partitionSpecLookup + schema *iceberg.Schema + + addedCache map[int64]*changelogDeleteIndex + beforeCache map[int64]*changelogDeleteIndex +} + +func (s *IncrementalChangelogScan) planChangelogTasks(ctx context.Context, plan changelogTaskPlan) ([]plannedChangelogTask, error) { + state, err := s.loadChangelogDeleteState(ctx, plan) + if err != nil { + return nil, err + } + + changedPaths := make(map[int64]map[string]struct{}) + tasks := make([]plannedChangelogTask, 0, len(plan.dataEntries)) + for _, entry := range plan.dataEntries { + ordinal, ok := plan.snapshotOrdinals[entry.SnapshotID()] + if !ok { + continue + } + paths := changedPaths[entry.SnapshotID()] + if paths == nil { + paths = make(map[string]struct{}) + changedPaths[entry.SnapshotID()] = paths + } + paths[entry.DataFile().FilePath()] = struct{}{} + + deletes, err := state.deletesFor(entry) + if err != nil { + return nil, fmt.Errorf("incremental changelog scan snapshot %d: %w", entry.SnapshotID(), err) + } + task, err := newChangelogScanTask(entry, ordinal, plan.residual, deletes) + if err != nil { + return nil, fmt.Errorf("incremental changelog scan snapshot %d: %w", entry.SnapshotID(), err) + } + tasks = append(tasks, plannedChangelogTask{task: task}) + } + + rowTasks, err := s.planDeletedRowTasks(ctx, plan, state, changedPaths) + if err != nil { + return nil, err + } + tasks = append(tasks, rowTasks...) + sortChangelogTasks(tasks) + + return tasks, nil +} + +func (s *IncrementalChangelogScan) loadChangelogDeleteState(ctx context.Context, plan changelogTaskPlan) (*changelogDeleteState, error) { + state := &changelogDeleteState{ + snapshots: plan.snapshots, + added: make(map[int64][]iceberg.ManifestEntry), + removed: make(map[int64]map[string]struct{}), + specs: plan.scan.metadata, + schema: plan.schema, + } + for _, snapshot := range plan.snapshots { + if err := ctx.Err(); err != nil { + return nil, err + } + manifests := plan.deleteManifests[snapshot.SnapshotID] + if len(manifests) == 0 { + continue + } + filtered, err := plan.scan.filterManifestsWithSchemaOptions( + manifests, plan.schema, plan.metrics, plan.partitionFilters, true) + if err != nil { + return nil, err + } + entries, err := readManifestEntries(ctx, plan.fs, filtered, false, false) + if err != nil { + return nil, err + } + for _, entry := range entries { + if entry.SnapshotID() != snapshot.SnapshotID { + continue + } + if entry.Status() == iceberg.EntryStatusDELETED { + if state.removed[snapshot.SnapshotID] == nil { + state.removed[snapshot.SnapshotID] = make(map[string]struct{}) + } + state.removed[snapshot.SnapshotID][entry.DataFile().FilePath()] = struct{}{} + + continue + } + state.added[snapshot.SnapshotID] = append(state.added[snapshot.SnapshotID], entry) + } + } + + deletedData := false + for _, entry := range plan.dataEntries { + if entry.Status() == iceberg.EntryStatusDELETED { + deletedData = true + + break + } + } + for _, entries := range state.added { + if len(entries) > 0 { + state.active = true + + break + } + } + if len(state.removed) > 0 || deletedData { + state.active = true + } + if !state.active { + return state, nil + } + + preRange, err := s.liveDeletesBeforeRange(ctx, plan.fs, plan.snapshots) + if err != nil { + return nil, err + } + state.preRange = preRange + + return state, nil +} + +// liveDeletesBeforeRange returns delete files that were already live on the +// snapshot immediately before the changelog range. +func (s *IncrementalChangelogScan) liveDeletesBeforeRange(ctx context.Context, fs io.IO, snapshots []Snapshot) ([]iceberg.ManifestEntry, error) { + parent := s.snapshotBeforeRange(snapshots) + if parent == nil { + return nil, nil Review Comment: When the exclusive `fromSnapshotID` (or the first in-range snapshot's parent) has been expired, `snapshotBeforeRange` returns nil via `SnapshotByID`, and we treat that as "no pre-range deletes" and return empty. `TestIncrementalChangelogScanAllowsExpiredExclusiveParent` deliberately allows the expired case, so this path is reachable. The consequence is the same silent re-emission as the DV case: any delete that was live before the range is now invisible, so when an in-range snapshot adds a delete on that same data file, rows deleted before the range come back out as DELETE, with no error for the caller to catch. If a parent ID exists but the snapshot is gone and in-range snapshots add deletes (`state.active`), I'd return an explicit error rather than silently approximate. If the lossy behavior is actually intended, let's document it and cover it with a test that asserts the degraded result. ########## table/incremental_changelog_deletes.go: ########## @@ -0,0 +1,658 @@ +// 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" + "slices" + + "github.com/apache/iceberg-go" + "github.com/apache/iceberg-go/io" +) + +// changelogTaskPlan is the snapshot range and the data-file changes already +// read for one changelog plan. +type changelogTaskPlan struct { + scan *Scan + fs io.IO + snapshots []Snapshot + manifestsBySnapshot map[int64][]iceberg.ManifestFile + deleteManifests map[int64][]iceberg.ManifestFile + dataEntries []iceberg.ManifestEntry + snapshotOrdinals map[int64]int + schema *iceberg.Schema + partitionFilters *keyDefaultMapErr[int, iceberg.BooleanExpression] + residual iceberg.BooleanExpression + metrics *scanMetricsAccumulator +} + +// changelogDeleteIndex matches delete files to a data file with the same +// sequence rules as a normal scan. +type changelogDeleteIndex struct { + pos *positionalDeleteIndex + dv map[string]iceberg.ManifestEntry + eq *equalityDeleteIndex +} + +// changelogDeleteState tracks delete files added and removed inside the +// changelog range, plus delete files that were already live before it. +type changelogDeleteState struct { + active bool + snapshots []Snapshot + preRange []iceberg.ManifestEntry + added map[int64][]iceberg.ManifestEntry + removed map[int64]map[string]struct{} + specs partitionSpecLookup + schema *iceberg.Schema + + addedCache map[int64]*changelogDeleteIndex + beforeCache map[int64]*changelogDeleteIndex +} + +func (s *IncrementalChangelogScan) planChangelogTasks(ctx context.Context, plan changelogTaskPlan) ([]plannedChangelogTask, error) { + state, err := s.loadChangelogDeleteState(ctx, plan) + if err != nil { + return nil, err + } + + changedPaths := make(map[int64]map[string]struct{}) + tasks := make([]plannedChangelogTask, 0, len(plan.dataEntries)) + for _, entry := range plan.dataEntries { + ordinal, ok := plan.snapshotOrdinals[entry.SnapshotID()] + if !ok { + continue + } + paths := changedPaths[entry.SnapshotID()] + if paths == nil { + paths = make(map[string]struct{}) + changedPaths[entry.SnapshotID()] = paths + } + paths[entry.DataFile().FilePath()] = struct{}{} + + deletes, err := state.deletesFor(entry) + if err != nil { + return nil, fmt.Errorf("incremental changelog scan snapshot %d: %w", entry.SnapshotID(), err) + } + task, err := newChangelogScanTask(entry, ordinal, plan.residual, deletes) + if err != nil { + return nil, fmt.Errorf("incremental changelog scan snapshot %d: %w", entry.SnapshotID(), err) + } + tasks = append(tasks, plannedChangelogTask{task: task}) + } + + rowTasks, err := s.planDeletedRowTasks(ctx, plan, state, changedPaths) + if err != nil { + return nil, err + } + tasks = append(tasks, rowTasks...) + sortChangelogTasks(tasks) + + return tasks, nil +} + +func (s *IncrementalChangelogScan) loadChangelogDeleteState(ctx context.Context, plan changelogTaskPlan) (*changelogDeleteState, error) { + state := &changelogDeleteState{ + snapshots: plan.snapshots, + added: make(map[int64][]iceberg.ManifestEntry), + removed: make(map[int64]map[string]struct{}), + specs: plan.scan.metadata, + schema: plan.schema, + } + for _, snapshot := range plan.snapshots { + if err := ctx.Err(); err != nil { + return nil, err + } + manifests := plan.deleteManifests[snapshot.SnapshotID] + if len(manifests) == 0 { + continue + } + filtered, err := plan.scan.filterManifestsWithSchemaOptions( + manifests, plan.schema, plan.metrics, plan.partitionFilters, true) + if err != nil { + return nil, err + } + entries, err := readManifestEntries(ctx, plan.fs, filtered, false, false) + if err != nil { + return nil, err + } + for _, entry := range entries { + if entry.SnapshotID() != snapshot.SnapshotID { + continue + } + if entry.Status() == iceberg.EntryStatusDELETED { + if state.removed[snapshot.SnapshotID] == nil { + state.removed[snapshot.SnapshotID] = make(map[string]struct{}) + } + state.removed[snapshot.SnapshotID][entry.DataFile().FilePath()] = struct{}{} + + continue + } + state.added[snapshot.SnapshotID] = append(state.added[snapshot.SnapshotID], entry) + } + } + + deletedData := false + for _, entry := range plan.dataEntries { + if entry.Status() == iceberg.EntryStatusDELETED { + deletedData = true + + break + } + } + for _, entries := range state.added { + if len(entries) > 0 { + state.active = true + + break + } + } + if len(state.removed) > 0 || deletedData { + state.active = true + } + if !state.active { + return state, nil + } + + preRange, err := s.liveDeletesBeforeRange(ctx, plan.fs, plan.snapshots) + if err != nil { + return nil, err + } + state.preRange = preRange + + return state, nil +} + +// liveDeletesBeforeRange returns delete files that were already live on the +// snapshot immediately before the changelog range. +func (s *IncrementalChangelogScan) liveDeletesBeforeRange(ctx context.Context, fs io.IO, snapshots []Snapshot) ([]iceberg.ManifestEntry, error) { + parent := s.snapshotBeforeRange(snapshots) + if parent == nil { + return nil, nil + } + manifests, err := parent.Manifests(fs) + if err != nil { + return nil, err + } + deleteManifests := make([]iceberg.ManifestFile, 0) + for _, manifest := range manifests { + if manifest.ManifestContent() == iceberg.ManifestContentDeletes { + deleteManifests = append(deleteManifests, manifest) + } + } + entries, err := readManifestEntries(ctx, fs, deleteManifests, false, false) + if err != nil { + return nil, err + } + + return liveManifestEntries(entries), nil +} + +func (s *IncrementalChangelogScan) snapshotBeforeRange(snapshots []Snapshot) *Snapshot { + if s.fromSnapshotID != nil && !s.fromInclusive { + return s.scan.metadata.SnapshotByID(*s.fromSnapshotID) + } + if len(snapshots) == 0 || snapshots[0].ParentSnapshotID == nil { + return nil + } + + return s.scan.metadata.SnapshotByID(*snapshots[0].ParentSnapshotID) +} + +func (s *IncrementalChangelogScan) planDeletedRowTasks( + ctx context.Context, + plan changelogTaskPlan, + state *changelogDeleteState, + changedPaths map[int64]map[string]struct{}, +) ([]plannedChangelogTask, error) { + if state == nil || !state.active { + return nil, nil + } + matcher, err := plan.scan.dataFileMatcher(plan.schema, plan.partitionFilters) + if err != nil { + return nil, err + } + + var tasks []plannedChangelogTask + for _, snapshot := range plan.snapshots { + if err := ctx.Err(); err != nil { + return nil, err + } + if len(state.added[snapshot.SnapshotID]) == 0 { + continue + } + addedIndex, err := state.addedIndex(snapshot.SnapshotID) + if err != nil { + return nil, fmt.Errorf("incremental changelog scan snapshot %d: %w", snapshot.SnapshotID, err) + } + manifests := plan.manifestsBySnapshot[snapshot.SnapshotID] + dataManifests := make([]iceberg.ManifestFile, 0, len(manifests)) + for _, manifest := range manifests { + if manifest.ManifestContent() == iceberg.ManifestContentData { + dataManifests = append(dataManifests, manifest) + } + } + // Prune with the scan partition filters before opening manifests. + // A local accumulator keeps carried-forward data manifests out of the + // changelog scan report, which already counted them. + var pruned scanMetricsAccumulator + dataManifests, err = plan.scan.filterManifestsWithSchemaOptions( + dataManifests, plan.schema, &pruned, plan.partitionFilters, true) + if err != nil { + return nil, err + } + live, err := liveDataEntries(ctx, plan.fs, dataManifests) Review Comment: For every snapshot that adds deletes we read all live data manifests for the whole table and run the matcher over every live file, narrowing only by partition filter when the user supplies one. A streaming consumer reading many small ranges against a large table pays O(table files) per delete-adding snapshot; Java narrows candidates using the delete file's partition and referenced paths. This isn't a blocker, but I'd really like to see us narrow candidates up front from the added deletes (referenced data file for DVs, `file_path` bounds for position deletes) before opening data manifests, falling back to a full scan only for equality and unreferenced position deletes. At minimum, let's document the cost. ########## table/changelog_reader_external_test.go: ########## @@ -0,0 +1,596 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package table_test + +import ( + "context" + "fmt" + "iter" + "os" + "path/filepath" + "strconv" + "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" + iceio "github.com/apache/iceberg-go/io" + "github.com/apache/iceberg-go/table" + "github.com/stretchr/testify/require" +) + +func TestIncrementalChangelogReadEmptyTable(t *testing.T) { + schema, records, err := newV2RowLineageTestTable(t).NewIncrementalChangelogScan().ToArrowRecords(t.Context()) + require.NoError(t, err) + require.Equal(t, []string{ + "id", "data", + table.ChangelogChangeTypeColumn, + table.ChangelogChangeOrdinalColumn, + table.ChangelogCommitSnapshotIDColumn, + }, schemaFieldNames(schema)) + + require.Empty(t, collectChangelogRows(t, records)) +} + +func TestIncrementalChangelogReadFileChanges(t *testing.T) { + ctx := t.Context() + tbl := changelogOverwriteTable(t, ctx) + + tasks, err := tbl.NewIncrementalChangelogScan().PlanFiles(ctx) + require.NoError(t, err) + require.Len(t, tasks, 3) + + schema, records, err := tbl.NewIncrementalChangelogScan().Read(ctx, tasks) + require.NoError(t, err) + require.Equal(t, []string{ + "id", "data", + table.ChangelogChangeTypeColumn, + table.ChangelogChangeOrdinalColumn, + table.ChangelogCommitSnapshotIDColumn, + }, schemaFieldNames(schema)) + + got := collectChangelogRows(t, records) + require.Equal(t, []changelogRow{ + {id: 1, data: "a", op: string(table.ChangelogOpInsert), ordinal: int64(tasks[0].ChangeOrdinal()), snapshot: tasks[0].CommitSnapshotID()}, + {id: 1, data: "a", op: string(table.ChangelogOpDelete), ordinal: int64(tasks[1].ChangeOrdinal()), snapshot: tasks[1].CommitSnapshotID()}, + {id: 2, data: "b", op: string(table.ChangelogOpInsert), ordinal: int64(tasks[2].ChangeOrdinal()), snapshot: tasks[2].CommitSnapshotID()}, + }, got) + require.NotEqual(t, tasks[0].CommitSnapshotID(), tasks[1].CommitSnapshotID()) + require.Equal(t, tasks[1].CommitSnapshotID(), tasks[2].CommitSnapshotID()) + require.Equal(t, 0, tasks[0].ChangeOrdinal()) + require.Equal(t, 1, tasks[1].ChangeOrdinal()) + require.Equal(t, 1, tasks[2].ChangeOrdinal()) +} + +func TestIncrementalChangelogReadHonorsProjectionFilterAndLimit(t *testing.T) { + ctx := t.Context() + tbl := changelogOverwriteTable(t, ctx) + + schema, records, err := tbl.NewIncrementalChangelogScan( + table.WithSelectedFields("id"), + table.WithRowFilter(iceberg.EqualTo(iceberg.Reference("id"), int64(2))), + table.WithLimit(1), + ).ToArrowRecords(ctx) + require.NoError(t, err) + require.Equal(t, []string{ + "id", + table.ChangelogChangeTypeColumn, + table.ChangelogChangeOrdinalColumn, + table.ChangelogCommitSnapshotIDColumn, + }, schemaFieldNames(schema)) + + got := collectChangelogRows(t, records) + require.Equal(t, []changelogRow{ + {id: 2, op: string(table.ChangelogOpInsert), ordinal: 1}, + }, clearChangelogSnapshot(got)) +} + +func changelogOverwriteTable(t *testing.T, ctx context.Context) *table.Table { + t.Helper() + + mem := memory.DefaultAllocator + tbl := newV3RowLineageTestTable(t) + arrowSchema := arrow.NewSchema([]arrow.Field{ + {Name: "id", Type: arrow.PrimitiveTypes.Int64, Nullable: false}, + {Name: "data", Type: arrow.BinaryTypes.String, Nullable: true}, + }, nil) + + initial, err := array.TableFromJSON(mem, arrowSchema, []string{ + `[{"id": 1, "data": "a"}]`, + }) + require.NoError(t, err) + t.Cleanup(initial.Release) + tbl, err = tbl.Append(ctx, array.NewTableReader(initial, -1), nil) + require.NoError(t, err) + + replacement, err := array.TableFromJSON(mem, arrowSchema, []string{ + `[{"id": 2, "data": "b"}]`, + }) + require.NoError(t, err) + t.Cleanup(replacement.Release) + tbl, err = tbl.Overwrite(ctx, array.NewTableReader(replacement, -1), nil, table.WithOverwriteConcurrency(1)) + require.NoError(t, err) + + return tbl +} + +type changelogRow struct { + id int64 + data string + op string + ordinal int64 + snapshot int64 +} + +func collectChangelogRows(t *testing.T, records iter.Seq2[arrow.RecordBatch, error]) []changelogRow { + t.Helper() + + var got []changelogRow + for rec, err := range records { + require.NoError(t, err) + + got = append(got, changelogRowsFromBatch(rec)...) + rec.Release() + } + + return got +} + +func changelogRowsFromBatch(rec arrow.RecordBatch) []changelogRow { + ids := rec.Column(0).(*array.Int64) + opIdx := 1 + var data *array.String + if rec.NumCols() == 5 { + data = rec.Column(1).(*array.String) + opIdx = 2 + } + ops := rec.Column(opIdx).(*array.String) + ordinals := rec.Column(opIdx + 1).(*array.Int64) + snapshots := rec.Column(opIdx + 2).(*array.Int64) + + rows := make([]changelogRow, 0, rec.NumRows()) + for i := range int(rec.NumRows()) { + row := changelogRow{ + id: ids.Value(i), + op: ops.Value(i), + ordinal: ordinals.Value(i), + snapshot: snapshots.Value(i), + } + if data != nil && !data.IsNull(i) { + row.data = data.Value(i) + } + rows = append(rows, row) + } + + return rows +} + +func schemaFieldNames(schema *arrow.Schema) []string { + names := make([]string, schema.NumFields()) + for i := range names { + names[i] = schema.Field(i).Name + } + + return names +} + +func 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.AddFiles(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)) +} + +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 TestIncrementalChangelogReadReplacedDeletionVector(t *testing.T) { Review Comment: Every DV test here uses one DV per Puffin file, so the shared-Puffin collision in `incremental_changelog_deletes.go` can't surface. I'd add a sibling to this test with two DVs in one Puffin file where one is replaced and the other must stay in `existing`, plus an expired-parent case with a pre-range delete. Those two are the likeliest to break the exactly-once guarantee, so they're the ones worth pinning. ########## table/changelog_deleted_rows.go: ########## @@ -0,0 +1,291 @@ +// 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" + +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.addedDeletes} + as.addedPosLoader = newLazyPositionDeleteLoader(as.fs, tasks) + + dvLoader, err := newLazyDeletionVectorLoader(as.fs, tasks) + if err != nil { + as.addedPosLoader.release() + as.addedPosLoader = nil + + return err + } + as.addedDVLoader = dvLoader + + eqLoader, err := newLazyEqualityDeleteLoader( + as.fs, invariants.tableSchema, nil, invariants.nameMapping, tasks) + if err != nil { + as.addedPosLoader.release() Review Comment: If `newLazyEqualityDeleteLoader` fails here, we release `addedPosLoader` but `addedDVLoader` is already assigned and stays set, so its cached state never gets released and the scan is left half-initialized (`addedPosLoader` nil, `addedDVLoader` non-nil). A later `loadAddedDeletes` on that scan then derefs the three loader fields unconditionally. The same gap is in `GetRecords`: the `positionDeleteLoader` built just before the `prepareAddedDeleteLoaders` call isn't released when prepare returns an error. I'd build the three loaders into locals and assign the fields only on full success, so the error path has nothing to unwind. A single cleanup helper that releases everything created so far would do it too. ########## table/changelog_reader.go: ########## @@ -0,0 +1,297 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package table + +import ( + "context" + "fmt" + "iter" + "maps" + "strconv" + "sync/atomic" + + "github.com/apache/arrow-go/v18/arrow" + "github.com/apache/arrow-go/v18/arrow/array" + "github.com/apache/arrow-go/v18/arrow/compute" + "github.com/apache/arrow-go/v18/arrow/memory" + "github.com/apache/iceberg-go" +) + +const ( + // ChangelogChangeTypeColumn is the changelog operation: INSERT, DELETE, + // UPDATE_BEFORE, or UPDATE_AFTER. + ChangelogChangeTypeColumn = "_change_type" + // ChangelogChangeOrdinalColumn is the zero-based ordinal of the snapshot + // that produced the change, shared by every row from that snapshot. + ChangelogChangeOrdinalColumn = "_change_ordinal" + // ChangelogCommitSnapshotIDColumn is the snapshot ID that committed the change. + ChangelogCommitSnapshotIDColumn = "_commit_snapshot_id" +) + +// ToArrowRecords plans the changelog and reads it. Added and removed data +// files are read with the deletes already stored on their scan 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 { + return nil, nil, err + } + + return s.Read(ctx, tasks) +} + +// Read materializes changelog rows for tasks. Each batch has the scan +// projection followed by _change_type, _change_ordinal, and +// _commit_snapshot_id. 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 emits rows matched by AddedDeletes and skips +// rows matched by ExistingDeletes. The returned iterator is single-use. The +// caller releases every batch. +func (s *IncrementalChangelogScan) Read(ctx context.Context, tasks []ChangelogScanTask) (*arrow.Schema, iter.Seq2[arrow.RecordBatch, error], error) { + if s == nil || s.scan == nil { + return nil, nil, fmt.Errorf("%w: incremental changelog scan is not initialized", ErrInvalidOperation) + } + if err := ctx.Err(); err != nil { + return nil, nil, err + } + for _, task := range tasks { + if err := readableChangelogTask(task); err != nil { + return nil, nil, err + } + } + + schema, useLargeTypes, err := changelogArrowSchema(s.scan) + if err != nil { + return nil, nil, err + } + if s.scan.limit == 0 || len(tasks) == 0 { + return schema, func(yield func(arrow.RecordBatch, error) bool) {}, nil + } + + readScan := s.scan.clone() + limit := s.scan.limit + + return schema, func(yield func(arrow.RecordBatch, error) bool) { + var emitted int64 + for _, task := range tasks { + if err := ctx.Err(); err != nil { + yield(nil, err) + + return + } + if limit > 0 { + readScan.limit = limit - emitted + } + + 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) + + return + } + + for rec, err := range records { + if err != nil { + yield(nil, err) + + return + } + + stamped, rows, err := stampChangelogBatch(ctx, schema, useLargeTypes, rec, task, limit, emitted) + rec.Release() + if err != nil { + yield(nil, err) + + return + } + if stamped == nil { + continue + } + if !yield(stamped, nil) { + return + } + emitted += rows + if limit > 0 && emitted >= limit { + return + } + } + } + }, nil +} + +func readableChangelogTask(task ChangelogScanTask) error { + switch task.(type) { + case AddedRowsScanTask, DeletedDataFileScanTask, DeletedRowsScanTask: + return nil + default: + return fmt.Errorf("%w: unsupported changelog scan task %T", iceberg.ErrInvalidArgument, task) + } +} + +func changelogArrowSchema(scan *Scan) (*arrow.Schema, bool, error) { + if atomic.LoadUint32(&scan.closed) != 0 { + return nil, false, fmt.Errorf("%w: scan is closed", ErrInvalidOperation) + } + if scan.selectorErr != nil { + return nil, false, scan.selectorErr + } + + projected, err := scan.Projection() + if err != nil { + return nil, false, err + } + + useLargeTypes, err := strconv.ParseBool(scan.options.Get(ScanOptionArrowUseLargeTypes, "false")) + if err != nil { + useLargeTypes = false + } + + props := scan.metadata.Properties() + if scan.arrowBatchSize > 0 { + props = maps.Clone(props) + if props == nil { + props = iceberg.Properties{} + } + props[ParquetBatchSizeKey] = strconv.Itoa(scan.arrowBatchSize) + } + + dataSchema, err := SchemaToArrowSchemaWithOptions(projected, ArrowSchemaOptions{ + UseLargeTypes: useLargeTypes, + TableProperties: props, + }) + if err != nil { + return nil, false, err + } + + schema := dataSchema + for _, field := range changelogArrowFields(useLargeTypes) { + schema, err = schema.AddField(schema.NumFields(), field) + if err != nil { + return nil, false, err + } + } + + return schema, useLargeTypes, nil +} + +func changelogArrowFields(useLargeTypes bool) []arrow.Field { + changeType := arrow.BinaryTypes.String + if useLargeTypes { + changeType = arrow.BinaryTypes.LargeString + } + + return []arrow.Field{ + {Name: ChangelogChangeTypeColumn, Type: changeType, Nullable: false}, + {Name: ChangelogChangeOrdinalColumn, Type: arrow.PrimitiveTypes.Int64, Nullable: false}, Review Comment: `_change_ordinal` is emitted as Int64, but Java's `MetadataColumns.CHANGE_ORDINAL` is a 32-bit `IntegerType` (and `ChangelogScanTask.ChangeOrdinal()` returns `int`). Schema-strict consumers that union this with Spark or Java changelog output will see int32 vs int64 and reject it. I'd emit Int32 to match, or document the deviation. While we're here, Java marks `_change_ordinal` and `_commit_snapshot_id` optional, whereas we set all three `Nullable: false`. Stricter and harmless, but worth a note. ########## table/changelog_deleted_rows.go: ########## @@ -0,0 +1,291 @@ +// 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" + +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.addedDeletes} + as.addedPosLoader = newLazyPositionDeleteLoader(as.fs, tasks) + + dvLoader, err := newLazyDeletionVectorLoader(as.fs, tasks) + if err != nil { + as.addedPosLoader.release() + as.addedPosLoader = nil + + return err + } + as.addedDVLoader = dvLoader + + eqLoader, err := newLazyEqualityDeleteLoader( + as.fs, invariants.tableSchema, nil, invariants.nameMapping, tasks) + if err != nil { + as.addedPosLoader.release() + as.addedPosLoader = nil + + return err + } + if eqLoader.needsSchemaHistory() { + eqLoader.tableSchemas = as.metadata.Schemas() + } + eqLoader.addFieldIDs(invariants.projectedIDs) + as.addedEqLoader = eqLoader + + return nil +} + +func (as *arrowScan) loadAddedDeletes(ctx context.Context, task FileScanTask) (positionDeletes, *dv.RoaringPositionBitmap, []*equalityDeleteSet, error) { + if as.addedDeletes == nil { + return nil, nil, nil, nil + } + + dataPath := "" + if task.File != nil { + dataPath = task.File.FilePath() + } + if as.addedDeletes.File == nil || as.addedDeletes.File.FilePath() != dataPath { + return nil, nil, nil, fmt.Errorf("%w: added delete files do not match data file %s", + iceberg.ErrInvalidArgument, dataPath) + } + + pos, err := as.addedPosLoader.load(ctx, *as.addedDeletes) + if err != nil { + return nil, nil, nil, err + } + bitmap, err := as.addedDVLoader.load(ctx, dataPath) + if err != nil { + return nil, nil, nil, err + } + eq, err := as.addedEqLoader.load(ctx, *as.addedDeletes) + 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 := compute.GetAllocator(ctx) Review Comment: `stampChangelogFilePos` (and `keepAddedDeleteRows` below) call `compute.GetAllocator(ctx)` directly and hand the result to the builders, while the rest of this code uses `allocatorFrom(ctx)` with its nil fallback. If `GetAllocator` ever returns nil the builder panics. I'd use `allocatorFrom` in both spots. ########## table/incremental_changelog_deletes.go: ########## @@ -0,0 +1,658 @@ +// 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" + "slices" + + "github.com/apache/iceberg-go" + "github.com/apache/iceberg-go/io" +) + +// changelogTaskPlan is the snapshot range and the data-file changes already +// read for one changelog plan. +type changelogTaskPlan struct { + scan *Scan + fs io.IO + snapshots []Snapshot + manifestsBySnapshot map[int64][]iceberg.ManifestFile + deleteManifests map[int64][]iceberg.ManifestFile + dataEntries []iceberg.ManifestEntry + snapshotOrdinals map[int64]int + schema *iceberg.Schema + partitionFilters *keyDefaultMapErr[int, iceberg.BooleanExpression] + residual iceberg.BooleanExpression + metrics *scanMetricsAccumulator +} + +// changelogDeleteIndex matches delete files to a data file with the same +// sequence rules as a normal scan. +type changelogDeleteIndex struct { + pos *positionalDeleteIndex + dv map[string]iceberg.ManifestEntry + eq *equalityDeleteIndex +} + +// changelogDeleteState tracks delete files added and removed inside the +// changelog range, plus delete files that were already live before it. +type changelogDeleteState struct { + active bool + snapshots []Snapshot + preRange []iceberg.ManifestEntry + added map[int64][]iceberg.ManifestEntry + removed map[int64]map[string]struct{} + specs partitionSpecLookup + schema *iceberg.Schema + + addedCache map[int64]*changelogDeleteIndex + beforeCache map[int64]*changelogDeleteIndex +} + +func (s *IncrementalChangelogScan) planChangelogTasks(ctx context.Context, plan changelogTaskPlan) ([]plannedChangelogTask, error) { + state, err := s.loadChangelogDeleteState(ctx, plan) + if err != nil { + return nil, err + } + + changedPaths := make(map[int64]map[string]struct{}) + tasks := make([]plannedChangelogTask, 0, len(plan.dataEntries)) + for _, entry := range plan.dataEntries { + ordinal, ok := plan.snapshotOrdinals[entry.SnapshotID()] + if !ok { + continue + } + paths := changedPaths[entry.SnapshotID()] + if paths == nil { + paths = make(map[string]struct{}) + changedPaths[entry.SnapshotID()] = paths + } + paths[entry.DataFile().FilePath()] = struct{}{} + + deletes, err := state.deletesFor(entry) + if err != nil { + return nil, fmt.Errorf("incremental changelog scan snapshot %d: %w", entry.SnapshotID(), err) + } + task, err := newChangelogScanTask(entry, ordinal, plan.residual, deletes) + if err != nil { + return nil, fmt.Errorf("incremental changelog scan snapshot %d: %w", entry.SnapshotID(), err) + } + tasks = append(tasks, plannedChangelogTask{task: task}) + } + + rowTasks, err := s.planDeletedRowTasks(ctx, plan, state, changedPaths) + if err != nil { + return nil, err + } + tasks = append(tasks, rowTasks...) + sortChangelogTasks(tasks) + + return tasks, nil +} + +func (s *IncrementalChangelogScan) loadChangelogDeleteState(ctx context.Context, plan changelogTaskPlan) (*changelogDeleteState, error) { + state := &changelogDeleteState{ + snapshots: plan.snapshots, + added: make(map[int64][]iceberg.ManifestEntry), + removed: make(map[int64]map[string]struct{}), + specs: plan.scan.metadata, + schema: plan.schema, + } + for _, snapshot := range plan.snapshots { + if err := ctx.Err(); err != nil { + return nil, err + } + manifests := plan.deleteManifests[snapshot.SnapshotID] + if len(manifests) == 0 { + continue + } + filtered, err := plan.scan.filterManifestsWithSchemaOptions( + manifests, plan.schema, plan.metrics, plan.partitionFilters, true) + if err != nil { + return nil, err + } + entries, err := readManifestEntries(ctx, plan.fs, filtered, false, false) + if err != nil { + return nil, err + } + for _, entry := range entries { + if entry.SnapshotID() != snapshot.SnapshotID { + continue + } + if entry.Status() == iceberg.EntryStatusDELETED { + if state.removed[snapshot.SnapshotID] == nil { + state.removed[snapshot.SnapshotID] = make(map[string]struct{}) + } + state.removed[snapshot.SnapshotID][entry.DataFile().FilePath()] = struct{}{} + + continue + } + state.added[snapshot.SnapshotID] = append(state.added[snapshot.SnapshotID], entry) + } + } + + deletedData := false + for _, entry := range plan.dataEntries { + if entry.Status() == iceberg.EntryStatusDELETED { + deletedData = true + + break + } + } + for _, entries := range state.added { + if len(entries) > 0 { + state.active = true + + break + } + } + if len(state.removed) > 0 || deletedData { + state.active = true + } + if !state.active { + return state, nil + } + + preRange, err := s.liveDeletesBeforeRange(ctx, plan.fs, plan.snapshots) + if err != nil { + return nil, err + } + state.preRange = preRange + + return state, nil +} + +// liveDeletesBeforeRange returns delete files that were already live on the +// snapshot immediately before the changelog range. +func (s *IncrementalChangelogScan) liveDeletesBeforeRange(ctx context.Context, fs io.IO, snapshots []Snapshot) ([]iceberg.ManifestEntry, error) { + parent := s.snapshotBeforeRange(snapshots) + if parent == nil { + return nil, nil + } + manifests, err := parent.Manifests(fs) + if err != nil { + return nil, err + } + deleteManifests := make([]iceberg.ManifestFile, 0) + for _, manifest := range manifests { + if manifest.ManifestContent() == iceberg.ManifestContentDeletes { + deleteManifests = append(deleteManifests, manifest) + } + } + entries, err := readManifestEntries(ctx, fs, deleteManifests, false, false) + if err != nil { + return nil, err + } + + return liveManifestEntries(entries), nil +} + +func (s *IncrementalChangelogScan) snapshotBeforeRange(snapshots []Snapshot) *Snapshot { + if s.fromSnapshotID != nil && !s.fromInclusive { + return s.scan.metadata.SnapshotByID(*s.fromSnapshotID) + } + if len(snapshots) == 0 || snapshots[0].ParentSnapshotID == nil { + return nil + } + + return s.scan.metadata.SnapshotByID(*snapshots[0].ParentSnapshotID) +} + +func (s *IncrementalChangelogScan) planDeletedRowTasks( + ctx context.Context, + plan changelogTaskPlan, + state *changelogDeleteState, + changedPaths map[int64]map[string]struct{}, +) ([]plannedChangelogTask, error) { + if state == nil || !state.active { + return nil, nil + } + matcher, err := plan.scan.dataFileMatcher(plan.schema, plan.partitionFilters) + if err != nil { + return nil, err + } + + var tasks []plannedChangelogTask + for _, snapshot := range plan.snapshots { + if err := ctx.Err(); err != nil { + return nil, err + } + if len(state.added[snapshot.SnapshotID]) == 0 { + continue + } + addedIndex, err := state.addedIndex(snapshot.SnapshotID) + if err != nil { + return nil, fmt.Errorf("incremental changelog scan snapshot %d: %w", snapshot.SnapshotID, err) + } + manifests := plan.manifestsBySnapshot[snapshot.SnapshotID] + dataManifests := make([]iceberg.ManifestFile, 0, len(manifests)) + for _, manifest := range manifests { + if manifest.ManifestContent() == iceberg.ManifestContentData { + dataManifests = append(dataManifests, manifest) + } + } + // Prune with the scan partition filters before opening manifests. + // A local accumulator keeps carried-forward data manifests out of the + // changelog scan report, which already counted them. + var pruned scanMetricsAccumulator + dataManifests, err = plan.scan.filterManifestsWithSchemaOptions( + dataManifests, plan.schema, &pruned, plan.partitionFilters, true) + if err != nil { + return nil, err + } + live, err := liveDataEntries(ctx, plan.fs, dataManifests) + if err != nil { + return nil, err + } + ordinal := plan.snapshotOrdinals[snapshot.SnapshotID] + seen := make(map[string]struct{}) + for _, entry := range live { + path := entry.DataFile().FilePath() + if _, changed := changedPaths[snapshot.SnapshotID][path]; changed { + continue + } + if _, ok := seen[path]; ok { + continue + } + seen[path] = struct{}{} + + keep, err := matcher(entry.DataFile()) + if err != nil { + return nil, err + } + if !keep { + continue + } + added, err := addedIndex.forDataFile(entry) + if err != nil { + return nil, fmt.Errorf("incremental changelog scan snapshot %d: %w", snapshot.SnapshotID, err) + } + if len(added) == 0 { + continue + } + var existing []iceberg.DataFile + before, err := state.beforeIndex(snapshot.SnapshotID) + if err != nil { + return nil, fmt.Errorf("incremental changelog scan snapshot %d: %w", snapshot.SnapshotID, err) + } + if before != nil { + existing, err = before.forDataFile(entry) + if err != nil { + return nil, fmt.Errorf("incremental changelog scan snapshot %d: %w", snapshot.SnapshotID, err) + } + } + + task, err := NewDeletedRowsScanTask(entry.DataFile(), added, existing, ordinal, snapshot.SnapshotID) + if err != nil { + return nil, fmt.Errorf("incremental changelog scan snapshot %d: %w", snapshot.SnapshotID, err) + } + configureChangelogFileTask(&task.FileScanTask, entry, plan.residual) + tasks = append(tasks, plannedChangelogTask{task: task}) + } + } + + return tasks, nil +} + +func (st *changelogDeleteState) deletesFor(entry iceberg.ManifestEntry) ([]iceberg.DataFile, error) { + if st == nil || !st.active { + return nil, nil + } + + var ( + idx *changelogDeleteIndex + err error + ) + switch entry.Status() { + case iceberg.EntryStatusADDED: + idx, err = st.addedIndex(entry.SnapshotID()) + case iceberg.EntryStatusDELETED: + idx, err = st.beforeIndex(entry.SnapshotID()) + default: + return nil, fmt.Errorf("%w: unknown manifest entry status %d", ErrInvalidMetadata, entry.Status()) + } + if err != nil || idx == nil { + return nil, err + } + + return idx.forDataFile(entry) +} + +func (st *changelogDeleteState) addedIndex(snapshotID int64) (*changelogDeleteIndex, error) { + if st.addedCache == nil { + st.addedCache = make(map[int64]*changelogDeleteIndex) + } + if idx, ok := st.addedCache[snapshotID]; ok { + return idx, nil + } + idx, err := buildChangelogDeleteIndex(st.added[snapshotID], st.specs, st.schema) + if err != nil { + return nil, err + } + st.addedCache[snapshotID] = idx + + return idx, nil +} + +func (st *changelogDeleteState) beforeIndex(snapshotID int64) (*changelogDeleteIndex, error) { + if st.beforeCache == nil { + st.beforeCache = make(map[int64]*changelogDeleteIndex) + } + if idx, ok := st.beforeCache[snapshotID]; ok { + return idx, nil + } + idx, err := buildChangelogDeleteIndex(st.entriesBefore(snapshotID), st.specs, st.schema) + if err != nil { + return nil, err + } + st.beforeCache[snapshotID] = idx + + return idx, nil +} + +// entriesBefore returns delete-file entries that applied strictly before +// snapshotID. Removals drop a path from the running set. A path removed and +// added again in one snapshot stays live as the added entry. +func (st *changelogDeleteState) entriesBefore(snapshotID int64) []iceberg.ManifestEntry { + removed := make(map[string]struct{}) + var accumulated []iceberg.ManifestEntry + for _, snapshot := range st.snapshots { + if snapshot.SnapshotID == snapshotID { + break + } + currentRemoved := st.removed[snapshot.SnapshotID] + if len(currentRemoved) > 0 { + for path := range currentRemoved { + removed[path] = struct{}{} + } + accumulated = dropManifestEntries(accumulated, currentRemoved) + } + accumulated = append(accumulated, st.added[snapshot.SnapshotID]...) + } + + out := make([]iceberg.ManifestEntry, 0, len(st.preRange)+len(accumulated)) + for _, entry := range st.preRange { + if _, gone := removed[entry.DataFile().FilePath()]; gone { + continue + } + out = append(out, entry) + } + out = append(out, accumulated...) + + return keepNewestDeletionVector(out) +} + +func (idx *changelogDeleteIndex) forDataFile(entry iceberg.ManifestEntry) ([]iceberg.DataFile, error) { + if idx == nil { + return nil, nil + } + task, err := fileScanTaskForDataEntry(entry, idx.pos, idx.dv, idx.eq) + if err != nil { + return nil, err + } + files := allDeleteFiles(task) + if len(files) == 0 { + return nil, nil + } + + return files, nil +} + +func buildChangelogDeleteIndex(entries []iceberg.ManifestEntry, specs partitionSpecLookup, schema *iceberg.Schema) (*changelogDeleteIndex, error) { + if len(entries) == 0 { + return nil, nil + } + entries = keepNewestDeletionVector(entries) + classified, err := classifyManifestEntries(entries) + if err != nil { + return nil, err + } + if len(classified.dataEntries) > 0 { + return nil, fmt.Errorf("%w: expected delete file, got %s", + ErrInvalidMetadata, classified.dataEntries[0].DataFile().FilePath()) + } + + pos, err := buildPositionalDeleteIndex(classified.positionalDeleteEntries) + if err != nil { + return nil, err + } + dv, err := buildDVIndex(classified.dvEntries) + if err != nil { + return nil, err + } + eq, err := buildEqualityDeleteIndex(classified.equalityDeleteEntries, specs, schema) + if err != nil { + return nil, err + } + + return &changelogDeleteIndex{pos: pos, dv: dv, eq: eq}, nil +} + +func (scan *Scan) dataFileMatcher( + schema *iceberg.Schema, + partitionFilters *keyDefaultMapErr[int, iceberg.BooleanExpression], +) (func(iceberg.DataFile) (bool, error), error) { + metricsEval, err := newInclusiveMetricsEvaluator( + schema, + scan.rowFilter, + scan.caseSensitive, + scan.options["include_empty_files"] == "true", + ) + if err != nil { + return nil, err + } + partitionEvaluators := newKeyDefaultMapWrapErr(func(specID int) (func(iceberg.DataFile) (bool, error), error) { + return buildPartitionEvaluator(specID, scan.metadata, schema, partitionFilters, scan.caseSensitive) + }) + + return func(file iceberg.DataFile) (bool, error) { + partEval, err := partitionEvaluators.Get(int(file.SpecID())) + if err != nil { + return false, fmt.Errorf("failed to build partition evaluator for spec %d: %w", file.SpecID(), err) + } + use, err := partEval(file) + if err != nil || !use { + return use, err + } + + return metricsEval(file) + }, nil +} + +func liveDataEntries(ctx context.Context, fs io.IO, manifests []iceberg.ManifestFile) ([]iceberg.ManifestEntry, error) { + dataManifests := make([]iceberg.ManifestFile, 0, len(manifests)) + for _, manifest := range manifests { + if manifest.ManifestContent() == iceberg.ManifestContentData { + dataManifests = append(dataManifests, manifest) + } + } + entries, err := readManifestEntries(ctx, fs, dataManifests, false, false) + if err != nil { + return nil, err + } + dataEntries := make([]iceberg.ManifestEntry, 0, len(entries)) + for _, entry := range entries { + if entry.DataFile().ContentType() == iceberg.EntryContentData { + dataEntries = append(dataEntries, entry) + } + } + + return liveManifestEntries(dataEntries), nil +} + +// liveManifestEntries keeps the highest-sequence entry for each path and +// drops paths whose winning entry is a deletion. Equal sequence numbers +// keep the later entry. +func liveManifestEntries(entries []iceberg.ManifestEntry) []iceberg.ManifestEntry { + latest := make(map[string]iceberg.ManifestEntry, len(entries)) + order := make([]string, 0, len(entries)) + for _, entry := range entries { + path := entry.DataFile().FilePath() + current, seen := latest[path] + if !seen { + order = append(order, path) + latest[path] = entry + + continue + } + if entry.SequenceNum() >= current.SequenceNum() { Review Comment: A DELETED entry keeps the file's original data sequence number, so with `>=` and a later-wins tie-break it can hide an EXISTING entry of the same path, and the outcome then depends on manifest ordering the spec doesn't guarantee. It's normally masked because the rewritten manifest replaces the original, but a mixed layout could resurrect a removed file or hide a live one. I'd filter to ADDED/EXISTING and dedupe by path, or let a DELETED entry win only when it comes from a strictly newer snapshot. ########## table/changelog_reader.go: ########## @@ -0,0 +1,297 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package table + +import ( + "context" + "fmt" + "iter" + "maps" + "strconv" + "sync/atomic" + + "github.com/apache/arrow-go/v18/arrow" + "github.com/apache/arrow-go/v18/arrow/array" + "github.com/apache/arrow-go/v18/arrow/compute" + "github.com/apache/arrow-go/v18/arrow/memory" + "github.com/apache/iceberg-go" +) + +const ( + // ChangelogChangeTypeColumn is the changelog operation: INSERT, DELETE, + // UPDATE_BEFORE, or UPDATE_AFTER. Review Comment: This doc lists UPDATE_BEFORE and UPDATE_AFTER, but the reader only emits INSERT and DELETE (a MoR update surfaces as a DELETE plus an INSERT with the same ordinal, matching the Java raw changelog). I'd reword so it doesn't imply we produce UPDATE rows. ########## table/incremental_changelog_deletes.go: ########## @@ -0,0 +1,658 @@ +// 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" + "slices" + + "github.com/apache/iceberg-go" + "github.com/apache/iceberg-go/io" +) + +// changelogTaskPlan is the snapshot range and the data-file changes already +// read for one changelog plan. +type changelogTaskPlan struct { + scan *Scan + fs io.IO + snapshots []Snapshot + manifestsBySnapshot map[int64][]iceberg.ManifestFile + deleteManifests map[int64][]iceberg.ManifestFile + dataEntries []iceberg.ManifestEntry + snapshotOrdinals map[int64]int + schema *iceberg.Schema + partitionFilters *keyDefaultMapErr[int, iceberg.BooleanExpression] + residual iceberg.BooleanExpression + metrics *scanMetricsAccumulator +} + +// changelogDeleteIndex matches delete files to a data file with the same +// sequence rules as a normal scan. +type changelogDeleteIndex struct { + pos *positionalDeleteIndex + dv map[string]iceberg.ManifestEntry + eq *equalityDeleteIndex +} + +// changelogDeleteState tracks delete files added and removed inside the +// changelog range, plus delete files that were already live before it. +type changelogDeleteState struct { + active bool + snapshots []Snapshot + preRange []iceberg.ManifestEntry + added map[int64][]iceberg.ManifestEntry + removed map[int64]map[string]struct{} + specs partitionSpecLookup + schema *iceberg.Schema + + addedCache map[int64]*changelogDeleteIndex + beforeCache map[int64]*changelogDeleteIndex +} + +func (s *IncrementalChangelogScan) planChangelogTasks(ctx context.Context, plan changelogTaskPlan) ([]plannedChangelogTask, error) { + state, err := s.loadChangelogDeleteState(ctx, plan) + if err != nil { + return nil, err + } + + changedPaths := make(map[int64]map[string]struct{}) + tasks := make([]plannedChangelogTask, 0, len(plan.dataEntries)) + for _, entry := range plan.dataEntries { + ordinal, ok := plan.snapshotOrdinals[entry.SnapshotID()] + if !ok { + continue + } + paths := changedPaths[entry.SnapshotID()] + if paths == nil { + paths = make(map[string]struct{}) + changedPaths[entry.SnapshotID()] = paths + } + paths[entry.DataFile().FilePath()] = struct{}{} + + deletes, err := state.deletesFor(entry) + if err != nil { + return nil, fmt.Errorf("incremental changelog scan snapshot %d: %w", entry.SnapshotID(), err) + } + task, err := newChangelogScanTask(entry, ordinal, plan.residual, deletes) + if err != nil { + return nil, fmt.Errorf("incremental changelog scan snapshot %d: %w", entry.SnapshotID(), err) + } + tasks = append(tasks, plannedChangelogTask{task: task}) + } + + rowTasks, err := s.planDeletedRowTasks(ctx, plan, state, changedPaths) + if err != nil { + return nil, err + } + tasks = append(tasks, rowTasks...) + sortChangelogTasks(tasks) + + return tasks, nil +} + +func (s *IncrementalChangelogScan) loadChangelogDeleteState(ctx context.Context, plan changelogTaskPlan) (*changelogDeleteState, error) { + state := &changelogDeleteState{ + snapshots: plan.snapshots, + added: make(map[int64][]iceberg.ManifestEntry), + removed: make(map[int64]map[string]struct{}), + specs: plan.scan.metadata, + schema: plan.schema, + } + for _, snapshot := range plan.snapshots { Review Comment: State is built only from `plan.snapshots`, which skips REPLACE snapshots (matching Java). But a REPLACE between two in-range snapshots can rewrite a position-delete file and consolidate positions, and `existing` is then computed from a delete set that doesn't reflect it. Safe when the rewrite is content-equal, fragile when positions get consolidated. I'd either track delete adds/removes across REPLACE snapshots for state reconstruction only, or document and test the invariant that replace-time rewrites preserve semantics. `TestIncrementalChangelogReadSkipsReplaceBeforeRowDelete` only covers one shape. ########## table/changelog_deleted_rows.go: ########## @@ -0,0 +1,291 @@ +// 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" + +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.addedDeletes} + as.addedPosLoader = newLazyPositionDeleteLoader(as.fs, tasks) + + dvLoader, err := newLazyDeletionVectorLoader(as.fs, tasks) + if err != nil { + as.addedPosLoader.release() + as.addedPosLoader = nil + + return err + } + as.addedDVLoader = dvLoader + + eqLoader, err := newLazyEqualityDeleteLoader( + as.fs, invariants.tableSchema, nil, invariants.nameMapping, tasks) + if err != nil { + as.addedPosLoader.release() + as.addedPosLoader = nil + + return err + } + if eqLoader.needsSchemaHistory() { + eqLoader.tableSchemas = as.metadata.Schemas() + } + eqLoader.addFieldIDs(invariants.projectedIDs) + as.addedEqLoader = eqLoader + + return nil +} + +func (as *arrowScan) loadAddedDeletes(ctx context.Context, task FileScanTask) (positionDeletes, *dv.RoaringPositionBitmap, []*equalityDeleteSet, error) { + if as.addedDeletes == nil { + return nil, nil, nil, nil + } + + dataPath := "" + if task.File != nil { + dataPath = task.File.FilePath() + } + if as.addedDeletes.File == nil || as.addedDeletes.File.FilePath() != dataPath { + return nil, nil, nil, fmt.Errorf("%w: added delete files do not match data file %s", + iceberg.ErrInvalidArgument, dataPath) + } + + pos, err := as.addedPosLoader.load(ctx, *as.addedDeletes) + if err != nil { + return nil, nil, nil, err + } + bitmap, err := as.addedDVLoader.load(ctx, dataPath) + if err != nil { + return nil, nil, nil, err + } + eq, err := as.addedEqLoader.load(ctx, *as.addedDeletes) + 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 := compute.GetAllocator(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() + + n := int(in.NumCols()) + if n == 0 { + in.Retain() + + return in, nil + } + + fields := in.Schema().Fields() + schema := arrow.NewSchema(fields[:n-1], nil) Review Comment: `dropChangelogFilePos` finds the stamp positionally (last column), so every pipeline step between stamp and drop has to preserve column order for this to stay correct. A future change to the equality or row-id steps breaks it silently, dropping the wrong column. I'd look the column up by `changelogFilePosColumn` name and fail loudly if it's missing. (The `n == 0` branch can't occur once the stamp was added.) -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
