laskoviymishka commented on code in PR #1434:
URL: https://github.com/apache/iceberg-go/pull/1434#discussion_r3578710660
##########
table/compaction/compaction.go:
##########
@@ -344,23 +344,42 @@ func fileScopedDeletedRows(t table.FileScanTask) int64 {
}
// isFileScoped reports whether a delete file applies to exactly one data file.
-// Mirrors Java ContentFileUtil.isFileScoped/referencedDataFile: a deletion
-// vector or path-scoped positional delete carries a referenced data file
-// either explicitly or, when referenced_data_file is absent (it is optional in
-// V2 and our own scan planning never sets it — see matchDeletesToData),
through
-// equal file_path lower/upper bounds. Equality deletes and partition-scoped
-// positional deletes are not file-scoped.
+// Equality deletes and partition-scoped positional deletes are not
file-scoped.
func isFileScoped(d iceberg.DataFile) bool {
if d.ContentType() == iceberg.EntryContentEqDeletes {
return false
}
- if d.ReferencedDataFile() != nil {
- return true
+
+ return referencedDataFilePath(d) != ""
Review Comment:
heads up that this quietly changes `isFileScoped` for a non-nil empty-string
`referenced_data_file`: the old path returned true on any non-nil ref, the new
one returns false because `referencedDataFilePath` treats `""` as unset. That's
the more correct answer, and our builder rejects empty refs at build anyway, so
it only bites foreign writers.
But it's silent, and the Java-parity paragraph that used to explain the
bounds fallback is gone from `isFileScoped` entirely. I'd keep the change, move
a line of that explanation back onto `isFileScoped`, and add a direct
`isFileScoped` case for an empty-string ref (expect false) so the tightening is
pinned. wdyt?
##########
table/compaction/pos_delete_collect.go:
##########
@@ -0,0 +1,221 @@
+// 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"
+ "slices"
+
+ "github.com/apache/iceberg-go"
+ iceio "github.com/apache/iceberg-go/io"
+ "github.com/apache/iceberg-go/table"
+)
+
+// CollectDeadPositionDeletes walks the given snapshot's manifests and returns
+// the position-delete files made dead by a rewrite that removes
rewrittenPaths.
+// A dead position delete references only data files the rewrite is removing,
so
+// after the rewrite it applies to nothing and is safe to expunge in the same
+// commit.
+//
+// Scope resolution mirrors Java ContentFileUtil.referencedDataFile and
+// DeleteFileIndex:
+//
+// - File-scoped — explicit referenced_data_file, or equal file_path
+// lower/upper bounds — is dead iff its single target is in rewrittenPaths.
+// - Partition-scoped — no single target — is dead iff no surviving
+// (non-rewritten) data file in the same (specID, partition) has a sequence
+// number <= the delete's. Such a survivor is one the delete still applies
+// to, so the delete must be retained for a later dangling-delete pass.
+//
+// Deletion vectors (Puffin position deletes) are intentionally excluded: they
+// are 1:1 with their data file and expunged per-group via
+// [table.CompactionGroupResult.SafeDeletionVectors].
+//
+// rewrittenPaths is the union of every old data file path being replaced
across
+// all rewrite groups. This is the whole-rewrite re-check that
+// [table.CollectSafePositionDeletes]'s caller contract requires whenever
+// multiple groups land in one rewrite snapshot; the result feeds
+// [table.RewriteDataFilesOptions].ExtraDeleteFilesToRemove. Like
+// [CollectDeadEqualityDeletes], the returned files are safe to remove in the
+// same commit that stages the rewrite: a concurrent
+// commit cannot resurrect them, because any concurrent delete touching a
+// rewritten file is rejected by the rewrite validator and any concurrent data
+// file gets a sequence number strictly greater than every preexisting delete.
+//
+// Two-pass design: the first pass walks delete manifests to resolve
candidates,
+// deciding file-scoped ones immediately; the more expensive data-manifest walk
+// runs only when a partition-scoped candidate needs a survivor check.
+func CollectDeadPositionDeletes(
+ 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, resolving each candidate's expunge
scope.
+ candidates, err := collectPositionDeleteCandidates(ctx, fs, manifests)
+ if err != nil {
+ return nil, err
+ }
+
+ // Second pass runs only when a partition-scoped candidate needs a
survivor
+ // check; file-scoped ones are decided from rewrittenPaths alone.
+ var minSurvivorSeq map[string]int64
+ if slices.ContainsFunc(candidates, func(c positionDeleteCandidate) bool
{ return c.fileScopedTarget == "" }) {
Review Comment:
small one — `collectPositionDeleteCandidates` already walks every candidate,
so it could hand back a `hasPartitionScoped` bool and save this re-scan (and
drop the `slices` import). Not important, just while we're here.
##########
table/compaction/pos_delete_collect.go:
##########
@@ -0,0 +1,221 @@
+// 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"
+ "slices"
+
+ "github.com/apache/iceberg-go"
+ iceio "github.com/apache/iceberg-go/io"
+ "github.com/apache/iceberg-go/table"
+)
+
+// CollectDeadPositionDeletes walks the given snapshot's manifests and returns
+// the position-delete files made dead by a rewrite that removes
rewrittenPaths.
+// A dead position delete references only data files the rewrite is removing,
so
+// after the rewrite it applies to nothing and is safe to expunge in the same
+// commit.
+//
+// Scope resolution mirrors Java ContentFileUtil.referencedDataFile and
+// DeleteFileIndex:
+//
+// - File-scoped — explicit referenced_data_file, or equal file_path
+// lower/upper bounds — is dead iff its single target is in rewrittenPaths.
+// - Partition-scoped — no single target — is dead iff no surviving
+// (non-rewritten) data file in the same (specID, partition) has a sequence
+// number <= the delete's. Such a survivor is one the delete still applies
+// to, so the delete must be retained for a later dangling-delete pass.
+//
+// Deletion vectors (Puffin position deletes) are intentionally excluded: they
+// are 1:1 with their data file and expunged per-group via
+// [table.CompactionGroupResult.SafeDeletionVectors].
+//
+// rewrittenPaths is the union of every old data file path being replaced
across
+// all rewrite groups. This is the whole-rewrite re-check that
+// [table.CollectSafePositionDeletes]'s caller contract requires whenever
+// multiple groups land in one rewrite snapshot; the result feeds
+// [table.RewriteDataFilesOptions].ExtraDeleteFilesToRemove. Like
+// [CollectDeadEqualityDeletes], the returned files are safe to remove in the
+// same commit that stages the rewrite: a concurrent
+// commit cannot resurrect them, because any concurrent delete touching a
+// rewritten file is rejected by the rewrite validator and any concurrent data
+// file gets a sequence number strictly greater than every preexisting delete.
+//
+// Two-pass design: the first pass walks delete manifests to resolve
candidates,
+// deciding file-scoped ones immediately; the more expensive data-manifest walk
+// runs only when a partition-scoped candidate needs a survivor check.
+func CollectDeadPositionDeletes(
+ 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, resolving each candidate's expunge
scope.
+ candidates, err := collectPositionDeleteCandidates(ctx, fs, manifests)
+ if err != nil {
+ return nil, err
+ }
+
+ // Second pass runs only when a partition-scoped candidate needs a
survivor
+ // check; file-scoped ones are decided from rewrittenPaths alone.
+ var minSurvivorSeq map[string]int64
+ if slices.ContainsFunc(candidates, func(c positionDeleteCandidate) bool
{ return c.fileScopedTarget == "" }) {
+ minSurvivorSeq, err = minSurvivorSeqByPartition(ctx, fs,
manifests, rewrittenPaths)
+ if err != nil {
+ return nil, err
+ }
+ }
+
+ return decideDeadPositionDeletes(candidates, rewrittenPaths,
minSurvivorSeq), nil
+}
+
+// positionDeleteCandidate is a classic position-delete file resolved to its
+// expunge scope: fileScopedTarget is the single referenced data file for a
+// file-scoped delete, or "" for a partition-scoped one, which is then keyed by
+// partitionKey and gated by the delete's sequence number seq.
+type positionDeleteCandidate struct {
+ df iceberg.DataFile
+ fileScopedTarget string
+ partitionKey string
+ seq int64
+}
+
+// collectPositionDeleteCandidates walks the delete manifests and returns the
+// classic position-delete files (deletion vectors excluded), deduplicated by
+// path and resolved to their expunge scope.
+func collectPositionDeleteCandidates(ctx context.Context, fs iceio.IO,
manifests []iceberg.ManifestFile) ([]positionDeleteCandidate, error) {
+ var candidates []positionDeleteCandidate
+ seen := make(map[string]struct{})
+ for _, m := range manifests {
+ if cerr := ctx.Err(); cerr != nil {
+ return nil, cerr
+ }
+ if m.ManifestContent() != iceberg.ManifestContentDeletes {
+ continue
+ }
+ for e, err := range m.Entries(fs, true) {
+ if err != nil {
+ return nil, err
+ }
+ df := e.DataFile()
+ if df.ContentType() != iceberg.EntryContentPosDeletes
|| isDeletionVector(df) {
+ continue
+ }
+ path := df.FilePath()
+ if _, ok := seen[path]; ok {
+ continue
+ }
+ seen[path] = struct{}{}
+
+ if target := referencedDataFilePath(df); target != "" {
+ candidates = append(candidates,
positionDeleteCandidate{df: df, fileScopedTarget: target})
+
+ continue
+ }
+
+ // Partition-scoped. A negative sequence number is the
"unset"
+ // sentinel; retain rather than risk expunging a delete
whose
+ // applicability cannot be established.
+ if seq := e.SequenceNum(); seq >= 0 {
+ candidates = append(candidates,
positionDeleteCandidate{
+ df: df,
+ partitionKey:
partitionBucketKey(df.SpecID(), df.Partition()),
+ seq: seq,
+ })
+ }
+ }
+ }
+
+ return candidates, nil
+}
+
+// minSurvivorSeqByPartition walks the data manifests and returns, per (specID,
+// partition), the smallest sequence number among surviving (non-rewritten)
data
+// files. A missing key means the partition has no survivor.
+func minSurvivorSeqByPartition(ctx context.Context, fs iceio.IO, manifests
[]iceberg.ManifestFile, rewrittenPaths map[string]struct{}) (map[string]int64,
error) {
+ minSeq := make(map[string]int64)
+ for _, m := range manifests {
+ if cerr := ctx.Err(); cerr != nil {
+ return nil, cerr
+ }
+ if m.ManifestContent() != iceberg.ManifestContentData {
+ continue
+ }
+ for e, err := range m.Entries(fs, true) {
+ if err != nil {
+ return nil, err
+ }
+ df := e.DataFile()
+ if df.ContentType() != iceberg.EntryContentData {
+ continue
+ }
+ if _, rewritten := rewrittenPaths[df.FilePath()];
rewritten {
+ continue
+ }
+ seq := max(e.SequenceNum(), 0)
+ key := partitionBucketKey(df.SpecID(), df.Partition())
+ if cur, ok := minSeq[key]; ok {
+ seq = min(seq, cur)
+ }
+ minSeq[key] = seq
+ }
+ }
+
+ return minSeq, nil
+}
+
+// decideDeadPositionDeletes returns the candidates that are dead after the
+// rewrite. A file-scoped delete is dead when its single target is rewritten. A
+// partition-scoped delete is dead when no surviving data file in its partition
+// predates it — position deletes apply when data.seq <= delete.seq, so a
+// survivor with seq <= the delete's seq keeps it alive.
+func decideDeadPositionDeletes(candidates []positionDeleteCandidate,
rewrittenPaths map[string]struct{}, minSurvivorSeq map[string]int64)
[]iceberg.DataFile {
+ dead := make([]iceberg.DataFile, 0, len(candidates))
+ for _, c := range candidates {
+ if c.fileScopedTarget != "" {
+ if _, rewritten := rewrittenPaths[c.fileScopedTarget];
rewritten {
+ dead = append(dead, c.df)
+ }
+
+ continue
+ }
+ if cur, ok := minSurvivorSeq[c.partitionKey]; !ok || cur >
c.seq {
+ dead = append(dead, c.df)
+ }
+ }
+
+ return dead
+}
+
+// isDeletionVector reports whether df is a deletion vector: a Puffin file with
+// position-delete content. Mirrors table.isDeletionVector, reimplemented here
+// because that predicate is unexported.
+func isDeletionVector(df iceberg.DataFile) bool {
Review Comment:
this is now the third byte-for-byte copy of `table.isDeletionVector`
(`scanner.go` and `mor_delete_pruning_test.go` have the others). Since
compaction is the second real consumer, I'd lean toward exporting it from the
`table` package and dropping the copies. If we'd rather not widen that surface,
a `//nolint:dupl` with a TODO would at least mark it as known. wdyt?
##########
table/compaction/pos_delete_collect_ext_test.go:
##########
@@ -0,0 +1,194 @@
+// 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_test
+
+import (
+ "context"
+ "fmt"
+ "path/filepath"
+ "testing"
+
+ "github.com/apache/iceberg-go"
+ iceio "github.com/apache/iceberg-go/io"
+ "github.com/apache/iceberg-go/table"
+ "github.com/apache/iceberg-go/table/compaction"
+ "github.com/stretchr/testify/require"
+)
+
+// TestCollectDeadPositionDeletes drives the full manifest walk: a
+// partition-scoped position delete (no referenced_data_file, no file_path
+// bounds — Spark's default granularity) covering three data files is dead only
+// when every file it covers is rewritten, and retained the moment one is not.
+func TestCollectDeadPositionDeletes(t *testing.T) {
+ ctx := t.Context()
+ fs := iceio.LocalFS{}
+ tbl := newCDCStressTable(t)
+
+ arrowSc, err := table.SchemaToArrowSchema(tbl.Schema(), nil, false,
false)
+ require.NoError(t, err)
+
+ for i := range 3 {
+ p := tbl.Location() + fmt.Sprintf("/data/d-%d.parquet", i)
+ writeParquet(t, p, arrowSc, fmt.Sprintf(`[{"id": %d, "data":
"r%d"}]`, i+1, i+1))
+ tx := tbl.NewTransaction()
+ require.NoError(t, tx.AddFiles(ctx, []string{p}, nil, false))
+ tbl, err = tx.Commit(ctx)
+ require.NoError(t, err)
+ }
+
+ dataPaths := dataFilePaths(t, tbl)
+ require.Len(t, dataPaths, 3)
+
+ // A partition-scoped position delete: unpartitioned spec, no ref, no
+ // bounds. Committed after the data files, so its sequence number
exceeds
+ // theirs and it applies to all three.
+ delDF, err := iceberg.NewDataFileBuilder(
+ *iceberg.UnpartitionedSpec, iceberg.EntryContentPosDeletes,
+ tbl.Location()+"/data/pos-del.parquet", iceberg.ParquetFile,
nil, nil, nil, 1, 128)
+ require.NoError(t, err)
+
+ tx := tbl.NewTransaction()
+ rd := tx.NewRowDelta(nil)
+ rd.AddDeletes(delDF.Build())
+ require.NoError(t, rd.Commit(ctx))
+ tbl, err = tx.Commit(ctx)
+ require.NoError(t, err)
+
+ t.Run("all covered files rewritten is dead", func(t *testing.T) {
+ rewritten := pathSet(dataPaths...)
+ dead, err := compaction.CollectDeadPositionDeletes(ctx, fs,
tbl.CurrentSnapshot(), rewritten)
+ require.NoError(t, err)
+ require.Len(t, dead, 1, "a partition-scoped delete whose every
covered file is rewritten must be expunged")
+ })
Review Comment:
this asserts the right files come back, but nothing here commits the expunge
through `RewriteDataFiles` and re-scans — so the actual integration path
(Collect → `ExtraDeleteFilesToRemove` → commit → scan) is never exercised. The
eq-delete side has `TestCDCStress` doing exactly that round-trip.
I'd add one test that routes the collected files through
`ExtraDeleteFilesToRemove`, commits, and re-scans to prove no deleted row
reappears — it'd also immediately catch the `RemovedEqualityDeleteFiles`
miscount I raised in the review. Fine to gate under `-short` if it's heavy.
##########
table/compaction/pos_delete_collect.go:
##########
@@ -0,0 +1,221 @@
+// 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"
+ "slices"
+
+ "github.com/apache/iceberg-go"
+ iceio "github.com/apache/iceberg-go/io"
+ "github.com/apache/iceberg-go/table"
+)
+
+// CollectDeadPositionDeletes walks the given snapshot's manifests and returns
+// the position-delete files made dead by a rewrite that removes
rewrittenPaths.
+// A dead position delete references only data files the rewrite is removing,
so
+// after the rewrite it applies to nothing and is safe to expunge in the same
+// commit.
+//
+// Scope resolution mirrors Java ContentFileUtil.referencedDataFile and
+// DeleteFileIndex:
+//
+// - File-scoped — explicit referenced_data_file, or equal file_path
+// lower/upper bounds — is dead iff its single target is in rewrittenPaths.
+// - Partition-scoped — no single target — is dead iff no surviving
+// (non-rewritten) data file in the same (specID, partition) has a sequence
+// number <= the delete's. Such a survivor is one the delete still applies
+// to, so the delete must be retained for a later dangling-delete pass.
+//
+// Deletion vectors (Puffin position deletes) are intentionally excluded: they
+// are 1:1 with their data file and expunged per-group via
+// [table.CompactionGroupResult.SafeDeletionVectors].
+//
+// rewrittenPaths is the union of every old data file path being replaced
across
+// all rewrite groups. This is the whole-rewrite re-check that
+// [table.CollectSafePositionDeletes]'s caller contract requires whenever
+// multiple groups land in one rewrite snapshot; the result feeds
+// [table.RewriteDataFilesOptions].ExtraDeleteFilesToRemove. Like
+// [CollectDeadEqualityDeletes], the returned files are safe to remove in the
+// same commit that stages the rewrite: a concurrent
+// commit cannot resurrect them, because any concurrent delete touching a
+// rewritten file is rejected by the rewrite validator and any concurrent data
+// file gets a sequence number strictly greater than every preexisting delete.
+//
+// Two-pass design: the first pass walks delete manifests to resolve
candidates,
+// deciding file-scoped ones immediately; the more expensive data-manifest walk
+// runs only when a partition-scoped candidate needs a survivor check.
+func CollectDeadPositionDeletes(
+ 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, resolving each candidate's expunge
scope.
+ candidates, err := collectPositionDeleteCandidates(ctx, fs, manifests)
+ if err != nil {
+ return nil, err
+ }
+
+ // Second pass runs only when a partition-scoped candidate needs a
survivor
+ // check; file-scoped ones are decided from rewrittenPaths alone.
+ var minSurvivorSeq map[string]int64
+ if slices.ContainsFunc(candidates, func(c positionDeleteCandidate) bool
{ return c.fileScopedTarget == "" }) {
+ minSurvivorSeq, err = minSurvivorSeqByPartition(ctx, fs,
manifests, rewrittenPaths)
+ if err != nil {
+ return nil, err
+ }
+ }
+
+ return decideDeadPositionDeletes(candidates, rewrittenPaths,
minSurvivorSeq), nil
+}
+
+// positionDeleteCandidate is a classic position-delete file resolved to its
+// expunge scope: fileScopedTarget is the single referenced data file for a
+// file-scoped delete, or "" for a partition-scoped one, which is then keyed by
+// partitionKey and gated by the delete's sequence number seq.
+type positionDeleteCandidate struct {
+ df iceberg.DataFile
+ fileScopedTarget string
+ partitionKey string
+ seq int64
+}
+
+// collectPositionDeleteCandidates walks the delete manifests and returns the
+// classic position-delete files (deletion vectors excluded), deduplicated by
+// path and resolved to their expunge scope.
+func collectPositionDeleteCandidates(ctx context.Context, fs iceio.IO,
manifests []iceberg.ManifestFile) ([]positionDeleteCandidate, error) {
+ var candidates []positionDeleteCandidate
+ seen := make(map[string]struct{})
+ for _, m := range manifests {
+ if cerr := ctx.Err(); cerr != nil {
+ return nil, cerr
+ }
+ if m.ManifestContent() != iceberg.ManifestContentDeletes {
+ continue
+ }
+ for e, err := range m.Entries(fs, true) {
+ if err != nil {
+ return nil, err
+ }
+ df := e.DataFile()
+ if df.ContentType() != iceberg.EntryContentPosDeletes
|| isDeletionVector(df) {
+ continue
+ }
+ path := df.FilePath()
+ if _, ok := seen[path]; ok {
+ continue
+ }
+ seen[path] = struct{}{}
+
+ if target := referencedDataFilePath(df); target != "" {
+ candidates = append(candidates,
positionDeleteCandidate{df: df, fileScopedTarget: target})
+
+ continue
+ }
+
+ // Partition-scoped. A negative sequence number is the
"unset"
+ // sentinel; retain rather than risk expunging a delete
whose
+ // applicability cannot be established.
+ if seq := e.SequenceNum(); seq >= 0 {
+ candidates = append(candidates,
positionDeleteCandidate{
+ df: df,
+ partitionKey:
partitionBucketKey(df.SpecID(), df.Partition()),
+ seq: seq,
+ })
+ }
+ }
+ }
+
+ return candidates, nil
+}
+
+// minSurvivorSeqByPartition walks the data manifests and returns, per (specID,
+// partition), the smallest sequence number among surviving (non-rewritten)
data
+// files. A missing key means the partition has no survivor.
+func minSurvivorSeqByPartition(ctx context.Context, fs iceio.IO, manifests
[]iceberg.ManifestFile, rewrittenPaths map[string]struct{}) (map[string]int64,
error) {
+ minSeq := make(map[string]int64)
+ for _, m := range manifests {
+ if cerr := ctx.Err(); cerr != nil {
+ return nil, cerr
+ }
+ if m.ManifestContent() != iceberg.ManifestContentData {
+ continue
+ }
+ for e, err := range m.Entries(fs, true) {
+ if err != nil {
+ return nil, err
+ }
+ df := e.DataFile()
+ if df.ContentType() != iceberg.EntryContentData {
+ continue
+ }
+ if _, rewritten := rewrittenPaths[df.FilePath()];
rewritten {
+ continue
+ }
+ seq := max(e.SequenceNum(), 0)
Review Comment:
the two passes treat the negative-sequence sentinel differently: the delete
side drops candidates with `seq < 0` and has a comment explaining why, while
this survivor side silently clamps it to 0 with `max()`. The net effect is
correct, but a reader landing on one side without the other could easily read
it as a bug. Could we mirror the delete-side comment here so the two sentinel
choices read as deliberate?
##########
table/compaction/pos_delete_collect_test.go:
##########
@@ -0,0 +1,127 @@
+// 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 (
+ "testing"
+
+ "github.com/apache/iceberg-go"
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+)
+
+func TestReferencedDataFilePath(t *testing.T) {
+ filePathID := filePathFieldID
+
+ t.Run("explicit referenced_data_file", func(t *testing.T) {
+ df := newPosDelete(t,
"del-a.parquet").ReferencedDataFile("data-a.parquet").Build()
+ assert.Equal(t, "data-a.parquet", referencedDataFilePath(df))
Review Comment:
nit — these use `assert.Equal` while the rest of both files use `require`.
Just for consistency I'd switch to `require.Equal` here.
--
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]
