This is an automated email from the ASF dual-hosted git repository.
zeroshade 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 f4f0f0e76 fix(table): validate path-scoped position delete partitions
(#2471)
f4f0f0e76 is described below
commit f4f0f0e760c57a3681041ff21897d640ef4c2afb
Author: Minh Vu <[email protected]>
AuthorDate: Mon Nov 9 21:23:49 2026 +0100
fix(table): validate path-scoped position delete partitions (#2471)
fix(table): validate path-scoped position deletes
---
table/positional_delete_index.go | 37 ++++++++---
table/positional_delete_index_test.go | 77 ++++++++++++++++------
...write_position_delete_files_integration_test.go | 14 ++--
3 files changed, 91 insertions(+), 37 deletions(-)
diff --git a/table/positional_delete_index.go b/table/positional_delete_index.go
index 476a01b35..86296bae0 100644
--- a/table/positional_delete_index.go
+++ b/table/positional_delete_index.go
@@ -187,9 +187,10 @@ func buildPositionalDeleteIndex(entries
[]iceberg.ManifestEntry) (*positionalDel
return idx, nil
}
-// forDataFile returns positional deletes with a greater than or equal sequence
-// number. Partition-scoped candidates are pruned using file_path metrics and
-// returned before path-scoped deletes, matching Java's ordering.
+// forDataFile returns applicable positional deletes and validates partition
+// metadata for path-scoped deletes. Partition-scoped candidates are pruned
+// using file_path metrics and returned before path-scoped deletes, matching
+// Java's ordering.
func (idx *positionalDeleteIndex) forDataFile(dataEntry iceberg.ManifestEntry)
([]iceberg.DataFile, error) {
if len(idx.byPath) == 0 && len(idx.byPartition) == 0 &&
len(idx.byIntegerPartition) == 0 {
return nil, nil
@@ -216,8 +217,10 @@ func (idx *positionalDeleteIndex) forDataFile(dataEntry
iceberg.ManifestEntry) (
if err != nil {
return nil, err
}
- out = appendPositionalDeletesFromSequence(
- out, idx.byPath[dataFile.FilePath()], dataSeqNum)
+ out, err = appendPositionalDeletesFromSequence(out,
idx.byPath[dataFile.FilePath()], dataEntry)
+ if err != nil {
+ return nil, err
+ }
return out, nil
}
@@ -255,14 +258,30 @@ func filePathMayMatch(deleteFile iceberg.DataFile,
dataFilePath string) bool {
func appendPositionalDeletesFromSequence(
out []iceberg.DataFile,
entries []iceberg.ManifestEntry,
- dataSeqNum int64,
-) []iceberg.DataFile {
+ dataEntry iceberg.ManifestEntry,
+) ([]iceberg.DataFile, error) {
+ if len(entries) == 0 {
+ return out, nil
+ }
+
+ dataFile := dataEntry.DataFile()
+ dataPartition := dataFilePartition(dataFile)
+ dataSeqNum := dataEntry.DataSequenceNumber()
start := sort.Search(len(entries), func(i int) bool {
return entries[i].DataSequenceNumber() >= dataSeqNum
})
for _, entry := range entries[start:] {
- out = append(out, entry.DataFile())
+ deleteFile := entry.DataFile()
+ if deleteFile.SpecID() != dataFile.SpecID() {
+ return nil, fmt.Errorf("%w: position delete file %s
uses partition spec %d, but data file %s uses partition spec %d",
+ ErrInvalidMetadata, deleteFile.FilePath(),
deleteFile.SpecID(), dataFile.FilePath(), dataFile.SpecID())
+ }
+ if !partitionsMatch(dataPartition,
dataFilePartition(deleteFile)) {
+ return nil, fmt.Errorf("%w: position delete file %s
partition does not match data file %s",
+ ErrInvalidMetadata, deleteFile.FilePath(),
dataFile.FilePath())
+ }
+ out = append(out, deleteFile)
}
- return out
+ return out, nil
}
diff --git a/table/positional_delete_index_test.go
b/table/positional_delete_index_test.go
index 3f40700f6..d71c9858f 100644
--- a/table/positional_delete_index_test.go
+++ b/table/positional_delete_index_test.go
@@ -215,11 +215,11 @@ func
TestPositionalDeleteIndexMatchesReferenceApplicability(t *testing.T) {
"partition-us-after.parquet", 1, usPartition, 8, nil,
"data-x.parquet", "data-z.parquet"),
newPositionalDeleteIndexTestEntry(
- "path-explicit-a-same-sequence.parquet", 99,
euPartition, 5, &dataA, ""),
+ "path-explicit-a-same-sequence.parquet", 1,
usPartition, 5, &dataA, ""),
newPositionalDeleteIndexTestEntry(
"partition-eu-newer.parquet", 1, euPartition, 6, nil,
""),
newPositionalDeleteIndexTestEntry(
- "path-inferred-b-newer.parquet", 2, euPartition, 7,
nil, dataB),
+ "path-inferred-b-newer.parquet", 1, usPartition, 7,
nil, dataB),
newPositionalDeleteIndexTestEntry(
"partition-us-older.parquet", 1, usPartition, 4, nil,
""),
newPositionalDeleteIndexTestEntryWithBounds(
@@ -234,7 +234,7 @@ func
TestPositionalDeleteIndexMatchesReferenceApplicability(t *testing.T) {
"partition-us-before.parquet", 1, usPartition, 8, nil,
"data-0.parquet", "data-9.parquet"),
newPositionalDeleteIndexTestEntry(
- "path-inferred-a-newer.parquet", 2, euPartition, 7,
nil, dataA),
+ "path-inferred-a-newer.parquet", 1, usPartition, 7,
nil, dataA),
newPositionalDeleteIndexTestEntry(
"partition-us-same-sequence.parquet", 1, usPartition,
5, nil, ""),
newPositionalDeleteIndexTestEntry(
@@ -286,6 +286,10 @@ func referencePositionalDeleteApplies(
dataFile := dataEntry.DataFile()
deleteFile := deleteEntry.DataFile()
+ if deleteFile.SpecID() != dataFile.SpecID() ||
+ !referencePartitionsEqual(deleteFile.Partition(),
dataFile.Partition()) {
+ return false
+ }
if referencedPath := deleteFile.ReferencedDataFile(); referencedPath !=
nil {
return *referencedPath == dataFile.FilePath()
}
@@ -296,11 +300,6 @@ func referencePositionalDeleteApplies(
return string(lower) == dataFile.FilePath()
}
- if deleteFile.SpecID() != dataFile.SpecID() ||
- !referencePartitionsEqual(deleteFile.Partition(),
dataFile.Partition()) {
- return false
- }
-
path := []byte(dataFile.FilePath())
if hasLower && bytes.Compare(path, lower) < 0 {
return false
@@ -521,19 +520,59 @@ func
TestPositionalDeleteIndexHandlesUnknownSequenceNumbers(t *testing.T) {
positionalDeletePaths(matched))
}
-func TestPositionalDeleteIndexSkipsPartitionKeyForPathOnlyDeletes(t
*testing.T) {
+func TestPositionalDeleteIndexRejectsPathDeletePartitionMismatch(t *testing.T)
{
dataPath := "data.parquet"
- deleteEntry := newPositionalDeleteIndexTestEntry(
- "delete.parquet", 1, nil, 2, &dataPath, "")
-
- idx, err :=
buildPositionalDeleteIndex([]iceberg.ManifestEntry{deleteEntry})
- require.NoError(t, err)
+ partition := map[int]any{1000: "us"}
+
+ for _, tc := range []struct {
+ name string
+ deleteSpecID int32
+ deletePartition map[int]any
+ dataSequence int64
+ boundsScoped bool
+ wantErr bool
+ }{
+ {
+ name: "different spec ID", deleteSpecID: 2,
+ deletePartition: partition, dataSequence: 1, wantErr:
true,
+ },
+ {
+ name: "different partition values", deleteSpecID: 1,
+ deletePartition: map[int]any{1000: "eu"}, dataSequence:
1, wantErr: true,
+ },
+ {
+ name: "different partition values with inferred path
scope", deleteSpecID: 1,
+ deletePartition: map[int]any{1000: "eu"}, dataSequence:
1,
+ boundsScoped: true, wantErr: true,
+ },
+ {
+ name: "older mismatched delete is not applicable",
deleteSpecID: 1,
+ deletePartition: map[int]any{1000: "eu"}, dataSequence:
3,
+ },
+ } {
+ t.Run(tc.name, func(t *testing.T) {
+ var deleteEntry iceberg.ManifestEntry
+ if tc.boundsScoped {
+ deleteEntry =
newPositionalDeleteIndexTestEntryWithBounds(
+ "delete.parquet", tc.deleteSpecID,
tc.deletePartition, 2, nil, dataPath, dataPath)
+ } else {
+ deleteEntry = newPositionalDeleteIndexTestEntry(
+ "delete.parquet", tc.deleteSpecID,
tc.deletePartition, 2, &dataPath, "")
+ }
+ idx, err :=
buildPositionalDeleteIndex([]iceberg.ManifestEntry{deleteEntry})
+ require.NoError(t, err)
- dataEntry := newPositionalDeleteIndexDataEntry(
- dataPath, 1, map[int]any{1000: struct{}{}}, 1)
- matched, err := idx.forDataFile(dataEntry)
- require.NoError(t, err)
- assert.Equal(t, []string{"delete.parquet"},
positionalDeletePaths(matched))
+ dataEntry :=
newPositionalDeleteIndexDataEntry(dataPath, 1, partition, tc.dataSequence)
+ matched, err := idx.forDataFile(dataEntry)
+ if tc.wantErr {
+ require.ErrorIs(t, err, ErrInvalidMetadata)
+ require.Nil(t, matched)
+ } else {
+ require.NoError(t, err)
+ require.Empty(t, matched)
+ }
+ })
+ }
}
func TestPositionalDeleteIndexRejectsUnsupportedPartitionValue(t *testing.T) {
diff --git a/table/rewrite_position_delete_files_integration_test.go
b/table/rewrite_position_delete_files_integration_test.go
index 80bf94ff0..6c91a069e 100644
--- a/table/rewrite_position_delete_files_integration_test.go
+++ b/table/rewrite_position_delete_files_integration_test.go
@@ -237,7 +237,7 @@ func TestRewritePositionDeleteFilesAcrossSequences(t
*testing.T) {
"stale positions must not gain applicability from the
replacement sequence")
}
-func TestRewritePositionDeleteFilesPreservesCrossSpecPathScopedDeletes(t
*testing.T) {
+func TestRewritePositionDeleteFilesPreservesPathScopedDeletesOnOldSpec(t
*testing.T) {
ctx := t.Context()
tbl, _ := newConcurrentRewriteTestTable(t)
schema, err := table.SchemaToArrowSchema(tbl.Schema(), nil, false,
false)
@@ -250,24 +250,20 @@ func
TestRewritePositionDeleteFilesPreservesCrossSpecPathScopedDeletes(t *testin
tbl, err = tx.Commit(ctx)
require.NoError(t, err)
- // The data remains on the old unpartitioned spec, while the new delete
- // files refer to its path from the evolved partition spec.
+ // The data and path-scoped deletes keep using the old unpartitioned
spec
+ // after the table's default spec evolves.
tx = tbl.NewTransaction()
require.NoError(t, tx.UpdateSpec(false).
AddField("id", iceberg.IdentityTransform{},
"id_identity").Commit())
tbl, err = tx.Commit(ctx)
require.NoError(t, err)
- spec := tbl.Spec()
- require.Equal(t, 1, spec.NumFields())
- partition := map[int]any{spec.Field(0).FieldID: int64(1)}
-
var deletes []iceberg.DataFile
for i := range 2 {
path := fmt.Sprintf("%s/data/cross-spec-delete-%d.parquet",
tbl.Location(), i)
writeParquetFile(t, path, table.PositionalDeleteArrowSchema,
fmt.Sprintf(`[{"file_path":%q,"pos":0}]`, dataPath))
- builder, buildErr := iceberg.NewDataFileBuilder(spec,
iceberg.EntryContentPosDeletes,
- path, iceberg.ParquetFile, partition, nil, nil, 1, 128)
+ builder, buildErr :=
iceberg.NewDataFileBuilder(*iceberg.UnpartitionedSpec,
+ iceberg.EntryContentPosDeletes, path,
iceberg.ParquetFile, nil, nil, nil, 1, 128)
require.NoError(t, buildErr)
deletes = append(deletes,
builder.ReferencedDataFile(dataPath).Build())
}
