laskoviymishka commented on code in PR #2169: URL: https://github.com/apache/iceberg-go/pull/2169#discussion_r4326443849
########## table/incremental_changelog_deletes.go: ########## @@ -0,0 +1,879 @@ +// 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" + "strconv" + + "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 + timeline []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) { + // timeline includes replace snapshots so their delete-file adds and + // removals update later tasks. Task emission uses plan.snapshots, which + // omits those replace snapshots. + timeline := plan.timeline + if len(timeline) == 0 { + timeline = plan.snapshots + } + state := &changelogDeleteState{ + snapshots: timeline, + added: make(map[int64][]iceberg.ManifestEntry), + removed: make(map[int64]map[string]struct{}), + specs: plan.scan.metadata, + schema: plan.schema, + } + for _, snapshot := range timeline { + 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][deleteEntryKey(entry.DataFile())] = 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) + 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. When that snapshot has been +// expired, the first in-range snapshot's own delete manifests are used instead. +func (s *IncrementalChangelogScan) liveDeletesBeforeRange(ctx context.Context, plan changelogTaskPlan) ([]iceberg.ManifestEntry, error) { + snapshots := timelineSnapshots(plan) + parent := s.snapshotBeforeRange(snapshots) + if parent == nil { + if len(snapshots) == 0 || snapshots[0].ParentSnapshotID == nil { + return nil, nil + } + + return s.deletesLiveBeforeExpiredParent(ctx, plan, snapshots[0]) + } + manifests, err := parent.Manifests(plan.fs) + if err != nil { + return nil, err + } + entries, err := s.readDeleteManifestEntries(ctx, plan, manifests) + if err != nil { + return nil, err + } + + return liveDeleteEntries(entries, snapshotSequences(plan.scan.metadata)), nil +} + +// deletesLiveBeforeExpiredParent rebuilds deletes that applied before the +// range from the first timeline snapshot. A non-deleted entry written by an +// earlier snapshot was still live. A deleted entry written by this snapshot +// was live until this snapshot removed it. +func (s *IncrementalChangelogScan) deletesLiveBeforeExpiredParent( + ctx context.Context, plan changelogTaskPlan, snapshot Snapshot, +) ([]iceberg.ManifestEntry, error) { + manifests, err := snapshot.Manifests(plan.fs) + if err != nil { + return nil, err + } + entries, err := s.readDeleteManifestEntries(ctx, plan, manifests) + if err != nil { + return nil, err + } + + return entriesLiveBefore(entries, snapshot.SnapshotID), nil Review Comment: This rebuild only works if the first kept snapshot wrote a DELETED entry, under its own snapshot id, for every delete file it removed. Java's writer does that. A writer that just drops a fully deleted delete manifest from the list would leave those deletes out of `preRange`, and we're back to empty `ExistingDeletes` with nothing to signal it. I'd document that assumption on `deletesLiveBeforeExpiredParent` and add a cheap guard. When the summary has `removed-delete-files` and it's larger than the number of DELETED delete entries we recovered, return an error instead of guessing. Take that count before the partition pruning in `readDeleteManifestEntries`, or filtered scans will trip the guard. ########## table/incremental_changelog_scan_test.go: ########## @@ -768,6 +809,776 @@ func incrementalChangelogDeleteManifestTable(t *testing.T) *Table { }, nil) } +func TestIncrementalChangelogScanAttachesSameSnapshotDeletes(t *testing.T) { + spec := partitionedSpec() + data := newTestDataFile(t, spec, "mem://default/changelog-same/data.parquet", map[int]any{1000: int32(1)}) + posDelete := newTestPosDeleteFileForSpec(t, spec, + "mem://default/changelog-same/pos.parquet", map[int]any{1000: int32(1)}, data.FilePath()) + eqDelete := newChangelogEqDelete(t, spec, + "mem://default/changelog-same/eq.parquet", map[int]any{1000: int32(1)}) + tbl := writeChangelogScanTable(t, "changelog-same", []changelogScanStep{{ + operation: OpAppend, + data: []iceberg.ManifestEntry{changelogManifestEntry(iceberg.EntryStatusADDED, 1, 1, data)}, + deletes: []iceberg.ManifestEntry{ + changelogManifestEntry(iceberg.EntryStatusADDED, 1, 1, posDelete), + changelogManifestEntry(iceberg.EntryStatusADDED, 1, 1, eqDelete), + }, + }}) + + tasks, err := tbl.NewIncrementalChangelogScan().PlanFiles(context.Background()) + require.NoError(t, err) + require.Len(t, tasks, 1) + added, ok := tasks[0].(AddedRowsScanTask) + require.True(t, ok) + require.Equal(t, []string{posDelete.FilePath()}, changelogFilePaths(added.Deletes())) + require.Empty(t, added.ScanTask().EqualityDeleteFiles) +} + +func TestIncrementalChangelogScanSeparatesAddedAndExistingDeletes(t *testing.T) { + spec := partitionedSpec() + data := newTestDataFile(t, spec, "mem://default/changelog-prior/data.parquet", map[int]any{1000: int32(1)}) + first := newTestPosDeleteFileForSpec(t, spec, + "mem://default/changelog-prior/first.parquet", map[int]any{1000: int32(1)}, data.FilePath()) + second := newTestPosDeleteFileForSpec(t, spec, + "mem://default/changelog-prior/second.parquet", map[int]any{1000: int32(1)}, data.FilePath()) + tbl := writeChangelogScanTable(t, "changelog-prior", []changelogScanStep{ + { + operation: OpAppend, + data: []iceberg.ManifestEntry{changelogManifestEntry(iceberg.EntryStatusADDED, 1, 1, data)}, + }, + { + operation: OpDelete, + deletes: []iceberg.ManifestEntry{changelogManifestEntry(iceberg.EntryStatusADDED, 2, 2, first)}, + }, + { + operation: OpDelete, + deletes: []iceberg.ManifestEntry{changelogManifestEntry(iceberg.EntryStatusADDED, 3, 3, second)}, + }, + }) + + tasks, err := tbl.NewIncrementalChangelogScan().PlanFiles(context.Background()) + require.NoError(t, err) + require.Len(t, tasks, 3) + + firstRows, ok := tasks[1].(DeletedRowsScanTask) + require.True(t, ok) + require.Equal(t, []string{first.FilePath()}, changelogFilePaths(firstRows.AddedDeletes())) + require.Empty(t, firstRows.ExistingDeletes()) + + secondRows, ok := tasks[2].(DeletedRowsScanTask) + require.True(t, ok) + require.Equal(t, 2, secondRows.ChangeOrdinal()) + require.Equal(t, []string{second.FilePath()}, changelogFilePaths(secondRows.AddedDeletes())) + require.Equal(t, []string{first.FilePath()}, changelogFilePaths(secondRows.ExistingDeletes())) +} + +func TestIncrementalChangelogScanPlansEqualityDeletes(t *testing.T) { + spec := partitionedSpec() + data := newTestDataFile(t, spec, "mem://default/changelog-eq/data.parquet", map[int]any{1000: int32(1)}) + eqDelete := newChangelogEqDelete(t, spec, + "mem://default/changelog-eq/eq.parquet", map[int]any{1000: int32(1)}) + tbl := writeChangelogScanTable(t, "changelog-eq", []changelogScanStep{ + { + operation: OpAppend, + data: []iceberg.ManifestEntry{changelogManifestEntry(iceberg.EntryStatusADDED, 1, 1, data)}, + }, + { + operation: OpDelete, + deletes: []iceberg.ManifestEntry{changelogManifestEntry(iceberg.EntryStatusADDED, 2, 2, eqDelete)}, + }, + }) + + tasks, err := tbl.NewIncrementalChangelogScan().PlanFiles(context.Background()) + require.NoError(t, err) + require.Len(t, tasks, 2) + rows, ok := tasks[1].(DeletedRowsScanTask) + require.True(t, ok) + require.Equal(t, []string{eqDelete.FilePath()}, changelogFilePaths(rows.AddedDeletes())) + require.Empty(t, rows.ScanTask().EqualityDeleteFiles) + require.Equal(t, iceberg.EntryContentEqDeletes, rows.AddedDeletes()[0].ContentType()) +} + +func TestIncrementalChangelogScanPlansDeletionVectors(t *testing.T) { + spec := partitionedSpec() + data := newTestDataFile(t, spec, "mem://default/changelog-dv/data.parquet", map[int]any{1000: int32(1)}) + posDelete := newTestPosDeleteFileForSpec(t, spec, + "mem://default/changelog-dv/pos.parquet", map[int]any{1000: int32(1)}, data.FilePath()) + dv := newChangelogDeletionVector(t, spec, "mem://default/changelog-dv/dv.puffin", data.FilePath()) + tbl := writeChangelogScanTable(t, "changelog-dv", []changelogScanStep{ + { + operation: OpAppend, + data: []iceberg.ManifestEntry{changelogManifestEntry(iceberg.EntryStatusADDED, 1, 1, data)}, + }, + { + operation: OpDelete, + deletes: []iceberg.ManifestEntry{changelogManifestEntry(iceberg.EntryStatusADDED, 2, 2, posDelete)}, + }, + { + operation: OpDelete, + manifestVersion: 3, + deletes: []iceberg.ManifestEntry{changelogManifestEntry(iceberg.EntryStatusADDED, 3, 3, dv)}, + }, + }) + + tasks, err := tbl.NewIncrementalChangelogScan().PlanFiles(context.Background()) + require.NoError(t, err) + require.Len(t, tasks, 3) + rows, ok := tasks[2].(DeletedRowsScanTask) + require.True(t, ok) + require.Equal(t, []string{dv.FilePath()}, changelogFilePaths(rows.AddedDeletes())) + require.Equal(t, []string{posDelete.FilePath()}, changelogFilePaths(rows.ExistingDeletes())) + require.Empty(t, rows.ScanTask().DeletionVectorFiles) +} + +func TestIncrementalChangelogScanAppliesExistingDeletesToRemovedFiles(t *testing.T) { + spec := partitionedSpec() + data := newTestDataFile(t, spec, "mem://default/changelog-removed/data.parquet", map[int]any{1000: int32(1)}) + posDelete := newTestPosDeleteFileForSpec(t, spec, + "mem://default/changelog-removed/pos.parquet", map[int]any{1000: int32(1)}, data.FilePath()) + tbl := writeChangelogScanTable(t, "changelog-removed", []changelogScanStep{ + { + operation: OpAppend, + data: []iceberg.ManifestEntry{changelogManifestEntry(iceberg.EntryStatusADDED, 1, 1, data)}, + }, + { + operation: OpDelete, + deletes: []iceberg.ManifestEntry{changelogManifestEntry(iceberg.EntryStatusADDED, 2, 2, posDelete)}, + }, + { + operation: OpOverwrite, + data: []iceberg.ManifestEntry{changelogManifestEntry(iceberg.EntryStatusDELETED, 3, 1, data)}, + }, + }) + + tasks, err := tbl.NewIncrementalChangelogScan().PlanFiles(context.Background()) + require.NoError(t, err) + require.Len(t, tasks, 3) + removed, ok := tasks[2].(DeletedDataFileScanTask) + require.True(t, ok) + require.Equal(t, ChangelogOpDelete, removed.Operation()) + require.Equal(t, 2, removed.ChangeOrdinal()) + require.Equal(t, []string{posDelete.FilePath()}, changelogFilePaths(removed.ExistingDeletes())) + + bounded, err := tbl.NewIncrementalChangelogScan(). + FromSnapshotExclusive(2). + ToSnapshot(3). + PlanFiles(context.Background()) + require.NoError(t, err) + require.Len(t, bounded, 1) + removed, ok = bounded[0].(DeletedDataFileScanTask) + require.True(t, ok) + require.Equal(t, []string{posDelete.FilePath()}, changelogFilePaths(removed.ExistingDeletes())) +} + +func TestIncrementalChangelogScanFiltersDeleteBackedTasks(t *testing.T) { + tbl := incrementalChangelogDeleteManifestTable(t) + filter := iceberg.EqualTo(iceberg.Reference("id"), int32(3)) + + tasks, err := tbl.NewIncrementalChangelogScan(WithRowFilter(filter)).PlanFiles(context.Background()) + require.NoError(t, err) + require.Len(t, tasks, 1) + require.Equal(t, "mem://default/changelog-delete/data-3.parquet", tasks[0].ScanTask().File.FilePath()) + require.Equal(t, ChangelogOpInsert, tasks[0].Operation()) +} + +func TestIncrementalChangelogScanFiltersDeletedRowsByPartition(t *testing.T) { + spec := partitionedSpec() + partOne := newTestDataFile(t, spec, "mem://default/changelog-part/data-1.parquet", map[int]any{1000: int32(1)}) + partTwo := newTestDataFile(t, spec, "mem://default/changelog-part/data-2.parquet", map[int]any{1000: int32(2)}) + deleteOne := newTestPosDeleteFileForSpec(t, spec, + "mem://default/changelog-part/delete-1.parquet", map[int]any{1000: int32(1)}, partOne.FilePath()) + deleteTwo := newTestPosDeleteFileForSpec(t, spec, + "mem://default/changelog-part/delete-2.parquet", map[int]any{1000: int32(2)}, partTwo.FilePath()) + tbl := writeChangelogScanTable(t, "changelog-part", []changelogScanStep{ + { + operation: OpAppend, + data: []iceberg.ManifestEntry{ + changelogManifestEntry(iceberg.EntryStatusADDED, 1, 1, partOne), + changelogManifestEntry(iceberg.EntryStatusADDED, 1, 1, partTwo), + }, + }, + { + operation: OpDelete, + deletes: []iceberg.ManifestEntry{ + changelogManifestEntry(iceberg.EntryStatusADDED, 2, 2, deleteOne), + changelogManifestEntry(iceberg.EntryStatusADDED, 2, 2, deleteTwo), + }, + }, + }) + + tasks, err := tbl.NewIncrementalChangelogScan( + WithRowFilter(iceberg.EqualTo(iceberg.Reference("id"), int32(1))), + ).PlanFiles(context.Background()) + require.NoError(t, err) + + var deleted []DeletedRowsScanTask + for _, task := range tasks { + require.NotEqual(t, partTwo.FilePath(), task.ScanTask().File.FilePath()) + rows, ok := task.(DeletedRowsScanTask) + if !ok { + continue + } + deleted = append(deleted, rows) + } + require.Len(t, deleted, 1) + require.Equal(t, partOne.FilePath(), deleted[0].ScanTask().File.FilePath()) + require.Equal(t, []string{deleteOne.FilePath()}, changelogFilePaths(deleted[0].AddedDeletes())) +} + +func TestIncrementalChangelogScanAppliesDeletesBySequence(t *testing.T) { + spec := partitionedSpec() + data := newTestDataFile(t, spec, "mem://default/changelog-seq/data.parquet", map[int]any{1000: int32(1)}) + dataSeq := int64(5) + + tests := []struct { + name string + kind string + deleteSeq int64 + want bool + }{ + {name: "position delete below data sequence", kind: "pos", deleteSeq: 4, want: false}, Review Comment: I'd rebuild this table on sequence numbers a writer can actually produce. The data file is added by snapshot 1 at seq 5 and the delete by snapshot 2 at seq 4 or 5. A later snapshot's delete always has a higher seq than an earlier snapshot's data, so the "below" and "at" rows can't happen in a real table. Something like data at 2 in snapshot 2 and deletes in 3 and 4 would pin the rules on real metadata. Every row also goes through `addedIndex`. Nothing varies seq on the `beforeIndex` side (`ExistingDeletes` on a removed file or on a `DeletedRowsScanTask`). The same-snapshot case, where an eq delete at equal seq must not apply but pos/DV must, is still fixed at 1/1 in `AttachesSameSnapshotDeletes`. I'd add both to this table. ########## table/incremental_changelog_scan_test.go: ########## @@ -768,6 +809,776 @@ func incrementalChangelogDeleteManifestTable(t *testing.T) *Table { }, nil) } +func TestIncrementalChangelogScanAttachesSameSnapshotDeletes(t *testing.T) { + spec := partitionedSpec() + data := newTestDataFile(t, spec, "mem://default/changelog-same/data.parquet", map[int]any{1000: int32(1)}) + posDelete := newTestPosDeleteFileForSpec(t, spec, + "mem://default/changelog-same/pos.parquet", map[int]any{1000: int32(1)}, data.FilePath()) + eqDelete := newChangelogEqDelete(t, spec, + "mem://default/changelog-same/eq.parquet", map[int]any{1000: int32(1)}) + tbl := writeChangelogScanTable(t, "changelog-same", []changelogScanStep{{ + operation: OpAppend, + data: []iceberg.ManifestEntry{changelogManifestEntry(iceberg.EntryStatusADDED, 1, 1, data)}, + deletes: []iceberg.ManifestEntry{ + changelogManifestEntry(iceberg.EntryStatusADDED, 1, 1, posDelete), + changelogManifestEntry(iceberg.EntryStatusADDED, 1, 1, eqDelete), + }, + }}) + + tasks, err := tbl.NewIncrementalChangelogScan().PlanFiles(context.Background()) + require.NoError(t, err) + require.Len(t, tasks, 1) + added, ok := tasks[0].(AddedRowsScanTask) + require.True(t, ok) + require.Equal(t, []string{posDelete.FilePath()}, changelogFilePaths(added.Deletes())) + require.Empty(t, added.ScanTask().EqualityDeleteFiles) +} + +func TestIncrementalChangelogScanSeparatesAddedAndExistingDeletes(t *testing.T) { + spec := partitionedSpec() + data := newTestDataFile(t, spec, "mem://default/changelog-prior/data.parquet", map[int]any{1000: int32(1)}) + first := newTestPosDeleteFileForSpec(t, spec, + "mem://default/changelog-prior/first.parquet", map[int]any{1000: int32(1)}, data.FilePath()) + second := newTestPosDeleteFileForSpec(t, spec, + "mem://default/changelog-prior/second.parquet", map[int]any{1000: int32(1)}, data.FilePath()) + tbl := writeChangelogScanTable(t, "changelog-prior", []changelogScanStep{ + { + operation: OpAppend, + data: []iceberg.ManifestEntry{changelogManifestEntry(iceberg.EntryStatusADDED, 1, 1, data)}, + }, + { + operation: OpDelete, + deletes: []iceberg.ManifestEntry{changelogManifestEntry(iceberg.EntryStatusADDED, 2, 2, first)}, + }, + { + operation: OpDelete, + deletes: []iceberg.ManifestEntry{changelogManifestEntry(iceberg.EntryStatusADDED, 3, 3, second)}, + }, + }) + + tasks, err := tbl.NewIncrementalChangelogScan().PlanFiles(context.Background()) + require.NoError(t, err) + require.Len(t, tasks, 3) + + firstRows, ok := tasks[1].(DeletedRowsScanTask) + require.True(t, ok) + require.Equal(t, []string{first.FilePath()}, changelogFilePaths(firstRows.AddedDeletes())) + require.Empty(t, firstRows.ExistingDeletes()) + + secondRows, ok := tasks[2].(DeletedRowsScanTask) + require.True(t, ok) + require.Equal(t, 2, secondRows.ChangeOrdinal()) + require.Equal(t, []string{second.FilePath()}, changelogFilePaths(secondRows.AddedDeletes())) + require.Equal(t, []string{first.FilePath()}, changelogFilePaths(secondRows.ExistingDeletes())) +} + +func TestIncrementalChangelogScanPlansEqualityDeletes(t *testing.T) { + spec := partitionedSpec() + data := newTestDataFile(t, spec, "mem://default/changelog-eq/data.parquet", map[int]any{1000: int32(1)}) + eqDelete := newChangelogEqDelete(t, spec, + "mem://default/changelog-eq/eq.parquet", map[int]any{1000: int32(1)}) + tbl := writeChangelogScanTable(t, "changelog-eq", []changelogScanStep{ + { + operation: OpAppend, + data: []iceberg.ManifestEntry{changelogManifestEntry(iceberg.EntryStatusADDED, 1, 1, data)}, + }, + { + operation: OpDelete, + deletes: []iceberg.ManifestEntry{changelogManifestEntry(iceberg.EntryStatusADDED, 2, 2, eqDelete)}, + }, + }) + + tasks, err := tbl.NewIncrementalChangelogScan().PlanFiles(context.Background()) + require.NoError(t, err) + require.Len(t, tasks, 2) + rows, ok := tasks[1].(DeletedRowsScanTask) + require.True(t, ok) + require.Equal(t, []string{eqDelete.FilePath()}, changelogFilePaths(rows.AddedDeletes())) + require.Empty(t, rows.ScanTask().EqualityDeleteFiles) + require.Equal(t, iceberg.EntryContentEqDeletes, rows.AddedDeletes()[0].ContentType()) +} + +func TestIncrementalChangelogScanPlansDeletionVectors(t *testing.T) { + spec := partitionedSpec() + data := newTestDataFile(t, spec, "mem://default/changelog-dv/data.parquet", map[int]any{1000: int32(1)}) + posDelete := newTestPosDeleteFileForSpec(t, spec, + "mem://default/changelog-dv/pos.parquet", map[int]any{1000: int32(1)}, data.FilePath()) + dv := newChangelogDeletionVector(t, spec, "mem://default/changelog-dv/dv.puffin", data.FilePath()) + tbl := writeChangelogScanTable(t, "changelog-dv", []changelogScanStep{ + { + operation: OpAppend, + data: []iceberg.ManifestEntry{changelogManifestEntry(iceberg.EntryStatusADDED, 1, 1, data)}, + }, + { + operation: OpDelete, + deletes: []iceberg.ManifestEntry{changelogManifestEntry(iceberg.EntryStatusADDED, 2, 2, posDelete)}, + }, + { + operation: OpDelete, + manifestVersion: 3, + deletes: []iceberg.ManifestEntry{changelogManifestEntry(iceberg.EntryStatusADDED, 3, 3, dv)}, + }, + }) + + tasks, err := tbl.NewIncrementalChangelogScan().PlanFiles(context.Background()) + require.NoError(t, err) + require.Len(t, tasks, 3) + rows, ok := tasks[2].(DeletedRowsScanTask) + require.True(t, ok) + require.Equal(t, []string{dv.FilePath()}, changelogFilePaths(rows.AddedDeletes())) + require.Equal(t, []string{posDelete.FilePath()}, changelogFilePaths(rows.ExistingDeletes())) + require.Empty(t, rows.ScanTask().DeletionVectorFiles) +} + +func TestIncrementalChangelogScanAppliesExistingDeletesToRemovedFiles(t *testing.T) { + spec := partitionedSpec() + data := newTestDataFile(t, spec, "mem://default/changelog-removed/data.parquet", map[int]any{1000: int32(1)}) + posDelete := newTestPosDeleteFileForSpec(t, spec, + "mem://default/changelog-removed/pos.parquet", map[int]any{1000: int32(1)}, data.FilePath()) + tbl := writeChangelogScanTable(t, "changelog-removed", []changelogScanStep{ + { + operation: OpAppend, + data: []iceberg.ManifestEntry{changelogManifestEntry(iceberg.EntryStatusADDED, 1, 1, data)}, + }, + { + operation: OpDelete, + deletes: []iceberg.ManifestEntry{changelogManifestEntry(iceberg.EntryStatusADDED, 2, 2, posDelete)}, + }, + { + operation: OpOverwrite, + data: []iceberg.ManifestEntry{changelogManifestEntry(iceberg.EntryStatusDELETED, 3, 1, data)}, + }, + }) + + tasks, err := tbl.NewIncrementalChangelogScan().PlanFiles(context.Background()) + require.NoError(t, err) + require.Len(t, tasks, 3) + removed, ok := tasks[2].(DeletedDataFileScanTask) + require.True(t, ok) + require.Equal(t, ChangelogOpDelete, removed.Operation()) + require.Equal(t, 2, removed.ChangeOrdinal()) + require.Equal(t, []string{posDelete.FilePath()}, changelogFilePaths(removed.ExistingDeletes())) + + bounded, err := tbl.NewIncrementalChangelogScan(). + FromSnapshotExclusive(2). + ToSnapshot(3). + PlanFiles(context.Background()) + require.NoError(t, err) + require.Len(t, bounded, 1) + removed, ok = bounded[0].(DeletedDataFileScanTask) + require.True(t, ok) + require.Equal(t, []string{posDelete.FilePath()}, changelogFilePaths(removed.ExistingDeletes())) +} + +func TestIncrementalChangelogScanFiltersDeleteBackedTasks(t *testing.T) { + tbl := incrementalChangelogDeleteManifestTable(t) + filter := iceberg.EqualTo(iceberg.Reference("id"), int32(3)) + + tasks, err := tbl.NewIncrementalChangelogScan(WithRowFilter(filter)).PlanFiles(context.Background()) + require.NoError(t, err) + require.Len(t, tasks, 1) + require.Equal(t, "mem://default/changelog-delete/data-3.parquet", tasks[0].ScanTask().File.FilePath()) + require.Equal(t, ChangelogOpInsert, tasks[0].Operation()) +} + +func TestIncrementalChangelogScanFiltersDeletedRowsByPartition(t *testing.T) { + spec := partitionedSpec() + partOne := newTestDataFile(t, spec, "mem://default/changelog-part/data-1.parquet", map[int]any{1000: int32(1)}) + partTwo := newTestDataFile(t, spec, "mem://default/changelog-part/data-2.parquet", map[int]any{1000: int32(2)}) + deleteOne := newTestPosDeleteFileForSpec(t, spec, + "mem://default/changelog-part/delete-1.parquet", map[int]any{1000: int32(1)}, partOne.FilePath()) + deleteTwo := newTestPosDeleteFileForSpec(t, spec, + "mem://default/changelog-part/delete-2.parquet", map[int]any{1000: int32(2)}, partTwo.FilePath()) + tbl := writeChangelogScanTable(t, "changelog-part", []changelogScanStep{ + { + operation: OpAppend, + data: []iceberg.ManifestEntry{ + changelogManifestEntry(iceberg.EntryStatusADDED, 1, 1, partOne), + changelogManifestEntry(iceberg.EntryStatusADDED, 1, 1, partTwo), + }, + }, + { + operation: OpDelete, + deletes: []iceberg.ManifestEntry{ + changelogManifestEntry(iceberg.EntryStatusADDED, 2, 2, deleteOne), + changelogManifestEntry(iceberg.EntryStatusADDED, 2, 2, deleteTwo), + }, + }, + }) + + tasks, err := tbl.NewIncrementalChangelogScan( + WithRowFilter(iceberg.EqualTo(iceberg.Reference("id"), int32(1))), + ).PlanFiles(context.Background()) + require.NoError(t, err) + + var deleted []DeletedRowsScanTask + for _, task := range tasks { + require.NotEqual(t, partTwo.FilePath(), task.ScanTask().File.FilePath()) + rows, ok := task.(DeletedRowsScanTask) + if !ok { + continue + } + deleted = append(deleted, rows) + } + require.Len(t, deleted, 1) + require.Equal(t, partOne.FilePath(), deleted[0].ScanTask().File.FilePath()) + require.Equal(t, []string{deleteOne.FilePath()}, changelogFilePaths(deleted[0].AddedDeletes())) +} + +func TestIncrementalChangelogScanAppliesDeletesBySequence(t *testing.T) { + spec := partitionedSpec() + data := newTestDataFile(t, spec, "mem://default/changelog-seq/data.parquet", map[int]any{1000: int32(1)}) + dataSeq := int64(5) + + tests := []struct { + name string + kind string + deleteSeq int64 + want bool + }{ + {name: "position delete below data sequence", kind: "pos", deleteSeq: 4, want: false}, + {name: "position delete at data sequence", kind: "pos", deleteSeq: 5, want: true}, + {name: "position delete above data sequence", kind: "pos", deleteSeq: 6, want: true}, + {name: "equality delete below data sequence", kind: "eq", deleteSeq: 4, want: false}, + {name: "equality delete at data sequence", kind: "eq", deleteSeq: 5, want: false}, + {name: "equality delete above data sequence", kind: "eq", deleteSeq: 6, want: true}, + {name: "deletion vector below data sequence", kind: "dv", deleteSeq: 4, want: false}, + {name: "deletion vector at data sequence", kind: "dv", deleteSeq: 5, want: true}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + var deleteFile iceberg.DataFile + version := 2 + switch tt.kind { + case "pos": + deleteFile = newTestPosDeleteFileForSpec(t, spec, + "mem://default/changelog-seq/pos.parquet", map[int]any{1000: int32(1)}, data.FilePath()) + case "eq": + deleteFile = newChangelogEqDelete(t, spec, + "mem://default/changelog-seq/eq.parquet", map[int]any{1000: int32(1)}) + default: + deleteFile = newChangelogDeletionVector(t, spec, "mem://default/changelog-seq/dv.puffin", data.FilePath()) + version = 3 + } + tbl := writeChangelogScanTable(t, "changelog-seq", []changelogScanStep{ + { + operation: OpAppend, + data: []iceberg.ManifestEntry{changelogManifestEntry(iceberg.EntryStatusADDED, 1, dataSeq, data)}, + }, + { + operation: OpDelete, + manifestVersion: version, + deletes: []iceberg.ManifestEntry{changelogManifestEntry(iceberg.EntryStatusADDED, 2, tt.deleteSeq, deleteFile)}, + }, + }) + + tasks, err := tbl.NewIncrementalChangelogScan().PlanFiles(context.Background()) + require.NoError(t, err) + if !tt.want { + require.Len(t, tasks, 1) + require.Equal(t, ChangelogOpInsert, tasks[0].Operation()) + + return + } + require.Len(t, tasks, 2) + rows, ok := tasks[1].(DeletedRowsScanTask) + require.True(t, ok) + require.Equal(t, []string{deleteFile.FilePath()}, changelogFilePaths(rows.AddedDeletes())) + }) + } +} + +func TestIncrementalChangelogScanAppliesEveryDeleteInOneSnapshot(t *testing.T) { + spec := partitionedSpec() + data := newTestDataFile(t, spec, "mem://default/changelog-multi/data.parquet", map[int]any{1000: int32(1)}) + first := newTestPosDeleteFileForSpec(t, spec, + "mem://default/changelog-multi/first.parquet", map[int]any{1000: int32(1)}, data.FilePath()) + second := newTestPosDeleteFileForSpec(t, spec, + "mem://default/changelog-multi/second.parquet", map[int]any{1000: int32(1)}, data.FilePath()) + tbl := writeChangelogScanTable(t, "changelog-multi", []changelogScanStep{ + { + operation: OpAppend, + data: []iceberg.ManifestEntry{changelogManifestEntry(iceberg.EntryStatusADDED, 1, 1, data)}, + }, + { + operation: OpDelete, + deletes: []iceberg.ManifestEntry{ + changelogManifestEntry(iceberg.EntryStatusADDED, 2, 2, first), + changelogManifestEntry(iceberg.EntryStatusADDED, 2, 2, second), + }, + }, + }) + + tasks, err := tbl.NewIncrementalChangelogScan().PlanFiles(context.Background()) + require.NoError(t, err) + require.Len(t, tasks, 2) + rows, ok := tasks[1].(DeletedRowsScanTask) + require.True(t, ok) + require.ElementsMatch(t, []string{first.FilePath(), second.FilePath()}, changelogFilePaths(rows.AddedDeletes())) +} + +func TestIncrementalChangelogScanFoldsReplaceDeleteRewrites(t *testing.T) { + spec := partitionedSpec() + data := newTestDataFile(t, spec, "mem://default/changelog-replace-del/data.parquet", map[int]any{1000: int32(1)}) + oldDelete := newTestPosDeleteFileForSpec(t, spec, + "mem://default/changelog-replace-del/old.parquet", map[int]any{1000: int32(1)}, data.FilePath()) + newDelete := newTestPosDeleteFileForSpec(t, spec, + "mem://default/changelog-replace-del/new.parquet", map[int]any{1000: int32(1)}, data.FilePath()) + tbl := writeChangelogScanTable(t, "changelog-replace-del", []changelogScanStep{ + { + operation: OpAppend, + data: []iceberg.ManifestEntry{changelogManifestEntry(iceberg.EntryStatusADDED, 1, 1, data)}, + }, + { + operation: OpDelete, + deletes: []iceberg.ManifestEntry{changelogManifestEntry(iceberg.EntryStatusADDED, 2, 2, oldDelete)}, + }, + { + operation: OpReplace, + deletes: []iceberg.ManifestEntry{ + changelogManifestEntry(iceberg.EntryStatusDELETED, 3, 2, oldDelete), + changelogManifestEntry(iceberg.EntryStatusADDED, 3, 3, newDelete), + }, + }, + { + operation: OpOverwrite, + data: []iceberg.ManifestEntry{changelogManifestEntry(iceberg.EntryStatusDELETED, 4, 1, data)}, + }, + }) + + tasks, err := tbl.NewIncrementalChangelogScan().PlanFiles(context.Background()) + require.NoError(t, err) + require.Len(t, tasks, 3) + for _, task := range tasks { + require.NotEqual(t, int64(3), task.CommitSnapshotID()) + } + removed, ok := tasks[2].(DeletedDataFileScanTask) + require.True(t, ok) + require.Equal(t, []string{newDelete.FilePath()}, changelogFilePaths(removed.ExistingDeletes())) +} + +func TestIncrementalChangelogScanKeepsSiblingDeletionVector(t *testing.T) { + spec := partitionedSpec() + dataA := newTestDataFile(t, spec, "mem://default/changelog-puffin/a.parquet", map[int]any{1000: int32(1)}) + dataB := newTestDataFile(t, spec, "mem://default/changelog-puffin/b.parquet", map[int]any{1000: int32(2)}) + puffin := "mem://default/changelog-puffin/both.puffin" + dvA := newSharedDeletionVector(t, spec, puffin, dataA.FilePath(), 0) + dvB := newSharedDeletionVector(t, spec, puffin, dataB.FilePath(), 64) + rewrittenA := newSharedDeletionVector(t, spec, "mem://default/changelog-puffin/a.puffin", dataA.FilePath(), 0) + later := newTestPosDeleteFileForSpec(t, spec, + "mem://default/changelog-puffin/later.parquet", map[int]any{1000: int32(2)}, dataB.FilePath()) + tbl := writeChangelogScanTable(t, "changelog-puffin", []changelogScanStep{ + { + operation: OpAppend, + data: []iceberg.ManifestEntry{ + changelogManifestEntry(iceberg.EntryStatusADDED, 1, 1, dataA), + changelogManifestEntry(iceberg.EntryStatusADDED, 1, 1, dataB), + }, + }, + { + operation: OpDelete, + manifestVersion: 3, + deletes: []iceberg.ManifestEntry{ + changelogManifestEntry(iceberg.EntryStatusADDED, 2, 2, dvA), + changelogManifestEntry(iceberg.EntryStatusADDED, 2, 2, dvB), + }, + }, + { + operation: OpDelete, + manifestVersion: 3, + deletes: []iceberg.ManifestEntry{ + changelogManifestEntry(iceberg.EntryStatusDELETED, 3, 2, dvA), + changelogManifestEntry(iceberg.EntryStatusADDED, 3, 3, rewrittenA), + }, + }, + { + operation: OpDelete, + manifestVersion: 3, + deletes: []iceberg.ManifestEntry{changelogManifestEntry(iceberg.EntryStatusADDED, 4, 4, later)}, + }, + }) + + tasks, err := tbl.NewIncrementalChangelogScan().PlanFiles(context.Background()) + require.NoError(t, err) + var got DeletedRowsScanTask + var found bool + for _, task := range tasks { + rows, ok := task.(DeletedRowsScanTask) + if !ok || rows.ScanTask().File.FilePath() != dataB.FilePath() || rows.CommitSnapshotID() != 4 { + continue + } + got = rows + found = true + } + require.True(t, found) + require.Equal(t, []string{later.FilePath()}, changelogFilePaths(got.AddedDeletes())) + require.Equal(t, []string{dvB.FilePath()}, changelogFilePaths(got.ExistingDeletes())) + require.Equal(t, dataB.FilePath(), referencedDataFilePath(got.ExistingDeletes()[0])) +} + +func TestIncrementalChangelogScanPlansOverwriteWithExpiredParent(t *testing.T) { + spec := partitionedSpec() + txn, fs := createTestTransactionWithMemIO(t, spec) + schema := simpleSchema() + data := newTestDataFile(t, spec, "mem://default/changelog-expired-overwrite/data.parquet", map[int]any{1000: int32(1)}) + snapshotID := int64(2) + dataManifest := writeChangelogDataManifest(t, fs, spec, schema, + "mem://default/changelog-expired-overwrite/metadata/data.avro", snapshotID, 2, + []iceberg.ManifestEntry{changelogManifestEntry(iceberg.EntryStatusDELETED, snapshotID, 1, data)}) + listPath := "mem://default/changelog-expired-overwrite/metadata/snap-2.avro" + writeChangelogManifestList(t, fs, listPath, snapshotID, 2, []iceberg.ManifestFile{dataManifest}) + + expiredParent := int64(1) + txn.meta.snapshotList = []Snapshot{{ + SnapshotID: snapshotID, + ParentSnapshotID: &expiredParent, + TimestampMs: 2000, + ManifestList: listPath, + SequenceNumber: snapshotID, + SchemaID: &schema.ID, + Summary: &Summary{Operation: OpOverwrite}, + }} + txn.meta.snapshotLog = []SnapshotLogEntry{{SnapshotID: snapshotID, TimestampMs: 2000}} + txn.meta.currentSnapshotID = &snapshotID + meta, err := txn.meta.Build() + require.NoError(t, err) + tbl := New(Identifier{"changelog-expired-overwrite"}, meta, "metadata.json", func(context.Context) (iceio.IO, error) { + return fs, nil + }, nil) + + tasks, err := tbl.NewIncrementalChangelogScan(). + FromSnapshotExclusive(expiredParent). + ToSnapshot(snapshotID). + PlanFiles(context.Background()) + require.NoError(t, err) + require.Len(t, tasks, 1) + _, ok := tasks[0].(DeletedDataFileScanTask) + require.True(t, ok) +} + +func TestIncrementalChangelogScanKeepsCarriedDeleteWhenParentExpired(t *testing.T) { + spec := partitionedSpec() + txn, fs := createTestTransactionWithMemIO(t, spec) + schema := simpleSchema() + data := newTestDataFile(t, spec, "mem://default/changelog-expired-delete/data.parquet", map[int]any{1000: int32(1)}) + pos := newTestPosDeleteFileForSpec(t, spec, + "mem://default/changelog-expired-delete/pos.parquet", map[int]any{1000: int32(1)}, data.FilePath()) + snapshotID := int64(2) + expiredParent := int64(1) + dataManifest := writeChangelogDataManifest(t, fs, spec, schema, + "mem://default/changelog-expired-delete/metadata/data.avro", snapshotID, 2, + []iceberg.ManifestEntry{changelogManifestEntry(iceberg.EntryStatusDELETED, snapshotID, 1, data)}) + deleteManifest := writeChangelogDeleteManifest(t, fs, spec, schema, + "mem://default/changelog-expired-delete/metadata/deletes.avro", snapshotID, 2, + []iceberg.ManifestEntry{changelogManifestEntry(iceberg.EntryStatusEXISTING, expiredParent, 1, pos)}) Review Comment: Both expired-parent tests have a single snapshot and only hit the `carried` side of `entriesLiveBefore`, so removing the `removedHere` clause wouldn't fail anything. After `expire_snapshots` on a table that compacts, the common shape is a first kept snapshot that removes a data file together with its pos delete. Its `DeletedDataFileScanTask` should still list that delete in `ExistingDeletes`. I'd add that case once with `FromSnapshotExclusive` and once inclusive, plus one where a replace rewriting a DV is the first timeline snapshot. A small table-driven test on `entriesLiveBefore` itself (carried, removed here, added here excluded, duplicate key) would be cheap too. -- 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]
