This is an automated email from the ASF dual-hosted git repository.
laskoviymishka pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg-go.git
The following commit(s) were added to refs/heads/main by this push:
new 2a51abaf7 feat(table): plan delete-file changes in incremental
changelog scans (#2169)
2a51abaf7 is described below
commit 2a51abaf7490d9f1714785a5be2b144cd0b727b4
Author: Digvijay <[email protected]>
AuthorDate: Fri Oct 30 17:26:57 2026 -0500
feat(table): plan delete-file changes in incremental changelog scans (#2169)
---
table/incremental_changelog_deletes.go | 940 +++++++++++++++++++++
table/incremental_changelog_scan.go | 186 +++--
table/incremental_changelog_scan_test.go | 1330 +++++++++++++++++++++++++++++-
3 files changed, 2363 insertions(+), 93 deletions(-)
diff --git a/table/incremental_changelog_deletes.go
b/table/incremental_changelog_deletes.go
new file mode 100644
index 000000000..fb57a0bdc
--- /dev/null
+++ b/table/incremental_changelog_deletes.go
@@ -0,0 +1,940 @@
+// 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
+ // beforeEntries is each snapshot's delete set strictly before that
+ // snapshot. afterEntries is the set after the last timeline snapshot.
+ beforeEntries map[int64][]iceberg.ManifestEntry
+ afterEntries []iceberg.ManifestEntry
+}
+
+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. This is exact when that snapshot
+// records a DELETED entry, under its own snapshot id, for every delete file
+// it removed, and carried entries keep the snapshot id that added them. A
+// writer that drops a fully removed delete manifest, instead of recording a
+// DELETED entry, cannot be reconstructed. When the snapshot summary reports
+// more removed-delete-files than the DELETED delete entries in its manifests,
+// planning returns an error instead of dropping those deletes. That count is
+// taken before partition pruning, so a filtered scan does not look like a
+// missing delete file.
+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
+ }
+ raw, err := readManifestEntries(ctx, plan.fs,
deleteContentManifests(manifests), false, false)
+ if err != nil {
+ return nil, err
+ }
+ if err := checkRemovedDeleteFiles(snapshot, raw); err != nil {
+ return nil, err
+ }
+ entries, err := s.readDeleteManifestEntries(ctx, plan, manifests)
+ if err != nil {
+ return nil, err
+ }
+
+ return entriesLiveBefore(entries, snapshot.SnapshotID), nil
+}
+
+func deleteContentManifests(manifests []iceberg.ManifestFile)
[]iceberg.ManifestFile {
+ out := make([]iceberg.ManifestFile, 0)
+ for _, manifest := range manifests {
+ if manifest.ManifestContent() == iceberg.ManifestContentDeletes
{
+ out = append(out, manifest)
+ }
+ }
+
+ return out
+}
+
+// checkRemovedDeleteFiles reports an error when the snapshot summary claims
+// more removed delete files than the DELETED delete entries recorded under
+// that snapshot id. A snapshot summary without removed-delete-files passes
+// the guard; the guard only fires when the summary claims more removals than
+// the DELETED entries recorded. Entries must be the unfiltered
+// delete-manifest contents.
+func checkRemovedDeleteFiles(snapshot Snapshot, entries
[]iceberg.ManifestEntry) error {
+ if snapshot.Summary == nil || len(snapshot.Summary.Properties) == 0 {
+ return nil
+ }
+ raw, ok := snapshot.Summary.Properties[removedDeleteFilesKey]
+ if !ok {
+ return nil
+ }
+ reported, err := strconv.Atoi(raw)
+ if err != nil {
+ return fmt.Errorf("%w: snapshot %d removed-delete-files %q",
iceberg.ErrInvalidArgument, snapshot.SnapshotID, raw)
+ }
+ recorded := 0
+ for _, entry := range entries {
+ if entry.Status() != iceberg.EntryStatusDELETED ||
entry.SnapshotID() != snapshot.SnapshotID {
+ continue
+ }
+ if entry.DataFile().ContentType() == iceberg.EntryContentData {
+ continue
+ }
+ recorded++
+ }
+ if reported > recorded {
+ return fmt.Errorf("%w: snapshot %d %s is %d but recorded %d
deleted delete entries",
+ iceberg.ErrInvalidArgument, snapshot.SnapshotID,
removedDeleteFilesKey, reported, recorded)
+ }
+
+ return nil
+}
+
+func (s *IncrementalChangelogScan) readDeleteManifestEntries(
+ ctx context.Context, plan changelogTaskPlan, manifests
[]iceberg.ManifestFile,
+) ([]iceberg.ManifestEntry, error) {
+ deleteManifests := deleteContentManifests(manifests)
+ // Prune with the scan partition filters. A local accumulator keeps this
+ // pass from counting the same delete manifests in the changelog report.
+ var pruned scanMetricsAccumulator
+ deleteManifests, err := plan.scan.filterManifestsWithSchemaOptions(
+ deleteManifests, plan.schema, &pruned, plan.partitionFilters,
true)
+ if err != nil {
+ return nil, err
+ }
+
+ return readManifestEntries(ctx, plan.fs, deleteManifests, false, false)
+}
+
+// snapshotBeforeRange returns the snapshot whose live delete files already
+// applied before the range. An exclusive start that is still present is that
+// snapshot. When it has been expired, the parent of the first snapshot in the
+// range is used instead. A nil result with a recorded parent means that parent
+// has expired.
+func (s *IncrementalChangelogScan) snapshotBeforeRange(snapshots []Snapshot)
*Snapshot {
+ if s.fromSnapshotID != nil && !s.fromInclusive {
+ if snapshot := s.scan.metadata.SnapshotByID(*s.fromSnapshotID);
snapshot != nil {
+ return snapshot
+ }
+ }
+ if len(snapshots) == 0 || snapshots[0].ParentSnapshotID == nil {
+ return nil
+ }
+
+ return s.scan.metadata.SnapshotByID(*snapshots[0].ParentSnapshotID)
+}
+
+func timelineSnapshots(plan changelogTaskPlan) []Snapshot {
+ if len(plan.timeline) > 0 {
+ return plan.timeline
+ }
+
+ return plan.snapshots
+}
+
+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
+ }
+ // Partition and metrics evaluators are constant for the scan, so a
manifest
+ // pruned while it is opened can be reused for every later snapshot.
+ partitionFilter, metricsEval, err :=
plan.scan.changelogDataFilters(plan.schema, plan.partitionFilters)
+ if err != nil {
+ return nil, err
+ }
+
+ var tasks []plannedChangelogTask
+ decodedManifests := make(map[string][]iceberg.ManifestEntry)
+ sequences := snapshotSequences(plan.scan.metadata)
+ 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
+ }
+ // Each snapshot that adds deletes reads that snapshot's live
data
+ // manifests. Equality deletes and position deletes without a
referenced
+ // data file cannot be narrowed to one path, so those snapshots
scan the
+ // manifests the partition filter keeps. Entries that fail the
scan's
+ // partition or metrics evaluators are dropped while the
manifest is
+ // opened, then cached by path.
+ live, err := liveDataEntries(ctx, plan.fs, dataManifests,
decodedManifests, sequences, partitionFilter, metricsEval)
+ 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{}{}
+
+ 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. Each snapshot's set is cached from the previous snapshot's
+// result: drop the deletes that snapshot removed, then append the deletes it
+// added. A delete removed and added again in one snapshot stays live as the
+// added entry. keepNewestDeletionVector runs on each result.
+func (st *changelogDeleteState) entriesBefore(snapshotID int64)
[]iceberg.ManifestEntry {
+ if st.beforeEntries == nil {
+ st.beforeEntries = make(map[int64][]iceberg.ManifestEntry,
len(st.snapshots))
+ running := keepNewestDeletionVector(slices.Clone(st.preRange))
+ for _, snapshot := range st.snapshots {
+ st.beforeEntries[snapshot.SnapshotID] = running
+ running = nextDeleteEntries(running,
st.added[snapshot.SnapshotID], st.removed[snapshot.SnapshotID])
+ }
+ st.afterEntries = running
+ }
+ if entries, ok := st.beforeEntries[snapshotID]; ok {
+ return entries
+ }
+
+ return st.afterEntries
+}
+
+// nextDeleteEntries returns before with removed identities dropped and added
+// entries appended. The result does not alias before, so a cached set stays
+// intact when a later snapshot adds deletes.
+func nextDeleteEntries(before, added []iceberg.ManifestEntry, removed
map[string]struct{}) []iceberg.ManifestEntry {
+ next := dropManifestEntries(before, removed)
+ if len(added) == 0 {
+ return keepNewestDeletionVector(next)
+ }
+ out := make([]iceberg.ManifestEntry, len(next), len(next)+len(added))
+ copy(out, next)
+ out = append(out, added...)
+
+ 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
+}
+
+// buildChangelogDeleteIndex groups delete files with the scan's sequence
+// rules. Equality deletes are resolved against schema, which is the schema at
+// the end of the changelog range. An equality delete whose columns were
+// dropped before that schema cannot be resolved.
+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
+}
+
+// changelogDataFilters returns the partition and metrics evaluators applied
+// while a data manifest is opened. Both are constant for one scan.
+func (scan *Scan) changelogDataFilters(
+ schema *iceberg.Schema,
+ partitionFilters *keyDefaultMapErr[int, iceberg.BooleanExpression],
+) (func(iceberg.DataFile) (bool, error), 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, nil, err
+ }
+ partitionEvaluators := newKeyDefaultMapWrapErr(func(specID int)
(func(iceberg.DataFile) (bool, error), error) {
+ return buildPartitionEvaluator(specID, scan.metadata, schema,
partitionFilters, scan.caseSensitive)
+ })
+ partitionFilter := 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)
+ }
+
+ return partEval(file)
+ }
+
+ return partitionFilter, metricsEval, nil
+}
+
+func liveDataEntries(
+ ctx context.Context,
+ fs io.IO,
+ manifests []iceberg.ManifestFile,
+ decoded map[string][]iceberg.ManifestEntry,
+ sequences snapshotSequenceLookup,
+ partitionFilter, metricsEval func(iceberg.DataFile) (bool, error),
+) ([]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 := readCachedManifestEntries(ctx, fs, dataManifests,
decoded, partitionFilter, metricsEval)
+ 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, sequences), nil
+}
+
+// snapshotSequenceLookup returns the sequence number of the snapshot that
+// wrote an entry. Snapshot IDs are random, so they are not an order. The
+// boolean is false when that snapshot is not in table metadata.
+type snapshotSequenceLookup func(snapshotID int64) (sequence int64, ok bool)
+
+func snapshotSequences(metadata Metadata) snapshotSequenceLookup {
+ return func(snapshotID int64) (int64, bool) {
+ snapshot := metadata.SnapshotByID(snapshotID)
+ if snapshot == nil {
+ return 0, false
+ }
+
+ return snapshot.SequenceNumber, true
+ }
+}
+
+// liveManifestEntries keeps the winning entry for each data-file path and
+// drops paths whose winning entry is a deletion. A higher entry sequence
+// number wins. Equal entry sequence numbers use the sequence number of the
+// snapshot that wrote each entry. When those are equal or unknown, a deletion
+// wins.
+func liveManifestEntries(entries []iceberg.ManifestEntry, sequences
snapshotSequenceLookup) []iceberg.ManifestEntry {
+ return liveEntriesByKey(entries, func(entry iceberg.ManifestEntry)
string {
+ return entry.DataFile().FilePath()
+ }, sequences)
+}
+
+// liveDeleteEntries is liveManifestEntries for delete files. Deletion vectors
+// that share a Puffin path stay distinct, keyed by that path plus the
+// referenced data file and content offset.
+func liveDeleteEntries(entries []iceberg.ManifestEntry, sequences
snapshotSequenceLookup) []iceberg.ManifestEntry {
+ return liveEntriesByKey(entries, func(entry iceberg.ManifestEntry)
string {
+ return deleteEntryKey(entry.DataFile())
+ }, sequences)
+}
+
+func liveEntriesByKey(entries []iceberg.ManifestEntry, key
func(iceberg.ManifestEntry) string, sequences snapshotSequenceLookup)
[]iceberg.ManifestEntry {
+ latest := make(map[string]iceberg.ManifestEntry, len(entries))
+ order := make([]string, 0, len(entries))
+ for _, entry := range entries {
+ id := key(entry)
+ current, seen := latest[id]
+ if !seen {
+ order = append(order, id)
+ latest[id] = entry
+
+ continue
+ }
+ if changelogEntrySupersedes(entry, current, sequences) {
+ latest[id] = entry
+ }
+ }
+ out := make([]iceberg.ManifestEntry, 0, len(latest))
+ for _, id := range order {
+ entry := latest[id]
+ if entry.Status() == iceberg.EntryStatusDELETED {
+ continue
+ }
+ out = append(out, entry)
+ }
+
+ return out
+}
+
+// changelogEntrySupersedes reports whether next replaces current. The entry
+// sequence number wins first. Equal entry sequence numbers use the sequence
+// number of the snapshot that wrote each entry. Snapshot IDs are not compared.
+// When the snapshot sequence numbers are equal, or either snapshot is unknown,
+// a deletion wins.
+func changelogEntrySupersedes(next, current iceberg.ManifestEntry, sequences
snapshotSequenceLookup) bool {
+ if next.DataSequenceNumber() != current.DataSequenceNumber() {
+ return next.DataSequenceNumber() > current.DataSequenceNumber()
+ }
+ if sequences != nil {
+ nextSequence, nextOK := sequences(next.SnapshotID())
+ currentSequence, currentOK := sequences(current.SnapshotID())
+ if nextOK && currentOK && nextSequence != currentSequence {
+ return nextSequence > currentSequence
+ }
+ }
+
+ return next.Status() == iceberg.EntryStatusDELETED && current.Status()
!= iceberg.EntryStatusDELETED
+}
+
+// entriesLiveBefore selects delete entries that were live before snapshotID.
+// A non-deleted entry written by another snapshot was carried forward. A
+// deleted entry written by snapshotID was live until this snapshot removed it.
+// One entry is kept per delete identity, preferring a non-deleted entry.
+func entriesLiveBefore(entries []iceberg.ManifestEntry, snapshotID int64)
[]iceberg.ManifestEntry {
+ latest := make(map[string]iceberg.ManifestEntry)
+ order := make([]string, 0)
+ for _, entry := range entries {
+ carried := entry.Status() != iceberg.EntryStatusDELETED &&
entry.SnapshotID() != snapshotID
+ removedHere := entry.Status() == iceberg.EntryStatusDELETED &&
entry.SnapshotID() == snapshotID
+ if !carried && !removedHere {
+ continue
+ }
+ key := deleteEntryKey(entry.DataFile())
+ current, seen := latest[key]
+ if !seen {
+ order = append(order, key)
+ latest[key] = entry
+
+ continue
+ }
+ if current.Status() == iceberg.EntryStatusDELETED &&
entry.Status() != iceberg.EntryStatusDELETED {
+ latest[key] = entry
+ }
+ }
+ out := make([]iceberg.ManifestEntry, 0, len(order))
+ for _, key := range order {
+ out = append(out, latest[key])
+ }
+
+ return out
+}
+
+// deleteEntryKey identifies a delete file. Position and equality deletes use
+// their path. Deletion vectors in one Puffin file share that path, so they
+// also use the referenced data file and content offset.
+func deleteEntryKey(file iceberg.DataFile) string {
+ if !IsDeletionVector(file) {
+ return file.FilePath()
+ }
+ offset := int64(-1)
+ if fileOffset := file.ContentOffset(); fileOffset != nil {
+ offset = *fileOffset
+ }
+
+ return file.FilePath() + "\x00" + referencedDataFilePath(file) + "\x00"
+ strconv.FormatInt(offset, 10)
+}
+
+// readCachedManifestEntries returns manifest entries, reusing entries already
+// decoded for the same manifest path. Manifest files are immutable, and the
+// partition and metrics evaluators do not change during the scan, so the
+// pruned entries stay valid for every snapshot that carries the manifest.
+func readCachedManifestEntries(
+ ctx context.Context,
+ fs io.IO,
+ manifests []iceberg.ManifestFile,
+ decoded map[string][]iceberg.ManifestEntry,
+ partitionFilter, metricsEval func(iceberg.DataFile) (bool, error),
+) ([]iceberg.ManifestEntry, error) {
+ if len(manifests) == 0 {
+ return nil, nil
+ }
+ out := make([]iceberg.ManifestEntry, 0)
+ for _, manifest := range manifests {
+ if err := ctx.Err(); err != nil {
+ return nil, err
+ }
+ path := manifest.FilePath()
+ entries, ok := decoded[path]
+ if !ok {
+ var err error
+ entries, err = openManifestWithOptions(fs, manifest,
partitionFilter, metricsEval, false, false)
+ if err != nil {
+ return nil, err
+ }
+ if entries == nil {
+ entries = []iceberg.ManifestEntry{}
+ }
+ if decoded != nil {
+ decoded[path] = entries
+ }
+ }
+ out = append(out, entries...)
+ }
+
+ return out, nil
+}
+
+func readManifestEntries(
+ ctx context.Context,
+ fs io.IO,
+ manifests []iceberg.ManifestFile,
+ discardDeleted, discardExisting bool,
+) ([]iceberg.ManifestEntry, error) {
+ if len(manifests) == 0 {
+ return nil, nil
+ }
+ keep := func(iceberg.DataFile) (bool, error) { return true, nil }
+ out := make([]iceberg.ManifestEntry, 0)
+ for _, manifest := range manifests {
+ if err := ctx.Err(); err != nil {
+ return nil, err
+ }
+ entries, err := openManifestWithOptions(fs, manifest, keep,
keep, discardDeleted, discardExisting)
+ if err != nil {
+ return nil, err
+ }
+ out = append(out, entries...)
+ }
+
+ return out, nil
+}
+
+func dropManifestEntries(entries []iceberg.ManifestEntry, gone
map[string]struct{}) []iceberg.ManifestEntry {
+ if len(gone) == 0 {
+ return entries
+ }
+ out := make([]iceberg.ManifestEntry, 0, len(entries))
+ for _, entry := range entries {
+ if _, drop := gone[deleteEntryKey(entry.DataFile())]; drop {
+ continue
+ }
+ out = append(out, entry)
+ }
+
+ return out
+}
+
+// keepNewestDeletionVector leaves one deletion vector per referenced data
+// file. Later entries win. A deletion vector is cumulative, so the newest one
+// already contains positions deleted by the ones it replaced.
+func keepNewestDeletionVector(entries []iceberg.ManifestEntry)
[]iceberg.ManifestEntry {
+ latest := make(map[string]int)
+ drop := make(map[int]struct{})
+ for i, entry := range entries {
+ file := entry.DataFile()
+ if !IsDeletionVector(file) {
+ continue
+ }
+ ref := referencedDataFilePath(file)
+ if ref == "" {
+ continue
+ }
+ if prev, ok := latest[ref]; ok {
+ drop[prev] = struct{}{}
+ }
+ latest[ref] = i
+ }
+ if len(drop) == 0 {
+ return entries
+ }
+ out := make([]iceberg.ManifestEntry, 0, len(entries)-len(drop))
+ for i, entry := range entries {
+ if _, ok := drop[i]; ok {
+ continue
+ }
+ out = append(out, entry)
+ }
+
+ return out
+}
+
+func sortChangelogTasks(tasks []plannedChangelogTask) {
+ slices.SortFunc(tasks, func(left, right plannedChangelogTask) int {
+ if ordinal := cmp.Compare(left.task.ChangeOrdinal(),
right.task.ChangeOrdinal()); ordinal != 0 {
+ return ordinal
+ }
+ if operation :=
cmp.Compare(changelogOperationOrder(left.task.Operation()),
changelogOperationOrder(right.task.Operation())); operation != 0 {
+ return operation
+ }
+ if path := cmp.Compare(left.task.ScanTask().File.FilePath(),
right.task.ScanTask().File.FilePath()); path != 0 {
+ return path
+ }
+
+ return cmp.Compare(changelogTaskKindOrder(left.task),
changelogTaskKindOrder(right.task))
+ })
+}
+
+func changelogTaskKindOrder(task ChangelogScanTask) int {
+ switch task.(type) {
+ case DeletedDataFileScanTask:
+ return 0
+ case DeletedRowsScanTask:
+ return 1
+ case AddedRowsScanTask:
+ return 2
+ default:
+ return 3
+ }
+}
+
+// changelogMetricsTask counts delete files that produce a row-level change as
+// well as deletes already stored on the scan task. Each delete is counted once
+// per task, matching a normal scan report.
+func changelogMetricsTask(task ChangelogScanTask) FileScanTask {
+ file := task.ScanTask()
+ rows, ok := task.(DeletedRowsScanTask)
+ if !ok {
+ return file
+ }
+ file.DeleteFiles = slices.Clone(file.DeleteFiles)
+ file.EqualityDeleteFiles = slices.Clone(file.EqualityDeleteFiles)
+ file.DeletionVectorFiles = slices.Clone(file.DeletionVectorFiles)
+ for _, deleteFile := range rows.AddedDeletes() {
+ switch {
+ case IsDeletionVector(deleteFile):
+ file.DeletionVectorFiles =
append(file.DeletionVectorFiles, deleteFile)
+ case deleteFile.ContentType() == iceberg.EntryContentEqDeletes:
+ file.EqualityDeleteFiles =
append(file.EqualityDeleteFiles, deleteFile)
+ default:
+ file.DeleteFiles = append(file.DeleteFiles, deleteFile)
+ }
+ }
+
+ return file
+}
diff --git a/table/incremental_changelog_scan.go
b/table/incremental_changelog_scan.go
index 7ec37d123..d1918b60f 100644
--- a/table/incremental_changelog_scan.go
+++ b/table/incremental_changelog_scan.go
@@ -18,7 +18,6 @@
package table
import (
- "cmp"
"context"
"fmt"
"maps"
@@ -29,11 +28,16 @@ import (
"github.com/apache/iceberg-go/metrics"
)
-// IncrementalChangelogScan plans data-file changes between snapshots. It
-// emits insert and delete tasks for data-manifest entries and skips replace
-// snapshots. PlanFiles returns an error if any in-range snapshot's manifest
-// list references a delete manifest, including ones carried forward from
-// earlier snapshots.
+// IncrementalChangelogScan plans data-file and delete-file changes between
+// snapshots. It emits insert tasks for added data files, delete tasks for
+// removed data files, and row-level delete tasks for existing data files
+// affected by delete files added in the range. Replace snapshots are not
+// emitted. Delete files they add or remove still apply to later tasks.
+// Remote planning is not supported.
+//
+// Java's changelog scan rejects delete manifests. The delete-aware tasks
+// here, including same-snapshot position deletes and deletion vectors on an
+// added file, are iceberg-go behavior.
type IncrementalChangelogScan struct {
scan *Scan
fromSnapshotID *int64
@@ -43,16 +47,20 @@ type IncrementalChangelogScan struct {
type plannedChangelogTask struct {
task ChangelogScanTask
- file FileScanTask
}
// NewIncrementalChangelogScan creates an incremental changelog planner.
// Projection and row limits are not applied to returned tasks. Auto planning
// falls back to local planning, while remote planning, including a catalog
-// `server` scan-planning-mode directive, is not supported. Use
-// ChangelogScanTask.ScanTask with Scan.ReadTasks to read the returned files.
-// Row filters are attached to each task as residuals without
partition-specific
-// simplification, matching the existing incremental append scan behavior.
+// `server` scan-planning-mode directive, is not supported.
+//
+// ScanTask carries the data file and the deletes applied before the change is
+// interpreted: deletes committed with an added data file, or deletes that
+// already applied to a removed data file or a row-level delete. AddedDeletes
+// on a row-level delete are the deletes that produce the change and are not
+// part of ScanTask. Use ScanTask with Scan.ReadTasks to read the returned
+// files. Row filters are attached to each task as residuals without
+// partition-specific simplification, matching incremental append scans.
func (t Table) NewIncrementalChangelogScan(opts ...ScanOption)
*IncrementalChangelogScan {
return &IncrementalChangelogScan{scan: t.Scan(opts...)}
}
@@ -87,14 +95,16 @@ func (s *IncrementalChangelogScan) ToSnapshot(snapshotID
int64) *IncrementalChan
return &out
}
-// PlanFiles returns one task for each added or deleted data-file entry. A
-// cancelled context returns its cancellation error before planning starts.
-// Tasks are ordered by change ordinal, then by DELETE before INSERT within an
-// ordinal, and finally by data-file path. When an ending snapshot is
-// available, it emits a ScanReport through the configured reporter on
-// successful planning. Changelog reports count every returned task in
-// ResultDataFiles and TotalFileSizeInBytes, so a file inserted and deleted
-// within the range is counted twice.
+// PlanFiles returns one task for each added or deleted data file, plus one
+// row-level delete task for each existing data file affected by delete files
+// added in a snapshot. A cancelled context returns its cancellation error
+// before planning starts. Tasks are ordered by change ordinal, then by DELETE
+// before INSERT within an ordinal, and finally by data-file path. When an
+// ending snapshot is available, it emits a ScanReport through the configured
+// reporter on successful planning. Changelog reports count every returned task
+// in ResultDataFiles and TotalFileSizeInBytes, so a file inserted and deleted
+// within the range is counted twice. Delete files are counted once per task
+// they apply to.
func (s *IncrementalChangelogScan) PlanFiles(ctx context.Context)
([]ChangelogScanTask, error) {
if s == nil || s.scan == nil {
return nil, fmt.Errorf("%w: incremental changelog scan is not
initialized", ErrInvalidOperation)
@@ -144,8 +154,8 @@ func (s *IncrementalChangelogScan) PlanFiles(ctx
context.Context) ([]ChangelogSc
fileTasks := make([]FileScanTask, len(plannedTasks))
for i, planned := range plannedTasks {
tasks[i] = planned.task
- fileTasks[i] = planned.file
- acc.totalFileSize += planned.file.File.FileSizeBytes()
+ fileTasks[i] = changelogMetricsTask(planned.task)
+ acc.totalFileSize += fileTasks[i].File.FileSizeBytes()
}
acc.applyResultDeleteMetrics(fileTasks)
planningDuration := time.Since(start)
@@ -187,7 +197,9 @@ func (s *IncrementalChangelogScan) PlanFiles(ctx
context.Context) ([]ChangelogSc
}
manifestsByPath := make(map[string]iceberg.ManifestFile)
- for _, snapshot := range snapshots {
+ manifestsBySnapshot := make(map[int64][]iceberg.ManifestFile,
len(snapshots))
+ deleteManifestsBySnapshot := make(map[int64][]iceberg.ManifestFile)
+ for _, snapshot := range snapshotRange {
if err := ctx.Err(); err != nil {
return nil, err
}
@@ -195,18 +207,32 @@ func (s *IncrementalChangelogScan) PlanFiles(ctx
context.Context) ([]ChangelogSc
if err != nil {
return nil, err
}
+ replace := snapshot.Summary != nil &&
snapshot.Summary.Operation == OpReplace
+ if !replace {
+ manifestsBySnapshot[snapshot.SnapshotID] = manifests
+ }
for _, manifest := range manifests {
- if manifest.ManifestContent() ==
iceberg.ManifestContentDeletes {
- return nil, fmt.Errorf("%w: incremental
changelog scan range references a delete manifest originating in snapshot %d",
- ErrInvalidOperation,
manifest.SnapshotID())
- }
- if manifest.ManifestContent() !=
iceberg.ManifestContentData {
- continue
- }
- if _, ok :=
changelogSnapshotIDs[manifest.SnapshotID()]; !ok {
+ switch manifest.ManifestContent() {
+ case iceberg.ManifestContentDeletes:
+ // Only manifests written by this snapshot are
new delete-file
+ // changes. Carried-forward delete manifests
stay available as
+ // deletes that already applied. Replace
snapshots are included
+ // so a rewrite of delete files updates later
tasks.
+ if manifest.SnapshotID() == snapshot.SnapshotID
{
+
deleteManifestsBySnapshot[snapshot.SnapshotID] = append(
+
deleteManifestsBySnapshot[snapshot.SnapshotID], manifest)
+ }
+ case iceberg.ManifestContentData:
+ if replace {
+ continue
+ }
+ if _, ok :=
changelogSnapshotIDs[manifest.SnapshotID()]; !ok {
+ continue
+ }
+ manifestsByPath[manifest.FilePath()] = manifest
+ default:
continue
}
- manifestsByPath[manifest.FilePath()] = manifest
}
}
@@ -233,45 +259,37 @@ func (s *IncrementalChangelogScan) PlanFiles(ctx
context.Context) ([]ChangelogSc
if err != nil {
return nil, err
}
- if len(manifestList) == 0 {
- return finish(nil), nil
- }
- entries, err := planningScan.collectManifestEntriesWithSchemaOptions(
- ctx, manifestList, schema,
- partitionFilters,
- /* discardDeleted= */ false,
- /* discardExisting= */ true,
- )
- if err != nil {
- return nil, err
- }
-
- plannedTasks := make([]plannedChangelogTask, 0,
len(entries.dataEntries))
- for _, entry := range entries.dataEntries {
- ordinal, ok := snapshotOrdinals[entry.SnapshotID()]
- if !ok {
- continue
- }
-
- task, err := newChangelogScanTask(entry, ordinal, residual)
+ var dataEntries []iceberg.ManifestEntry
+ if len(manifestList) > 0 {
+ entries, err :=
planningScan.collectManifestEntriesWithSchemaOptions(
+ ctx, manifestList, schema,
+ partitionFilters,
+ /* discardDeleted= */ false,
+ /* discardExisting= */ true,
+ )
if err != nil {
- return nil, fmt.Errorf("incremental changelog scan
snapshot %d: %w", entry.SnapshotID(), err)
+ return nil, err
}
- plannedTasks = append(plannedTasks, plannedChangelogTask{
- task: task,
- file: task.ScanTask(),
- })
+ dataEntries = entries.dataEntries
}
- slices.SortFunc(plannedTasks, func(left, right plannedChangelogTask)
int {
- if ordinal := cmp.Compare(left.task.ChangeOrdinal(),
right.task.ChangeOrdinal()); ordinal != 0 {
- return ordinal
- }
- if operation :=
cmp.Compare(changelogOperationOrder(left.task.Operation()),
changelogOperationOrder(right.task.Operation())); operation != 0 {
- return operation
- }
- return cmp.Compare(left.file.File.FilePath(),
right.file.File.FilePath())
+ plannedTasks, err := s.planChangelogTasks(ctx, changelogTaskPlan{
+ scan: &planningScan,
+ fs: fs,
+ snapshots: snapshots,
+ timeline: snapshotRange,
+ manifestsBySnapshot: manifestsBySnapshot,
+ deleteManifests: deleteManifestsBySnapshot,
+ dataEntries: dataEntries,
+ snapshotOrdinals: snapshotOrdinals,
+ schema: schema,
+ partitionFilters: partitionFilters,
+ residual: residual,
+ metrics: &acc,
})
+ if err != nil {
+ return nil, err
+ }
return finish(plannedTasks), nil
}
@@ -310,38 +328,27 @@ func changelogOperation(status
iceberg.ManifestEntryStatus) (ChangelogOperation,
}
}
-func newChangelogScanTask(entry iceberg.ManifestEntry, ordinal int, residual
iceberg.BooleanExpression) (ChangelogScanTask, error) {
+func newChangelogScanTask(entry iceberg.ManifestEntry, ordinal int, residual
iceberg.BooleanExpression, deletes []iceberg.DataFile) (ChangelogScanTask,
error) {
operation, err := changelogOperation(entry.Status())
if err != nil {
return nil, err
}
- file := entry.DataFile()
- configureFileScanTask := func(task *FileScanTask) {
- task.Start = 0
- task.Length = file.FileSizeBytes()
- task.Residual = residual
- task.FirstRowID = file.FirstRowID()
- if sequenceNumber := entry.DataSequenceNumber(); sequenceNumber
>= 0 {
- task.DataSequenceNumber = &sequenceNumber
- }
- }
-
switch operation {
case ChangelogOpInsert:
- task, err := NewAddedRowsScanTask(file, nil, ordinal,
entry.SnapshotID())
+ task, err := NewAddedRowsScanTask(entry.DataFile(), deletes,
ordinal, entry.SnapshotID())
if err != nil {
return nil, err
}
- configureFileScanTask(&task.FileScanTask)
+ configureChangelogFileTask(&task.FileScanTask, entry, residual)
return task, nil
case ChangelogOpDelete:
- task, err := NewDeletedDataFileScanTask(file, nil, ordinal,
entry.SnapshotID())
+ task, err := NewDeletedDataFileScanTask(entry.DataFile(),
deletes, ordinal, entry.SnapshotID())
if err != nil {
return nil, err
}
- configureFileScanTask(&task.FileScanTask)
+ configureChangelogFileTask(&task.FileScanTask, entry, residual)
return task, nil
default:
@@ -349,6 +356,17 @@ func newChangelogScanTask(entry iceberg.ManifestEntry,
ordinal int, residual ice
}
}
+func configureChangelogFileTask(task *FileScanTask, entry
iceberg.ManifestEntry, residual iceberg.BooleanExpression) {
+ file := entry.DataFile()
+ task.Start = 0
+ task.Length = file.FileSizeBytes()
+ task.Residual = residual
+ task.FirstRowID = file.FirstRowID()
+ if sequenceNumber := entry.DataSequenceNumber(); sequenceNumber >= 0 {
+ task.DataSequenceNumber = &sequenceNumber
+ }
+}
+
func (s *IncrementalChangelogScan) toSnapshot() (*Snapshot, error) {
if s.toSnapshotID != nil {
snapshot := s.scan.metadata.SnapshotByID(*s.toSnapshotID)
@@ -363,8 +381,10 @@ func (s *IncrementalChangelogScan) toSnapshot()
(*Snapshot, error) {
}
// changelogSnapshots retains every snapshot with a non-empty operation except
-// replace snapshots. Keeping unknown future operations matches Java's
-// incremental changelog scan behavior; only a missing operation is rejected.
+// replace snapshots, which are not emitted as changelog tasks. Delete-file
+// changes from those replace snapshots are still tracked. Keeping unknown
+// future operations matches Java's incremental changelog scan behavior; only
+// a missing operation is rejected.
func changelogSnapshots(snapshots []Snapshot) ([]Snapshot, error) {
result := make([]Snapshot, 0, len(snapshots))
for _, snapshot := range snapshots {
diff --git a/table/incremental_changelog_scan_test.go
b/table/incremental_changelog_scan_test.go
index cdf727a47..cabf3c0fc 100644
--- a/table/incremental_changelog_scan_test.go
+++ b/table/incremental_changelog_scan_test.go
@@ -20,9 +20,19 @@ package table
import (
"bytes"
"context"
+ "fmt"
+ "os"
+ "path/filepath"
+ "strconv"
+ "strings"
"sync/atomic"
"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/arrow-go/v18/parquet"
+ "github.com/apache/arrow-go/v18/parquet/pqarrow"
"github.com/apache/iceberg-go"
iceio "github.com/apache/iceberg-go/io"
"github.com/apache/iceberg-go/metrics"
@@ -256,26 +266,66 @@ func TestIncrementalChangelogScanEmitsScanReport(t
*testing.T) {
require.Equal(t, int64(3), report.Metrics.ScannedDataManifests.Value)
}
-func TestIncrementalChangelogScanRejectsDeleteManifests(t *testing.T) {
+func TestIncrementalChangelogScanPlansPositionDeletes(t *testing.T) {
+ reporter := &metrics.InMemoryReporter{}
tbl := incrementalChangelogDeleteManifestTable(t)
- tasks, err :=
tbl.NewIncrementalChangelogScan().PlanFiles(context.Background())
- require.ErrorIs(t, err, ErrInvalidOperation)
- require.ErrorContains(t, err, "scan range references a delete manifest")
- require.Nil(t, tasks)
+ tasks, err :=
tbl.NewIncrementalChangelogScan(WithReporter(reporter)).PlanFiles(context.Background())
+ require.NoError(t, err)
+ require.Len(t, tasks, 3)
+
+ require.Equal(t, ChangelogOpInsert, tasks[0].Operation())
+ require.Equal(t, "mem://default/changelog-delete/data-1.parquet",
tasks[0].ScanTask().File.FilePath())
+ require.Equal(t, 0, tasks[0].ChangeOrdinal())
+ require.Empty(t, tasks[0].ScanTask().DeleteFiles)
+
+ rows, ok := tasks[1].(DeletedRowsScanTask)
+ require.True(t, ok)
+ require.Equal(t, ChangelogOpDelete, rows.Operation())
+ require.Equal(t, 1, rows.ChangeOrdinal())
+ require.Equal(t, int64(2), rows.CommitSnapshotID())
+ require.Equal(t, "mem://default/changelog-delete/data-1.parquet",
rows.ScanTask().File.FilePath())
+ require.Equal(t,
[]string{"mem://default/changelog-delete/delete.parquet"},
changelogFilePaths(rows.AddedDeletes()))
+ require.Empty(t, rows.ExistingDeletes())
+ require.NotNil(t, rows.ScanTask().DataSequenceNumber)
+ require.Equal(t, int64(1), *rows.ScanTask().DataSequenceNumber)
+
+ require.Equal(t, ChangelogOpInsert, tasks[2].Operation())
+ require.Equal(t, "mem://default/changelog-delete/data-3.parquet",
tasks[2].ScanTask().File.FilePath())
+ require.Equal(t, 2, tasks[2].ChangeOrdinal())
+
+ onlyDelete, err := tbl.NewIncrementalChangelogScan().
+ FromSnapshotInclusive(2).
+ ToSnapshot(2).
+ PlanFiles(context.Background())
+ require.NoError(t, err)
+ require.Len(t, onlyDelete, 1)
+ _, ok = onlyDelete[0].(DeletedRowsScanTask)
+ require.True(t, ok)
+
+ reports := reporter.Reports()
+ require.Len(t, reports, 1)
+ report, ok := reports[0].(metrics.ScanReport)
+ require.True(t, ok)
+ require.Equal(t, int64(3), report.Metrics.ResultDataFiles.Value)
+ require.Equal(t, int64(1), report.Metrics.ResultDeleteFiles.Value)
+ require.Equal(t, int64(1), report.Metrics.PositionalDeleteFiles.Value)
+ require.Equal(t, int64(1), report.Metrics.TotalDeleteManifests.Value)
+ require.Equal(t, int64(1), report.Metrics.ScannedDeleteManifests.Value)
}
-func TestIncrementalChangelogScanRejectsCarriedForwardDeleteManifests(t
*testing.T) {
+func TestIncrementalChangelogScanIgnoresCarriedForwardDeleteManifests(t
*testing.T) {
tbl := incrementalChangelogDeleteManifestTable(t)
tasks, err := tbl.NewIncrementalChangelogScan().
FromSnapshotExclusive(2).
ToSnapshot(3).
PlanFiles(context.Background())
- require.ErrorIs(t, err, ErrInvalidOperation)
- require.ErrorContains(t, err, "scan range references a delete manifest")
- require.ErrorContains(t, err, "snapshot 2")
- require.Nil(t, tasks)
+ require.NoError(t, err)
+ require.Len(t, tasks, 1)
+ require.Equal(t, ChangelogOpInsert, tasks[0].Operation())
+ require.Equal(t, "mem://default/changelog-delete/data-3.parquet",
tasks[0].ScanTask().File.FilePath())
+ require.Empty(t, tasks[0].ScanTask().DeleteFiles)
}
func TestIncrementalChangelogScanRejectsMissingSnapshotOperation(t *testing.T)
{
@@ -768,6 +818,1124 @@ 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},
+ {
+ operation: OpAppend,
+ data:
[]iceberg.ManifestEntry{changelogManifestEntry(iceberg.EntryStatusADDED, 2, 2,
data)},
+ deletes: []iceberg.ManifestEntry{
+
changelogManifestEntry(iceberg.EntryStatusADDED, 2, 2, posDelete),
+
changelogManifestEntry(iceberg.EntryStatusADDED, 2, 2, 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()
+
+ // A writer commits the data file in snapshot 2 at sequence 2, then
delete
+ // files in snapshots 3 and 4 at those snapshots' sequences. A later
+ // snapshot cannot commit a delete sequence below that data sequence.
+ t.Run("later snapshots", func(t *testing.T) {
+ for _, kind := range []string{"pos", "eq", "dv"} {
+ t.Run(kind, func(t *testing.T) {
+ data := newTestDataFile(t, spec,
"mem://default/changelog-seq/data.parquet", map[int]any{1000: int32(1)})
+ first, version := changelogSequenceDelete(t,
spec, kind, "mem://default/changelog-seq/first", data)
+ second, _ := changelogSequenceDelete(t, spec,
kind, "mem://default/changelog-seq/second", data)
+ tbl := writeChangelogScanTable(t,
"changelog-seq-"+kind, []changelogScanStep{
+ {operation: OpAppend},
+ {
+ operation: OpAppend,
+ data:
[]iceberg.ManifestEntry{changelogManifestEntry(iceberg.EntryStatusADDED, 2, 2,
data)},
+ },
+ {
+ operation: OpDelete,
+ manifestVersion: version,
+ deletes:
[]iceberg.ManifestEntry{changelogManifestEntry(iceberg.EntryStatusADDED, 3, 3,
first)},
+ },
+ {
+ operation: OpDelete,
+ manifestVersion: version,
+ deletes:
[]iceberg.ManifestEntry{changelogManifestEntry(iceberg.EntryStatusADDED, 4, 4,
second)},
+ },
+ })
+
+ tasks, err :=
tbl.NewIncrementalChangelogScan().PlanFiles(context.Background())
+ require.NoError(t, err)
+ bySnapshot := changelogTasksBySnapshot(tasks)
+ require.Equal(t, ChangelogOpInsert,
bySnapshot[2].Operation())
+
+ firstRows, ok :=
bySnapshot[3].(DeletedRowsScanTask)
+ require.True(t, ok)
+ require.Equal(t, []string{first.FilePath()},
changelogFilePaths(firstRows.AddedDeletes()))
+ require.Empty(t, firstRows.ExistingDeletes())
+
+ secondRows, ok :=
bySnapshot[4].(DeletedRowsScanTask)
+ require.True(t, ok)
+ require.Equal(t, []string{second.FilePath()},
changelogFilePaths(secondRows.AddedDeletes()))
+ require.Equal(t, []string{first.FilePath()},
changelogFilePaths(secondRows.ExistingDeletes()))
+ })
+ }
+ })
+
+ // Equality deletes do not apply at the data file's own sequence.
Position
+ // deletes and deletion vectors do.
+ t.Run("same snapshot", func(t *testing.T) {
+ for _, tt := range []struct {
+ kind string
+ want bool
+ }{
+ {kind: "pos", want: true},
+ {kind: "eq", want: false},
+ {kind: "dv", want: true},
+ } {
+ t.Run(tt.kind, func(t *testing.T) {
+ data := newTestDataFile(t, spec,
"mem://default/changelog-seq-same/data.parquet", map[int]any{1000: int32(1)})
+ deleteFile, version :=
changelogSequenceDelete(t, spec, tt.kind,
"mem://default/changelog-seq-same/delete", data)
+ tbl := writeChangelogScanTable(t,
"changelog-seq-same-"+tt.kind, []changelogScanStep{
+ {operation: OpAppend},
+ {
+ operation: OpAppend,
+ manifestVersion: version,
+ data:
[]iceberg.ManifestEntry{changelogManifestEntry(iceberg.EntryStatusADDED, 2, 2,
data)},
+ deletes:
[]iceberg.ManifestEntry{changelogManifestEntry(iceberg.EntryStatusADDED, 2, 2,
deleteFile)},
+ },
+ })
+
+ 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)
+ if !tt.want {
+ require.Empty(t, added.Deletes())
+
+ return
+ }
+ require.Equal(t,
[]string{deleteFile.FilePath()}, changelogFilePaths(added.Deletes()))
+ })
+ }
+ })
+
+ // A delete committed before the data file keeps its lower sequence and
+ // must not come back as an existing delete when the file is removed.
+ t.Run("older delete", func(t *testing.T) {
+ data := newTestDataFile(t, spec,
"mem://default/changelog-seq-old/data.parquet", map[int]any{1000: int32(1)})
+ older := newTestPosDeleteFileForSpec(t, spec,
+ "mem://default/changelog-seq-old/older.parquet",
map[int]any{1000: int32(1)}, data.FilePath())
+ newer := newTestPosDeleteFileForSpec(t, spec,
+ "mem://default/changelog-seq-old/newer.parquet",
map[int]any{1000: int32(1)}, data.FilePath())
+ tbl := writeChangelogScanTable(t, "changelog-seq-old",
[]changelogScanStep{
+ {
+ operation: OpDelete,
+ deletes:
[]iceberg.ManifestEntry{changelogManifestEntry(iceberg.EntryStatusADDED, 1, 1,
older)},
+ },
+ {
+ operation: OpAppend,
+ data:
[]iceberg.ManifestEntry{changelogManifestEntry(iceberg.EntryStatusADDED, 2, 2,
data)},
+ },
+ {
+ operation: OpDelete,
+ deletes:
[]iceberg.ManifestEntry{changelogManifestEntry(iceberg.EntryStatusADDED, 3, 3,
newer)},
+ },
+ {
+ operation: OpOverwrite,
+ data:
[]iceberg.ManifestEntry{changelogManifestEntry(iceberg.EntryStatusDELETED, 4,
2, data)},
+ },
+ })
+
+ tasks, err :=
tbl.NewIncrementalChangelogScan().PlanFiles(context.Background())
+ require.NoError(t, err)
+ bySnapshot := changelogTasksBySnapshot(tasks)
+ rows, ok := bySnapshot[3].(DeletedRowsScanTask)
+ require.True(t, ok)
+ require.Equal(t, []string{newer.FilePath()},
changelogFilePaths(rows.AddedDeletes()))
+ require.Empty(t, rows.ExistingDeletes())
+
+ removed, ok := bySnapshot[4].(DeletedDataFileScanTask)
+ require.True(t, ok)
+ require.Equal(t, []string{newer.FilePath()},
changelogFilePaths(removed.ExistingDeletes()))
+ })
+}
+
+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)})
+ listPath :=
"mem://default/changelog-expired-delete/metadata/snap-2.avro"
+ writeChangelogManifestList(t, fs, listPath, snapshotID, 2,
[]iceberg.ManifestFile{dataManifest, deleteManifest})
+
+ 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-delete"}, 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)
+ removed, ok := tasks[0].(DeletedDataFileScanTask)
+ require.True(t, ok)
+ require.Equal(t, []string{pos.FilePath()},
changelogFilePaths(removed.ExistingDeletes()))
+}
+
+func TestIncrementalChangelogScanKeepsRemovedDeleteWhenParentExpired(t
*testing.T) {
+ for _, inclusive := range []bool{false, true} {
+ name := "exclusive"
+ if inclusive {
+ name = "inclusive"
+ }
+ t.Run(name, func(t *testing.T) {
+ tbl := expiredParentRemovedDeleteTable(t,
"changelog-expired-removed-"+name)
+ scan := tbl.NewIncrementalChangelogScan()
+ if inclusive {
+ scan = scan.FromSnapshotInclusive(2)
+ } else {
+ scan = scan.FromSnapshotExclusive(1)
+ }
+ tasks, err :=
scan.ToSnapshot(2).PlanFiles(context.Background())
+ require.NoError(t, err)
+ require.Len(t, tasks, 1)
+ removed, ok := tasks[0].(DeletedDataFileScanTask)
+ require.True(t, ok)
+ require.Equal(t,
[]string{"mem://default/changelog-expired-removed-" + name + "/pos.parquet"},
changelogFilePaths(removed.ExistingDeletes()))
+ })
+ }
+}
+
+func TestIncrementalChangelogScanReplaceRewritesDVWhenParentExpired(t
*testing.T) {
+ spec := partitionedSpec()
+ txn, fs := createTestTransactionWithMemIO(t, spec)
+ schema := simpleSchema()
+ data := newTestDataFile(t, spec,
"mem://default/changelog-expired-replace/data.parquet", map[int]any{1000:
int32(1)})
+ oldDV := newChangelogDeletionVector(t, spec,
"mem://default/changelog-expired-replace/old.puffin", data.FilePath())
+ newDV := newChangelogDeletionVector(t, spec,
"mem://default/changelog-expired-replace/new.puffin", data.FilePath())
+ expiredParent := int64(1)
+ replaceID := int64(2)
+ overwriteID := int64(3)
+ deleteManifest := writeChangelogDeleteManifest(t, fs, spec, schema,
+
"mem://default/changelog-expired-replace/metadata/deletes.avro", replaceID, 3,
+ []iceberg.ManifestEntry{
+ changelogManifestEntry(iceberg.EntryStatusDELETED,
replaceID, 2, oldDV),
+ changelogManifestEntry(iceberg.EntryStatusADDED,
replaceID, 3, newDV),
+ })
+ dataManifest := writeChangelogDataManifest(t, fs, spec, schema,
+ "mem://default/changelog-expired-replace/metadata/data.avro",
overwriteID, 3,
+
[]iceberg.ManifestEntry{changelogManifestEntry(iceberg.EntryStatusDELETED,
overwriteID, 2, data)})
+ replaceList :=
"mem://default/changelog-expired-replace/metadata/snap-2.avro"
+ overwriteList :=
"mem://default/changelog-expired-replace/metadata/snap-3.avro"
+ replaceManifests := writeChangelogManifestList(t, fs, replaceList,
replaceID, 3, []iceberg.ManifestFile{deleteManifest})
+ writeChangelogManifestList(t, fs, overwriteList, overwriteID, 3,
append(replaceManifests, dataManifest))
+
+ schemaID := schema.ID
+ txn.meta.snapshotList = []Snapshot{
+ {
+ SnapshotID: replaceID,
+ ParentSnapshotID: &expiredParent,
+ TimestampMs: 2000,
+ ManifestList: replaceList,
+ SequenceNumber: replaceID,
+ SchemaID: &schemaID,
+ Summary: &Summary{
+ Operation: OpReplace,
+ Properties:
iceberg.Properties{removedDeleteFilesKey: "1"},
+ },
+ },
+ {
+ SnapshotID: overwriteID,
+ ParentSnapshotID: &replaceID,
+ TimestampMs: 3000,
+ ManifestList: overwriteList,
+ SequenceNumber: overwriteID,
+ SchemaID: &schemaID,
+ Summary: &Summary{Operation: OpOverwrite},
+ },
+ }
+ txn.meta.snapshotLog = []SnapshotLogEntry{
+ {SnapshotID: replaceID, TimestampMs: 2000},
+ {SnapshotID: overwriteID, TimestampMs: 3000},
+ }
+ txn.meta.currentSnapshotID = &overwriteID
+ meta, err := txn.meta.Build()
+ require.NoError(t, err)
+ tbl := New(Identifier{"changelog-expired-replace"}, meta,
"metadata.json", func(context.Context) (iceio.IO, error) {
+ return fs, nil
+ }, nil)
+
+ tasks, err := tbl.NewIncrementalChangelogScan().
+ FromSnapshotExclusive(expiredParent).
+ PlanFiles(context.Background())
+ require.NoError(t, err)
+ require.Len(t, tasks, 1)
+ removed, ok := tasks[0].(DeletedDataFileScanTask)
+ require.True(t, ok)
+ require.Equal(t, overwriteID, removed.CommitSnapshotID())
+ require.Equal(t, []string{newDV.FilePath()},
changelogFilePaths(removed.ExistingDeletes()))
+}
+
+func TestIncrementalChangelogScanRejectsUnrecordedRemovedDeletes(t *testing.T)
{
+ spec := partitionedSpec()
+ txn, fs := createTestTransactionWithMemIO(t, spec)
+ schema := simpleSchema()
+ data := newTestDataFile(t, spec,
"mem://default/changelog-expired-guard/data.parquet", map[int]any{1000:
int32(1)})
+ snapshotID := int64(2)
+ dataManifest := writeChangelogDataManifest(t, fs, spec, schema,
+ "mem://default/changelog-expired-guard/metadata/data.avro",
snapshotID, 2,
+
[]iceberg.ManifestEntry{changelogManifestEntry(iceberg.EntryStatusDELETED,
snapshotID, 1, data)})
+ listPath := "mem://default/changelog-expired-guard/metadata/snap-2.avro"
+ writeChangelogManifestList(t, fs, listPath, snapshotID, 2,
[]iceberg.ManifestFile{dataManifest})
+
+ expiredParent := int64(1)
+ schemaID := schema.ID
+ txn.meta.snapshotList = []Snapshot{{
+ SnapshotID: snapshotID,
+ ParentSnapshotID: &expiredParent,
+ TimestampMs: 2000,
+ ManifestList: listPath,
+ SequenceNumber: snapshotID,
+ SchemaID: &schemaID,
+ Summary: &Summary{
+ Operation: OpOverwrite,
+ Properties: iceberg.Properties{removedDeleteFilesKey:
"1"},
+ },
+ }}
+ 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-guard"}, meta,
"metadata.json", func(context.Context) (iceio.IO, error) {
+ return fs, nil
+ }, nil)
+
+ _, err = tbl.NewIncrementalChangelogScan().
+ FromSnapshotExclusive(expiredParent).
+ PlanFiles(context.Background())
+ require.ErrorIs(t, err, iceberg.ErrInvalidArgument)
+ require.ErrorContains(t, err, removedDeleteFilesKey)
+}
+
+func TestIncrementalChangelogScanCountsRemovedDeletesBeforePartitionPrune(t
*testing.T) {
+ spec := partitionedSpec()
+ txn, fs := createTestTransactionWithMemIO(t, spec)
+ schema := simpleSchema()
+ location := "mem://default/changelog-expired-prune"
+ data := newTestDataFile(t, spec, location+"/data.parquet",
map[int]any{1000: int32(1)})
+ kept := newTestPosDeleteFileForSpec(t, spec, location+"/kept.parquet",
map[int]any{1000: int32(1)}, data.FilePath())
+ pruned := newTestPosDeleteFileForSpec(t, spec,
location+"/pruned.parquet", map[int]any{1000: int32(2)}, data.FilePath())
+ snapshotID := int64(2)
+ expiredParent := int64(1)
+ dataManifest := writeChangelogDataManifest(t, fs, spec, schema,
location+"/metadata/data.avro", snapshotID, 2,
+
[]iceberg.ManifestEntry{changelogManifestEntry(iceberg.EntryStatusDELETED,
snapshotID, 1, data)})
+ keptManifest := writeChangelogDeleteManifest(t, fs, spec, schema,
location+"/metadata/kept.avro", snapshotID, 2,
+
[]iceberg.ManifestEntry{changelogManifestEntry(iceberg.EntryStatusDELETED,
snapshotID, 1, kept)})
+ prunedManifest := writeChangelogDeleteManifest(t, fs, spec, schema,
location+"/metadata/pruned.avro", snapshotID, 2,
+
[]iceberg.ManifestEntry{changelogManifestEntry(iceberg.EntryStatusDELETED,
snapshotID, 1, pruned)})
+ listPath := location + "/metadata/snap-2.avro"
+ writeChangelogManifestList(t, fs, listPath, snapshotID, 2,
[]iceberg.ManifestFile{dataManifest, keptManifest, prunedManifest})
+
+ schemaID := schema.ID
+ txn.meta.snapshotList = []Snapshot{{
+ SnapshotID: snapshotID,
+ ParentSnapshotID: &expiredParent,
+ TimestampMs: 2000,
+ ManifestList: listPath,
+ SequenceNumber: snapshotID,
+ SchemaID: &schemaID,
+ Summary: &Summary{
+ Operation: OpOverwrite,
+ Properties: iceberg.Properties{removedDeleteFilesKey:
"2"},
+ },
+ }}
+ 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-prune"}, meta,
"metadata.json", func(context.Context) (iceio.IO, error) {
+ return fs, nil
+ }, nil)
+
+ tasks, err := tbl.NewIncrementalChangelogScan(
+ WithRowFilter(iceberg.EqualTo(iceberg.Reference("id"),
int32(1))),
+ ).FromSnapshotExclusive(expiredParent).PlanFiles(context.Background())
+ require.NoError(t, err)
+ require.Len(t, tasks, 1)
+ removed, ok := tasks[0].(DeletedDataFileScanTask)
+ require.True(t, ok)
+ require.Equal(t, []string{kept.FilePath()},
changelogFilePaths(removed.ExistingDeletes()))
+}
+
+func TestEntriesLiveBefore(t *testing.T) {
+ carried := changelogTestDataFile(t, "carried.parquet",
iceberg.EntryContentPosDeletes, iceberg.ParquetFile)
+ removed := changelogTestDataFile(t, "removed.parquet",
iceberg.EntryContentPosDeletes, iceberg.ParquetFile)
+ added := changelogTestDataFile(t, "added.parquet",
iceberg.EntryContentPosDeletes, iceberg.ParquetFile)
+ dup := changelogTestDataFile(t, "dup.parquet",
iceberg.EntryContentPosDeletes, iceberg.ParquetFile)
+ snapshotID := int64(5)
+
+ got := entriesLiveBefore([]iceberg.ManifestEntry{
+ changelogManifestEntry(iceberg.EntryStatusEXISTING, 2, 1,
carried),
+ changelogManifestEntry(iceberg.EntryStatusDELETED, snapshotID,
1, removed),
+ changelogManifestEntry(iceberg.EntryStatusADDED, snapshotID, 2,
added),
+ }, snapshotID)
+ require.Equal(t, []string{carried.FilePath(), removed.FilePath()},
changelogEntryPaths(got))
+
+ for _, entries := range [][]iceberg.ManifestEntry{
+ {
+ changelogManifestEntry(iceberg.EntryStatusDELETED,
snapshotID, 1, dup),
+ changelogManifestEntry(iceberg.EntryStatusEXISTING, 2,
1, dup),
+ },
+ {
+ changelogManifestEntry(iceberg.EntryStatusEXISTING, 2,
1, dup),
+ changelogManifestEntry(iceberg.EntryStatusDELETED,
snapshotID, 1, dup),
+ },
+ } {
+ got = entriesLiveBefore(entries, snapshotID)
+ require.Equal(t, []string{dup.FilePath()},
changelogEntryPaths(got))
+ require.Equal(t, iceberg.EntryStatusEXISTING, got[0].Status())
+ }
+}
+
+func TestLiveManifestEntriesUsesTheLatestStatus(t *testing.T) {
+ file := changelogTestDataFile(t, "data.parquet",
iceberg.EntryContentData, iceberg.ParquetFile)
+ // Snapshot IDs are not chronological. The delete is snapshot 3, whose
+ // sequence number is 2. The add is snapshot 20, whose sequence number
is 1.
+ added := changelogManifestEntry(iceberg.EntryStatusADDED, 20, 1, file)
+ deleted := changelogManifestEntry(iceberg.EntryStatusDELETED, 3, 1,
file)
+ readded := changelogManifestEntry(iceberg.EntryStatusADDED, 2, 1, file)
+ sameSnapshotDeleted :=
changelogManifestEntry(iceberg.EntryStatusDELETED, 20, 1, file)
+ sequences := func(snapshotID int64) (int64, bool) {
+ switch snapshotID {
+ case 20:
+ return 1, true
+ case 3:
+ return 2, true
+ case 2:
+ return 3, true
+ default:
+ return 0, false
+ }
+ }
+
+ for _, entries := range [][]iceberg.ManifestEntry{
+ {added, deleted},
+ {deleted, added},
+ {added, sameSnapshotDeleted},
+ {sameSnapshotDeleted, added},
+ } {
+ require.Empty(t, liveManifestEntries(entries, sequences))
+ }
+
+ for _, entries := range [][]iceberg.ManifestEntry{
+ {added, deleted, readded},
+ {readded, deleted, added},
+ } {
+ got := liveManifestEntries(entries, sequences)
+ require.Len(t, got, 1)
+ require.Equal(t, iceberg.EntryStatusADDED, got[0].Status())
+ require.Equal(t, int64(2), got[0].SnapshotID())
+ }
+}
+
+func TestEntriesBeforeDropsRemovedDeleteFiles(t *testing.T) {
+ keep := changelogTestDataFile(t, "keep.parquet",
iceberg.EntryContentPosDeletes, iceberg.ParquetFile)
+ drop := changelogTestDataFile(t, "drop.parquet",
iceberg.EntryContentPosDeletes, iceberg.ParquetFile)
+ state := &changelogDeleteState{
+ snapshots: []Snapshot{{SnapshotID: 1}, {SnapshotID: 2},
{SnapshotID: 3}},
+ preRange:
[]iceberg.ManifestEntry{changelogManifestEntry(iceberg.EntryStatusADDED, 1, 1,
drop)},
+ added: map[int64][]iceberg.ManifestEntry{
+ 1: {changelogManifestEntry(iceberg.EntryStatusADDED, 1,
1, keep)},
+ },
+ removed: map[int64]map[string]struct{}{
+ 2: {drop.FilePath(): {}},
+ },
+ }
+
+ require.Equal(t, []string{keep.FilePath()},
changelogEntryPaths(state.entriesBefore(3)))
+
+ readded := &changelogDeleteState{
+ snapshots: state.snapshots,
+ preRange: state.preRange,
+ added: map[int64][]iceberg.ManifestEntry{
+ 1: {changelogManifestEntry(iceberg.EntryStatusADDED, 1,
1, keep)},
+ 2: {changelogManifestEntry(iceberg.EntryStatusADDED, 2,
2, drop)},
+ },
+ removed: state.removed,
+ }
+ require.Equal(t, []string{keep.FilePath(), drop.FilePath()},
changelogEntryPaths(readded.entriesBefore(3)))
+}
+
+func TestKeepNewestDeletionVector(t *testing.T) {
+ spec := partitionedSpec()
+ older := newChangelogDeletionVector(t, spec, "older.puffin",
"data.parquet")
+ newer := newChangelogDeletionVector(t, spec, "newer.puffin",
"data.parquet")
+ pos := changelogTestDataFile(t, "pos.parquet",
iceberg.EntryContentPosDeletes, iceberg.ParquetFile)
+ got := keepNewestDeletionVector([]iceberg.ManifestEntry{
+ changelogManifestEntry(iceberg.EntryStatusADDED, 1, 1, older),
+ changelogManifestEntry(iceberg.EntryStatusADDED, 2, 2, pos),
+ changelogManifestEntry(iceberg.EntryStatusADDED, 3, 3, newer),
+ })
+ require.Equal(t, []string{pos.FilePath(), newer.FilePath()},
changelogEntryPaths(got))
+}
+
+func changelogFilePaths(files []iceberg.DataFile) []string {
+ out := make([]string, len(files))
+ for i, file := range files {
+ out[i] = file.FilePath()
+ }
+
+ return out
+}
+
+func changelogEntryPaths(entries []iceberg.ManifestEntry) []string {
+ out := make([]string, len(entries))
+ for i, entry := range entries {
+ out[i] = entry.DataFile().FilePath()
+ }
+
+ return out
+}
+
+func changelogManifestEntry(status iceberg.ManifestEntryStatus, snapshotID,
sequence int64, file iceberg.DataFile) iceberg.ManifestEntry {
+ return iceberg.NewManifestEntry(status, &snapshotID, &sequence,
&sequence, file)
+}
+
+func changelogTasksBySnapshot(tasks []ChangelogScanTask)
map[int64]ChangelogScanTask {
+ out := make(map[int64]ChangelogScanTask, len(tasks))
+ for _, task := range tasks {
+ out[task.CommitSnapshotID()] = task
+ }
+
+ return out
+}
+
+func changelogSequenceDelete(t *testing.T, spec iceberg.PartitionSpec, kind,
path string, data iceberg.DataFile) (iceberg.DataFile, int) {
+ t.Helper()
+
+ switch kind {
+ case "pos":
+ return newTestPosDeleteFileForSpec(t, spec, path+".parquet",
map[int]any{1000: int32(1)}, data.FilePath()), 2
+ case "eq":
+ return newChangelogEqDelete(t, spec, path+".parquet",
map[int]any{1000: int32(1)}), 2
+ default:
+ return newChangelogDeletionVector(t, spec, path+".puffin",
data.FilePath()), 3
+ }
+}
+
+func expiredParentRemovedDeleteTable(t *testing.T, location string) *Table {
+ t.Helper()
+
+ spec := partitionedSpec()
+ txn, fs := createTestTransactionWithMemIO(t, spec)
+ schema := simpleSchema()
+ data := newTestDataFile(t, spec,
"mem://default/"+location+"/data.parquet", map[int]any{1000: int32(1)})
+ pos := newTestPosDeleteFileForSpec(t, spec,
+ "mem://default/"+location+"/pos.parquet", map[int]any{1000:
int32(1)}, data.FilePath())
+ snapshotID := int64(2)
+ expiredParent := int64(1)
+ dataManifest := writeChangelogDataManifest(t, fs, spec, schema,
+ "mem://default/"+location+"/metadata/data.avro", snapshotID, 2,
+
[]iceberg.ManifestEntry{changelogManifestEntry(iceberg.EntryStatusDELETED,
snapshotID, 1, data)})
+ deleteManifest := writeChangelogDeleteManifest(t, fs, spec, schema,
+ "mem://default/"+location+"/metadata/deletes.avro", snapshotID,
2,
+
[]iceberg.ManifestEntry{changelogManifestEntry(iceberg.EntryStatusDELETED,
snapshotID, 1, pos)})
+ listPath := "mem://default/" + location + "/metadata/snap-2.avro"
+ writeChangelogManifestList(t, fs, listPath, snapshotID, 2,
[]iceberg.ManifestFile{dataManifest, deleteManifest})
+
+ schemaID := schema.ID
+ txn.meta.snapshotList = []Snapshot{{
+ SnapshotID: snapshotID,
+ ParentSnapshotID: &expiredParent,
+ TimestampMs: 2000,
+ ManifestList: listPath,
+ SequenceNumber: snapshotID,
+ SchemaID: &schemaID,
+ Summary: &Summary{
+ Operation: OpOverwrite,
+ Properties: iceberg.Properties{removedDeleteFilesKey:
"1"},
+ },
+ }}
+ txn.meta.snapshotLog = []SnapshotLogEntry{{SnapshotID: snapshotID,
TimestampMs: 2000}}
+ txn.meta.currentSnapshotID = &snapshotID
+ meta, err := txn.meta.Build()
+ require.NoError(t, err)
+
+ return New(Identifier{location}, meta, "metadata.json",
func(context.Context) (iceio.IO, error) {
+ return fs, nil
+ }, nil)
+}
+
+func newChangelogEqDelete(t *testing.T, spec iceberg.PartitionSpec, path
string, partition map[int]any) iceberg.DataFile {
+ t.Helper()
+
+ builder, err := iceberg.NewDataFileBuilder(
+ spec, iceberg.EntryContentEqDeletes, path, iceberg.ParquetFile,
partition, nil, nil, 1, 1)
+ require.NoError(t, err)
+
+ return builder.EqualityFieldIDs([]int{1}).Build()
+}
+
+func newChangelogDeletionVector(t *testing.T, spec iceberg.PartitionSpec,
path, dataPath string) iceberg.DataFile {
+ t.Helper()
+
+ return newSharedDeletionVector(t, spec, path, dataPath, 0)
+}
+
+func newSharedDeletionVector(t *testing.T, spec iceberg.PartitionSpec, path,
dataPath string, offset int64) iceberg.DataFile {
+ t.Helper()
+
+ builder, err := iceberg.NewDataFileBuilder(
+ spec, iceberg.EntryContentPosDeletes, path, iceberg.PuffinFile,
nil, nil, nil, 1, 1)
+ require.NoError(t, err)
+
+ return
builder.ReferencedDataFile(dataPath).ContentOffset(offset).ContentSizeInBytes(16).Build()
+}
+
+type changelogScanStep struct {
+ operation Operation
+ manifestVersion int
+ data []iceberg.ManifestEntry
+ deletes []iceberg.ManifestEntry
+}
+
+func writeChangelogScanTable(t *testing.T, location string, steps
[]changelogScanStep) *Table {
+ t.Helper()
+
+ spec := partitionedSpec()
+ txn, fs := createTestTransactionWithMemIO(t, spec)
+ schema := simpleSchema()
+ var manifestList []iceberg.ManifestFile
+ listVersion := 2
+ snapshots := make([]Snapshot, 0, len(steps))
+ log := make([]SnapshotLogEntry, 0, len(steps))
+ var parent *int64
+ for i, step := range steps {
+ snapshotID := int64(i + 1)
+ version := step.manifestVersion
+ if version == 0 {
+ version = 2
+ }
+ if version > listVersion {
+ listVersion = version
+ }
+ var written []iceberg.ManifestFile
+ if len(step.data) > 0 {
+ path := "mem://default/" + location + "/metadata/data-"
+ fmtSnapshot(snapshotID) + ".avro"
+ written = append(written, writeChangelogDataManifest(t,
fs, spec, schema, path, snapshotID, version, step.data))
+ }
+ if len(step.deletes) > 0 {
+ path := "mem://default/" + location +
"/metadata/deletes-" + fmtSnapshot(snapshotID) + ".avro"
+ written = append(written,
writeChangelogDeleteManifest(t, fs, spec, schema, path, snapshotID, version,
step.deletes))
+ }
+ listPath := "mem://default/" + location + "/metadata/snap-" +
fmtSnapshot(snapshotID) + ".avro"
+ manifestList = writeChangelogManifestList(t, fs, listPath,
snapshotID, listVersion, append(manifestList, written...))
+ snapshots = append(snapshots, Snapshot{
+ SnapshotID: snapshotID,
+ ParentSnapshotID: parent,
+ TimestampMs: snapshotID * 1000,
+ ManifestList: listPath,
+ SequenceNumber: snapshotID,
+ SchemaID: &schema.ID,
+ Summary: &Summary{Operation: step.operation},
+ })
+ log = append(log, SnapshotLogEntry{SnapshotID: snapshotID,
TimestampMs: snapshotID * 1000})
+ id := snapshotID
+ parent = &id
+ }
+ txn.meta.snapshotList = snapshots
+ txn.meta.snapshotLog = log
+ if len(snapshots) > 0 {
+ current := snapshots[len(snapshots)-1].SnapshotID
+ txn.meta.currentSnapshotID = ¤t
+ }
+ meta, err := txn.meta.Build()
+ require.NoError(t, err)
+
+ return New(Identifier{location}, meta, "metadata.json",
func(context.Context) (iceio.IO, error) {
+ return fs, nil
+ }, nil)
+}
+
+func fmtSnapshot(snapshotID int64) string {
+ return strconv.FormatInt(snapshotID, 10)
+}
+
+func writeChangelogDataManifest(
+ t *testing.T,
+ fs iceio.WriteFileIO,
+ spec iceberg.PartitionSpec,
+ schema *iceberg.Schema,
+ path string,
+ snapshotID int64,
+ version int,
+ entries []iceberg.ManifestEntry,
+) iceberg.ManifestFile {
+ t.Helper()
+
+ var buf bytes.Buffer
+ manifest, err := iceberg.WriteManifest(path, &buf, version, spec,
schema, snapshotID, entries)
+ require.NoError(t, err)
+ require.NoError(t, fs.WriteFile(context.Background(), path,
buf.Bytes()))
+
+ return manifest
+}
+
+func writeChangelogDeleteManifest(
+ t *testing.T,
+ fs iceio.WriteFileIO,
+ spec iceberg.PartitionSpec,
+ schema *iceberg.Schema,
+ path string,
+ snapshotID int64,
+ version int,
+ entries []iceberg.ManifestEntry,
+) iceberg.ManifestFile {
+ t.Helper()
+
+ var buf bytes.Buffer
+ writer, err := iceberg.NewManifestWriter(version, &buf, spec, schema,
snapshotID,
+
iceberg.WithManifestWriterContent(iceberg.ManifestContentDeletes))
+ require.NoError(t, err)
+ for _, entry := range entries {
+ switch entry.Status() {
+ case iceberg.EntryStatusDELETED:
+ require.NoError(t, writer.Delete(entry))
+ case iceberg.EntryStatusEXISTING:
+ require.NoError(t, writer.Existing(entry))
+ default:
+ require.NoError(t, writer.Add(entry))
+ }
+ }
+ require.NoError(t, writer.Close())
+ manifest, err := writer.ToManifestFile(path, int64(buf.Len()),
+ iceberg.WithManifestFileContent(iceberg.ManifestContentDeletes))
+ require.NoError(t, err)
+ require.NoError(t, fs.WriteFile(context.Background(), path,
buf.Bytes()))
+
+ return manifest
+}
+
+func writeChangelogManifestList(
+ t *testing.T,
+ fs iceio.WriteFileIO,
+ path string,
+ snapshotID int64,
+ version int,
+ manifests []iceberg.ManifestFile,
+) []iceberg.ManifestFile {
+ t.Helper()
+
+ var buf bytes.Buffer
+ sequenceNumber := snapshotID
+ require.NoError(t, iceberg.WriteManifestList(version, &buf, snapshotID,
nil, &sequenceNumber, 0, manifests))
+ require.NoError(t, fs.WriteFile(context.Background(), path,
buf.Bytes()))
+ listFile, err := fs.Open(context.Background(), path)
+ require.NoError(t, err)
+ list, err := iceberg.ReadManifestList(listFile)
+ require.NoError(t, err)
+ require.NoError(t, listFile.Close())
+
+ return list
+}
+
type countingOpenIO struct {
iceio.IO
opens *atomic.Int64
@@ -782,3 +1950,145 @@ func (io *countingOpenIO) Open(_ context.Context, name
string) (iceio.File, erro
return io.IO.Open(context.Background(), name)
}
+
+// TestIncrementalChangelogScanRebuildsExpiredParentFromRealWriters appends a
+// data file, adds a position delete with RowDelta, then replaces both through
+// the overwrite writer. Expiring the delete snapshot leaves the replace as
+// the first in-range snapshot, and planning rebuilds the position delete from
+// the DELETED entry that writer recorded.
+func TestIncrementalChangelogScanRebuildsExpiredParentFromRealWriters(t
*testing.T) {
+ ctx := context.Background()
+ dir := t.TempDir()
+ schema := simpleSchema()
+ meta, err := NewMetadata(schema, iceberg.UnpartitionedSpec,
UnsortedSortOrder, dir, iceberg.Properties{
+ PropertyFormatVersion: "2",
+ })
+ require.NoError(t, err)
+ cat := &countingCatalog{metadata: meta}
+ fsF := func(context.Context) (iceio.IO, error) { return
iceio.LocalFS{}, nil }
+ tbl := New(Identifier{"db", "changelog-real-writer"}, meta, "", fsF,
cat)
+
+ arrowSchema, err := SchemaToArrowSchema(schema, nil, false, false)
+ require.NoError(t, err)
+ tbl, err = appendChangelogRows(ctx, tbl, arrowSchema, `[{"id": 1}]`)
+ require.NoError(t, err)
+ require.NotNil(t, tbl.CurrentSnapshot())
+
+ planned, err := tbl.Scan().PlanFiles(ctx)
+ require.NoError(t, err)
+ require.Len(t, planned, 1)
+ dataFile := planned[0].File
+
+ posPath := filepath.Join(dir, "pos-del.parquet")
+ writeChangelogParquet(t, posPath, PositionalDeleteArrowSchema,
fmt.Sprintf(`[{"file_path": %q, "pos": 0}]`, dataFile.FilePath()))
+ posInfo, err := os.Stat(posPath)
+ require.NoError(t, err)
+ posBuilder, err := iceberg.NewDataFileBuilder(
+ *iceberg.UnpartitionedSpec,
+ iceberg.EntryContentPosDeletes,
+ posPath,
+ iceberg.ParquetFile,
+ nil, nil, nil,
+ 1,
+ posInfo.Size(),
+ )
+ require.NoError(t, err)
+ posDelete := posBuilder.ReferencedDataFile(dataFile.FilePath()).Build()
+
+ tx := tbl.NewTransaction()
+ require.NoError(t,
tx.NewRowDelta(nil).AddDeletes(posDelete).Commit(ctx))
+ tbl, err = tx.Commit(ctx)
+ require.NoError(t, err)
+ snapshotB := tbl.CurrentSnapshot().SnapshotID
+
+ planned, err = tbl.Scan().PlanFiles(ctx)
+ require.NoError(t, err)
+ require.Len(t, planned, 1)
+ require.NotEmpty(t, planned[0].DeleteFiles)
+
+ replacementPath := filepath.Join(dir, "replacement.parquet")
+ writeChangelogParquet(t, replacementPath, arrowSchema, `[{"id": 2}]`)
+ replacementInfo, err := os.Stat(replacementPath)
+ require.NoError(t, err)
+ replacementBuilder, err := iceberg.NewDataFileBuilder(
+ *iceberg.UnpartitionedSpec,
+ iceberg.EntryContentData,
+ replacementPath,
+ iceberg.ParquetFile,
+ nil, nil, nil,
+ 1,
+ replacementInfo.Size(),
+ )
+ require.NoError(t, err)
+
+ tx = tbl.NewTransaction()
+ require.NoError(t, tx.ReplaceFiles(ctx,
+ []iceberg.DataFile{planned[0].File},
+ []iceberg.DataFile{replacementBuilder.Build()},
+ append([]iceberg.DataFile{}, planned[0].DeleteFiles...),
+ nil,
+ ))
+ tbl, err = tx.Commit(ctx)
+ require.NoError(t, err)
+ snapshotC := tbl.CurrentSnapshot()
+ require.NotNil(t, snapshotC)
+ require.Contains(t, snapshotC.Summary.Properties, removedDeleteFilesKey)
+
+ builder, err := MetadataBuilderFromBase(tbl.Metadata(),
tbl.MetadataLocation())
+ require.NoError(t, err)
+ require.NoError(t, builder.RemoveSnapshots([]int64{snapshotB}, false))
+ expired, err := builder.Build()
+ require.NoError(t, err)
+ tbl = New(tbl.Identifier(), expired, tbl.MetadataLocation(), fsF, cat)
+
+ tasks, err :=
tbl.NewIncrementalChangelogScan().FromSnapshotExclusive(snapshotB).PlanFiles(ctx)
+ require.NoError(t, err)
+
+ var deleted DeletedDataFileScanTask
+ found := false
+ for _, task := range tasks {
+ got, ok := task.(DeletedDataFileScanTask)
+ if !ok {
+ continue
+ }
+ found = true
+ deleted = got
+ }
+ require.True(t, found)
+ require.Equal(t, snapshotC.SnapshotID, deleted.CommitSnapshotID())
+ paths := make([]string, 0, len(deleted.ExistingDeletes()))
+ for _, file := range deleted.ExistingDeletes() {
+ paths = append(paths, file.FilePath())
+ }
+ require.Contains(t, paths, posPath)
+}
+
+func appendChangelogRows(ctx context.Context, tbl *Table, schema
*arrow.Schema, jsonRows string) (*Table, error) {
+ batch, _, err := array.RecordFromJSON(memory.DefaultAllocator, schema,
strings.NewReader(jsonRows))
+ if err != nil {
+ return nil, err
+ }
+ defer batch.Release()
+ rows := array.NewTableFromRecords(schema, []arrow.RecordBatch{batch})
+ defer rows.Release()
+ reader := array.NewTableReader(rows, rows.NumRows())
+ defer reader.Release()
+
+ return tbl.Append(ctx, reader, nil)
+}
+
+func writeChangelogParquet(t *testing.T, path string, schema *arrow.Schema,
jsonData string) {
+ t.Helper()
+
+ batch, _, err := array.RecordFromJSON(memory.DefaultAllocator, schema,
strings.NewReader(jsonData))
+ require.NoError(t, err)
+ defer batch.Release()
+
+ out, err := iceio.LocalFS{}.Create(context.Background(), path)
+ require.NoError(t, err)
+ defer out.Close()
+
+ rows := array.NewTableFromRecords(schema, []arrow.RecordBatch{batch})
+ defer rows.Release()
+ require.NoError(t, pqarrow.WriteTable(rows, out, batch.NumRows(),
parquet.NewWriterProperties(), pqarrow.DefaultWriterProps()))
+}
