zeroshade commented on code in PR #2169: URL: https://github.com/apache/iceberg-go/pull/2169#discussion_r4426236191
########## 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 { Review Comment: Non-blocking: this keeps a list of live deletes for every snapshot in the range, and each snapshot that adds or removes deletes gets a fresh copy plus a `keepNewestDeletionVector` pass. On a wide CDC range with many live deletes, that's (delete-changing snapshots × live deletes) memory even for snapshots whose index is never built. Keeping lists only for snapshots that need one would bound it. Could also go in #2370. ########## 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) Review Comment: Non-blocking: this reads the snapshot's delete manifests twice, once unfiltered for the guard and again through `readDeleteManifestEntries` for the pruned set. You could prune `raw` in memory with the partition filter instead of re-opening the manifests. It's only on the expired-parent path, so fine as a follow-up. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
