zeroshade commented on code in PR #947: URL: https://github.com/apache/iceberg-go/pull/947#discussion_r3163717688
########## table/compaction/eq_delete_collect.go: ########## @@ -0,0 +1,121 @@ +// 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 compaction + +import ( + "context" + + "github.com/apache/iceberg-go" + iceio "github.com/apache/iceberg-go/io" + "github.com/apache/iceberg-go/table" +) + +// CollectDeadEqualityDeletes walks the given snapshot's manifests and +// returns the equality delete files that are provably dead after a +// rewrite removes the data files in rewrittenPaths. +// +// "Dead" means: no surviving data file in the snapshot (excluding +// rewrittenPaths) could ever apply to the eq-delete per the v2 reader +// predicate (see DecideDeadEqualityDeletes for the exact rule). +// +// rewrittenPaths is the union of every old data file path being +// replaced by a planned rewrite (across all groups). The caller +// typically computes this from a compaction plan. +// +// The returned files are safe to remove during the same commit that +// stages the rewrite — they cannot be resurrected by any concurrent +// commit because (1) any concurrent eq-delete is rejected by the +// rewrite-validator, and (2) any concurrent data file gets a fresh +// sequence number strictly greater than every preexisting eq-delete, +// so it cannot satisfy E.seq > D.seq for any preexisting E. +// +// Two-pass design: the first pass walks delete manifests only to +// discover candidate eq-deletes; if there are none, the more +// expensive data-manifest walk is skipped. +func CollectDeadEqualityDeletes( + ctx context.Context, + fs iceio.IO, + snap *table.Snapshot, + rewrittenPaths map[string]struct{}, +) ([]iceberg.DataFile, error) { + if snap == nil || len(rewrittenPaths) == 0 { + return nil, nil + } + + manifests, err := snap.Manifests(fs) + if err != nil { + return nil, err + } + + // First pass: delete manifests only, collecting eq-delete candidates. + var candidates []iceberg.ManifestEntry + for _, m := range manifests { Review Comment: don't we have a helper for this to simplify the loops? ########## table/compaction/eq_delete_collect.go: ########## @@ -0,0 +1,121 @@ +// 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 compaction + +import ( + "context" + + "github.com/apache/iceberg-go" + iceio "github.com/apache/iceberg-go/io" + "github.com/apache/iceberg-go/table" +) + +// CollectDeadEqualityDeletes walks the given snapshot's manifests and +// returns the equality delete files that are provably dead after a +// rewrite removes the data files in rewrittenPaths. +// +// "Dead" means: no surviving data file in the snapshot (excluding +// rewrittenPaths) could ever apply to the eq-delete per the v2 reader +// predicate (see DecideDeadEqualityDeletes for the exact rule). +// +// rewrittenPaths is the union of every old data file path being +// replaced by a planned rewrite (across all groups). The caller +// typically computes this from a compaction plan. +// +// The returned files are safe to remove during the same commit that +// stages the rewrite — they cannot be resurrected by any concurrent +// commit because (1) any concurrent eq-delete is rejected by the +// rewrite-validator, and (2) any concurrent data file gets a fresh +// sequence number strictly greater than every preexisting eq-delete, +// so it cannot satisfy E.seq > D.seq for any preexisting E. +// +// Two-pass design: the first pass walks delete manifests only to +// discover candidate eq-deletes; if there are none, the more +// expensive data-manifest walk is skipped. +func CollectDeadEqualityDeletes( + ctx context.Context, + fs iceio.IO, + snap *table.Snapshot, + rewrittenPaths map[string]struct{}, +) ([]iceberg.DataFile, error) { + if snap == nil || len(rewrittenPaths) == 0 { + return nil, nil + } + + manifests, err := snap.Manifests(fs) + if err != nil { + return nil, err + } + + // First pass: delete manifests only, collecting eq-delete candidates. + var candidates []iceberg.ManifestEntry + for _, m := range manifests { + if cerr := ctx.Err(); cerr != nil { + return nil, cerr + } + if m.ManifestContent() != iceberg.ManifestContentDeletes { + continue + } + entries, err := m.FetchEntries(fs, false) + if err != nil { + return nil, err + } + for _, e := range entries { + if e.Status() == iceberg.EntryStatusDELETED { + continue + } + if e.DataFile().ContentType() != iceberg.EntryContentEqDeletes { + continue + } + candidates = append(candidates, e) + } + } + if len(candidates) == 0 { + return nil, nil + } + + // Second pass: data manifests, building the survivor survey. + survey := NewSurvivorSurvey() + for _, m := range manifests { Review Comment: same comment as above ########## table/compaction/eq_delete_decision.go: ########## @@ -0,0 +1,214 @@ +// 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 compaction + +import ( + "fmt" + "math" + "slices" + + "github.com/apache/iceberg-go" +) + +// SurvivorSurvey describes the surviving data files in a snapshot +// AFTER a planned rewrite has logically removed its rewrite set. It +// is the input to DecideDeadEqualityDeletes. +// +// EmptyPartMinSeq is the smallest sequence number among unpartitioned +// surviving data files. Per the Iceberg v2 reader predicate (see +// table/scanner.go matchEqualityDeletesToData), an unpartitioned data +// file applies to every equality delete — so it is always part of an +// eq-delete's "applicable survivors" minimum. +// +// PartMinSeq maps the partition tuple (encoded via partitionMatchKey) +// to the smallest sequence number among surviving partitioned data +// files in that tuple. The key intentionally does NOT include +// SpecID — the reader's predicate ignores SpecID, so the writer-side +// cleanup must too. +// +// Sentinel "no survivor in this bucket" is math.MaxInt64. +type SurvivorSurvey struct { + EmptyPartMinSeq int64 + PartMinSeq map[string]int64 +} + +// NewSurvivorSurvey returns a survey initialized with the no-survivor +// sentinel for the empty-partition bucket and an empty per-partition +// map. Callers populate via AddSurvivor. +func NewSurvivorSurvey() *SurvivorSurvey { + return &SurvivorSurvey{ + EmptyPartMinSeq: math.MaxInt64, + PartMinSeq: make(map[string]int64), + } +} + +// AddSurvivor records a surviving data file's (partition, seq) into +// the survey. partition is the data file's partition tuple from +// iceberg.DataFile.Partition() (nil/empty for unpartitioned tables); +// seq is the data file's sequence number from the manifest entry. +// +// Defensive: if seq < 0 (sentinel for "unset"), the file is recorded +// with seq=0 (smallest real value), which keeps it permanently alive +// against every eq-delete. Better to preserve uncertain state than to +// drop deletes that may still apply. +// +// Pointer receiver: EmptyPartMinSeq is int64 (value type) and we need +// updates to persist. Callers must hold the survey by value or pointer +// consistently — typical pattern is `survey := NewSurvivorSurvey()` +// followed by `survey.AddSurvivor(...)` which Go auto-addresses. +func (s *SurvivorSurvey) AddSurvivor(partition map[int]any, seq int64) { + if seq < 0 { + seq = 0 + } + if len(partition) == 0 { + if seq < s.EmptyPartMinSeq { + s.EmptyPartMinSeq = seq + } Review Comment: ```suggestion s.EmptyPartMinSeq = min(seq, s.EmptyPartMinSeq) ``` ########## table/rewrite_data_files.go: ########## @@ -63,24 +71,54 @@ type CompactionTaskGroup struct { TotalSizeBytes int64 } -// RewriteDataFiles compacts the given groups by reading data with deletes -// applied, writing new consolidated files, and atomically replacing the -// old files. Position delete files that are fully applied are removed. -// -// Use table/compaction.Config.PlanCompaction() to produce the groups, -// then convert compaction.Group → CompactionTaskGroup and pass them here. +// RewriteDataFilesOptions bundles the per-rewrite knobs for +// Transaction.RewriteDataFiles. +type RewriteDataFilesOptions struct { + // PartialProgress, when true, stages each group via ReplaceFiles + // inside the loop so work survives a mid-loop write failure. When + // false (the default), all groups are committed in a single atomic + // snapshot. + // + // In both modes the final catalog commit happens once at + // Transaction.Commit() time. True per-group durability (matching + // Java's behavior) requires committing separate transactions per + // group, which is left to the caller. + PartialProgress bool + + // SnapshotProps are added to the rewrite snapshot's summary. + SnapshotProps iceberg.Properties + + // ExtraDeleteFilesToRemove are delete files (typically equality + // deletes that are dead after the rewrite) that the caller wants + // expunged in the same snapshot as the rewrite. The executor + // passes them through to ReplaceFiles unchanged. Honored only + // when PartialProgress is false. + // + // Use table/compaction.CollectDeadEqualityDeletes to compute this + // list from the current snapshot. Position delete files that are + // fully applied are removed automatically and do NOT need to be + // passed in here. + ExtraDeleteFilesToRemove []iceberg.DataFile +} + +// RewriteDataFiles compacts the given groups by reading data with +// deletes applied, writing new consolidated files, and atomically +// replacing the old files. Position delete files that are fully +// applied (every referenced data file is in the rewrite set) are +// removed automatically. // -// Note: partialProgress accumulates all group commits within this -// transaction. True per-group durability (matching Java's behavior) -// requires committing separate transactions per group, which is left -// to the caller. When partialProgress is false, all groups are committed -// in a single atomic snapshot. +// Equality-delete cleanup is the caller's responsibility: compute the +// dead set with table/compaction.CollectDeadEqualityDeletes (against +// the same snapshot the rewrite is staged on) and pass it via +// opts.ExtraDeleteFilesToRemove. The executor only orchestrates the +// commit; it does not impose a cleanup policy. This split keeps the +// pure spec predicate in table/compaction and the unexported snapshot +// machinery in table. Review Comment: surround the references with brackets (i.e. `[compaction.CollectDeadEqualityDeletes]`) so that the docstrings will cross link in pkg.go.dev ########## table/compaction/eq_delete_decision.go: ########## @@ -0,0 +1,214 @@ +// 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 compaction + +import ( + "fmt" + "math" + "slices" + + "github.com/apache/iceberg-go" +) + +// SurvivorSurvey describes the surviving data files in a snapshot +// AFTER a planned rewrite has logically removed its rewrite set. It +// is the input to DecideDeadEqualityDeletes. +// +// EmptyPartMinSeq is the smallest sequence number among unpartitioned +// surviving data files. Per the Iceberg v2 reader predicate (see +// table/scanner.go matchEqualityDeletesToData), an unpartitioned data +// file applies to every equality delete — so it is always part of an +// eq-delete's "applicable survivors" minimum. +// +// PartMinSeq maps the partition tuple (encoded via partitionMatchKey) +// to the smallest sequence number among surviving partitioned data +// files in that tuple. The key intentionally does NOT include +// SpecID — the reader's predicate ignores SpecID, so the writer-side +// cleanup must too. +// +// Sentinel "no survivor in this bucket" is math.MaxInt64. +type SurvivorSurvey struct { + EmptyPartMinSeq int64 + PartMinSeq map[string]int64 +} + +// NewSurvivorSurvey returns a survey initialized with the no-survivor +// sentinel for the empty-partition bucket and an empty per-partition +// map. Callers populate via AddSurvivor. +func NewSurvivorSurvey() *SurvivorSurvey { + return &SurvivorSurvey{ + EmptyPartMinSeq: math.MaxInt64, + PartMinSeq: make(map[string]int64), + } +} + +// AddSurvivor records a surviving data file's (partition, seq) into +// the survey. partition is the data file's partition tuple from +// iceberg.DataFile.Partition() (nil/empty for unpartitioned tables); +// seq is the data file's sequence number from the manifest entry. +// +// Defensive: if seq < 0 (sentinel for "unset"), the file is recorded +// with seq=0 (smallest real value), which keeps it permanently alive +// against every eq-delete. Better to preserve uncertain state than to +// drop deletes that may still apply. +// +// Pointer receiver: EmptyPartMinSeq is int64 (value type) and we need +// updates to persist. Callers must hold the survey by value or pointer +// consistently — typical pattern is `survey := NewSurvivorSurvey()` +// followed by `survey.AddSurvivor(...)` which Go auto-addresses. +func (s *SurvivorSurvey) AddSurvivor(partition map[int]any, seq int64) { + if seq < 0 { + seq = 0 + } + if len(partition) == 0 { + if seq < s.EmptyPartMinSeq { + s.EmptyPartMinSeq = seq + } + + return + } + key := partitionMatchKey(partition) + if cur, ok := s.PartMinSeq[key]; !ok || seq < cur { + s.PartMinSeq[key] = seq + } +} + +// applicableMinSeq returns the smallest surviving-D seq number that +// the eq-delete with the given partition could apply to. Mirrors the +// scanner's predicate: an unpartitioned surviving D applies to every +// E (so EmptyPartMinSeq is always in the min); an unpartitioned E +// applies to every surviving D (so we min across every PartMinSeq). +func (s *SurvivorSurvey) applicableMinSeq(eqPartition map[int]any) int64 { + if len(eqPartition) == 0 { + min := s.EmptyPartMinSeq + for _, v := range s.PartMinSeq { + if v < min { + min = v + } + } Review Comment: `return min(s.EmptyPartMinSeq, slices.Min(maps.Values(s.PartMinSeq)))` ? ########## table/compaction/eq_delete_decision.go: ########## @@ -0,0 +1,214 @@ +// 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 compaction + +import ( + "fmt" + "math" + "slices" + + "github.com/apache/iceberg-go" +) + +// SurvivorSurvey describes the surviving data files in a snapshot +// AFTER a planned rewrite has logically removed its rewrite set. It +// is the input to DecideDeadEqualityDeletes. +// +// EmptyPartMinSeq is the smallest sequence number among unpartitioned +// surviving data files. Per the Iceberg v2 reader predicate (see +// table/scanner.go matchEqualityDeletesToData), an unpartitioned data +// file applies to every equality delete — so it is always part of an +// eq-delete's "applicable survivors" minimum. +// +// PartMinSeq maps the partition tuple (encoded via partitionMatchKey) +// to the smallest sequence number among surviving partitioned data +// files in that tuple. The key intentionally does NOT include +// SpecID — the reader's predicate ignores SpecID, so the writer-side +// cleanup must too. +// +// Sentinel "no survivor in this bucket" is math.MaxInt64. +type SurvivorSurvey struct { + EmptyPartMinSeq int64 + PartMinSeq map[string]int64 +} + +// NewSurvivorSurvey returns a survey initialized with the no-survivor +// sentinel for the empty-partition bucket and an empty per-partition +// map. Callers populate via AddSurvivor. +func NewSurvivorSurvey() *SurvivorSurvey { + return &SurvivorSurvey{ + EmptyPartMinSeq: math.MaxInt64, + PartMinSeq: make(map[string]int64), + } +} + +// AddSurvivor records a surviving data file's (partition, seq) into +// the survey. partition is the data file's partition tuple from +// iceberg.DataFile.Partition() (nil/empty for unpartitioned tables); +// seq is the data file's sequence number from the manifest entry. +// +// Defensive: if seq < 0 (sentinel for "unset"), the file is recorded +// with seq=0 (smallest real value), which keeps it permanently alive +// against every eq-delete. Better to preserve uncertain state than to +// drop deletes that may still apply. +// +// Pointer receiver: EmptyPartMinSeq is int64 (value type) and we need +// updates to persist. Callers must hold the survey by value or pointer +// consistently — typical pattern is `survey := NewSurvivorSurvey()` +// followed by `survey.AddSurvivor(...)` which Go auto-addresses. +func (s *SurvivorSurvey) AddSurvivor(partition map[int]any, seq int64) { + if seq < 0 { + seq = 0 + } + if len(partition) == 0 { + if seq < s.EmptyPartMinSeq { + s.EmptyPartMinSeq = seq + } + + return + } + key := partitionMatchKey(partition) + if cur, ok := s.PartMinSeq[key]; !ok || seq < cur { + s.PartMinSeq[key] = seq + } +} + +// applicableMinSeq returns the smallest surviving-D seq number that +// the eq-delete with the given partition could apply to. Mirrors the +// scanner's predicate: an unpartitioned surviving D applies to every +// E (so EmptyPartMinSeq is always in the min); an unpartitioned E +// applies to every surviving D (so we min across every PartMinSeq). +func (s *SurvivorSurvey) applicableMinSeq(eqPartition map[int]any) int64 { + if len(eqPartition) == 0 { + min := s.EmptyPartMinSeq + for _, v := range s.PartMinSeq { + if v < min { + min = v + } + } + + return min + } + min := s.EmptyPartMinSeq + if v, ok := s.PartMinSeq[partitionMatchKey(eqPartition)]; ok && v < min { + min = v + } + + return min +} + +// DecideDeadEqualityDeletes is the pure spec-correctness predicate for +// equality-delete cleanup during compaction. Given a survey of +// surviving data file seqs (already excluding the rewrite set) and a +// list of equality-delete manifest entries, it returns the eq-delete +// files that no surviving data file could ever apply to. +// +// The rule is identical to scanner.matchEqualityDeletesToData (the +// reader-side filter): +// +// E applies to D iff E.seq > D.seq AND ( +// len(E.partition) == 0 || +// len(D.partition) == 0 || +// partitionsMatch(E.partition, D.partition) +// ) +// +// E is dead iff no applicable surviving D has D.seq < E.seq — +// equivalently, the applicable min-seq is >= E.seq. +// +// SpecID is intentionally NOT part of the predicate. The Iceberg-go +// reader does not consult it; if the executor used a stricter +// predicate it could drop eq-deletes the reader still applies, causing +// silent data loss under partition-spec evolution. +// +// Defensive: candidates with sequence number < 0 (sentinel for unset) +// are skipped — preserved rather than risk dropping an unidentifiable +// file. +// +// Dedup by file path: the same eq-delete file may appear in multiple +// manifest entries after manifest merging. +func DecideDeadEqualityDeletes(survey *SurvivorSurvey, candidates []iceberg.ManifestEntry) []iceberg.DataFile { + if survey == nil { + return nil + } + if len(candidates) == 0 { + return nil + } Review Comment: just combine these into a single check `if survey == nil || len(candidates) == 0 { return nil }` -- 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]
