laskoviymishka commented on code in PR #947: URL: https://github.com/apache/iceberg-go/pull/947#discussion_r3167093879
########## 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: Done, added slices.Collect because slices.Min takes a slice, not iter.Seq ########## 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: make sense -- 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]
