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 19c888720 feat(table): rewrite v2 position delete files (#2409)
19c888720 is described below
commit 19c888720daf8f3fd8d0a2bb14cd9e8e32ce0a51
Author: Minh Vu <[email protected]>
AuthorDate: Wed Nov 4 19:38:51 2026 +0100
feat(table): rewrite v2 position delete files (#2409)
* feat(table): rewrite position delete files
Signed-off-by: Minh Vu <[email protected]>
* fix(table): preserve unscoped position deletes and validate positions
* fix(table): skip ineligible position delete rewrites and clean partial
outputs
* test(table): cover no-op delete rewrites with read-only file IO
* fix(table): repair delete rewrite lint and preserve no-op behavior
* fix(table): compact position deletes across source sequences
* test(table): cover source sequences in delete rewrite chunks
* test(table): retain delete applicability across commit sequences
* style(table): satisfy nlreturn in position delete rewrite
* test(table): assert position delete rewrites stop on pinned ref conflict
* fix(table): preserve cross-spec position deletes
* refactor(table): stage position delete rewrites in transactions
* fix(table): match location provider signature in position delete rewrite
* style(table): fix position delete rewrite test formatting
* style(table): fix position delete rewrite test formatting
---------
Signed-off-by: Minh Vu <[email protected]>
---
table/rewrite_position_delete_files.go | 480 +++++++++++++++++++++
...write_position_delete_files_integration_test.go | 341 +++++++++++++++
table/rewrite_position_delete_files_test.go | 141 ++++++
3 files changed, 962 insertions(+)
diff --git a/table/rewrite_position_delete_files.go
b/table/rewrite_position_delete_files.go
new file mode 100644
index 000000000..f60696892
--- /dev/null
+++ b/table/rewrite_position_delete_files.go
@@ -0,0 +1,480 @@
+// 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 (
+ "context"
+ "fmt"
+ "maps"
+ "slices"
+ "strings"
+
+ "github.com/apache/arrow-go/v18/arrow"
+ "github.com/apache/arrow-go/v18/arrow/array"
+ "github.com/apache/arrow-go/v18/arrow/compute"
+ "github.com/apache/iceberg-go"
+ tblutils "github.com/apache/iceberg-go/table/internal"
+ "github.com/google/uuid"
+)
+
+// PositionDeleteRewriteOptions controls position delete compaction.
+// Zero values use a minimum of two files and a maximum of 128 MiB of input
+// file sizes per group. Decoded rows are held in memory for sorting.
+type PositionDeleteRewriteOptions struct {
+ MinInputFiles int
+ MaxInputBytes int64
+}
+
+type positionDeleteRewriteKey struct {
+ partition string
+}
+
+type positionDeleteRewriteGroup struct {
+ files []iceberg.DataFile
+ partition map[int]any
+ sequences map[string]int64
+ maxSequence int64
+ bytes int64
+}
+
+// A job owns a file subset that fits the input budget and minimum count.
+type positionDeleteRewriteJob struct {
+ key positionDeleteRewriteKey
+ group *positionDeleteRewriteGroup
+}
+
+type positionDeleteLiveFile struct {
+ partition string
+ sequence int64
+}
+
+type positionDeleteRow struct {
+ path string
+ position int64
+}
+
+// PositionDeleteRewriteResult summarizes a position delete file rewrite.
+// Table is set when the table-level action has committed. Transaction callers
+// receive a nil Table until they commit their transaction.
+type PositionDeleteRewriteResult struct {
+ Table *Table
+
+ RemovedDeleteFiles int
+ AddedDeleteFiles int
+ BytesBefore int64
+ BytesAfter int64
+}
+
+// RewritePositionDeleteFiles compacts compatible v2 position delete files on
+// main and commits the rewrite. Use [Transaction.RewritePositionDeleteFiles]
+// to target another branch or combine the rewrite with other updates.
+//
+// The action holds decoded delete rows in memory. A concurrent commit
advancing
+// main invalidates the pinned planning snapshot and fails the whole commit
+// instead of retrying against the changed snapshot.
+func (t Table) RewritePositionDeleteFiles(
+ ctx context.Context, opts PositionDeleteRewriteOptions,
+) (*PositionDeleteRewriteResult, error) {
+ txn, err := t.NewTransactionOnBranch(MainBranch)
+ if err != nil {
+ return nil, err
+ }
+
+ result, err := txn.RewritePositionDeleteFiles(ctx, opts)
+ if err != nil {
+ return result, err
+ }
+
+ committed, err := txn.Commit(ctx)
+ if err == nil {
+ result.Table = committed
+ }
+
+ return result, err
+}
+
+// RewritePositionDeleteFiles stages a rewrite of compatible classic Parquet
+// position delete files without rewriting data files. It removes records
+// targeting data files that are no longer live in the target branch snapshot.
+//
+// Only v2 files in the current partition spec without optional deleted-row
+// payloads are selected. Groups never mix partition tuples. Each source's data
+// sequence controls its applicable rows; replacements inherit the maximum
+// source sequence. Physical source files remain available to older snapshots.
+//
+// The target branch snapshot is pinned. A concurrent commit advancing that
+// branch fails the eventual transaction commit rather than replaying the
+// rewrite. Decoded delete rows are kept in memory during sorting. Successfully
+// staged outputs must be retained until Transaction.Commit; abandoned
+// transactions may leave files for orphan cleanup.
+func (t *Transaction) RewritePositionDeleteFiles(
+ ctx context.Context, opts PositionDeleteRewriteOptions,
+) (*PositionDeleteRewriteResult, error) {
+ meta, err := t.txnMeta()
+ if err != nil {
+ return nil, err
+ }
+ if meta.formatVersion != 2 {
+ return nil, fmt.Errorf("%w: position delete rewrites require v2
table format", iceberg.ErrNotImplemented)
+ }
+ if opts.MinInputFiles <= 0 {
+ opts.MinInputFiles = 2
+ }
+ if opts.MaxInputBytes <= 0 {
+ opts.MaxInputBytes = 128 << 20
+ }
+ snap := t.planningSnapshot(meta)
+ if snap == nil {
+ return &PositionDeleteRewriteResult{}, nil
+ }
+ spec, err := meta.CurrentSpec()
+ if err != nil {
+ return nil, err
+ }
+
+ fs, err := t.tbl.FS(ctx)
+ if err != nil {
+ return nil, err
+ }
+ manifests, err := snap.Manifests(fs)
+ if err != nil {
+ return nil, err
+ }
+
+ live := make(map[string]positionDeleteLiveFile)
+ // Build a snapshot-scoped live-file index. This is independent of
+ // removeDanglingDeleteFiles; the two features can merge in either
order.
+ for _, manifest := range manifests {
+ if err := ctx.Err(); err != nil {
+ return nil, err
+ }
+ if manifest.ManifestContent() != iceberg.ManifestContentData {
+ continue
+ }
+ for entry, err := range manifest.Entries(fs, true) {
+ if err != nil {
+ return nil, err
+ }
+ df := entry.DataFile()
+ if df.ContentType() != iceberg.EntryContentData {
+ continue
+ }
+ partition, err := canonicalPartitionKey(df.SpecID(),
dataFilePartition(df))
+ if err != nil {
+ return nil, fmt.Errorf("data file %s: %w",
df.FilePath(), err)
+ }
+ if entry.DataSequenceNumber() < 0 {
+ return nil, fmt.Errorf("%w: data file %s has
unknown sequence number", ErrInvalidMetadata, df.FilePath())
+ }
+ live[df.FilePath()] = positionDeleteLiveFile{partition:
partition, sequence: entry.DataSequenceNumber()}
+ }
+ }
+
+ groups := make(map[positionDeleteRewriteKey]*positionDeleteRewriteGroup)
+ seen := make(map[string]struct{})
+ for _, manifest := range manifests {
+ if err := ctx.Err(); err != nil {
+ return nil, err
+ }
+ if manifest.ManifestContent() != iceberg.ManifestContentDeletes
{
+ continue
+ }
+ for entry, err := range manifest.Entries(fs, true) {
+ if err != nil {
+ return nil, err
+ }
+ df := entry.DataFile()
+ if df.ContentType() != iceberg.EntryContentPosDeletes ||
+ df.FileFormat() != iceberg.ParquetFile ||
+ IsDeletionVector(df) || int(df.SpecID()) !=
spec.ID() ||
+ entry.DataSequenceNumber() < 0 {
+ continue
+ }
+ if _, ok := seen[df.FilePath()]; ok {
+ continue
+ }
+ seen[df.FilePath()] = struct{}{}
+ partition, err := canonicalPartitionKey(df.SpecID(),
dataFilePartition(df))
+ if err != nil {
+ return nil, fmt.Errorf("position delete %s:
%w", df.FilePath(), err)
+ }
+ key := positionDeleteRewriteKey{partition: partition}
+ group := groups[key]
+ if group == nil {
+ group = &positionDeleteRewriteGroup{
+ partition: df.Partition(), sequences:
make(map[string]int64),
+ }
+ groups[key] = group
+ }
+ group.files = append(group.files, df)
+ group.sequences[df.FilePath()] =
entry.DataSequenceNumber()
+ group.maxSequence = max(group.maxSequence,
entry.DataSequenceNumber())
+ group.bytes += df.FileSizeBytes()
+ }
+ }
+
+ // Avoid opening a writer or requiring write-capable I/O when no group
+ // qualifies. Split before writing so the selected work is
deterministic.
+ var jobs []positionDeleteRewriteJob
+ for key, group := range groups {
+ for _, chunk := range splitPositionDeleteRewriteGroup(group,
opts.MaxInputBytes) {
+ if len(chunk.files) < opts.MinInputFiles || chunk.bytes
> opts.MaxInputBytes {
+ continue
+ }
+ jobs = append(jobs, positionDeleteRewriteJob{key: key,
group: chunk})
+ }
+ }
+ if len(jobs) == 0 {
+ return &PositionDeleteRewriteResult{}, nil
+ }
+ slices.SortFunc(jobs, func(a, b positionDeleteRewriteJob) int {
+ if order := strings.Compare(a.key.partition, b.key.partition);
order != 0 {
+ return order
+ }
+
+ return strings.Compare(a.group.files[0].FilePath(),
b.group.files[0].FilePath())
+ })
+
+ wfs, err := requireWriteFileIO(fs)
+ if err != nil {
+ return nil, err
+ }
+
+ if err := t.AssertRefSnapshotID(""); err != nil {
+ return nil, err
+ }
+ rewrite := t.NewRewrite(nil)
+ props := maps.Clone(meta.props)
+ if props == nil {
+ props = make(iceberg.Properties)
+ }
+ props[WriteFormatDefaultKey] = string(iceberg.ParquetFile)
+ writer, err := newPositionDeleteWriter(meta.loc, wfs, meta, props)
+ if err != nil {
+ return nil, err
+ }
+
+ var produced []string
+ staged := false
+ defer func() {
+ if !staged {
+ for _, path := range produced {
+ _ = wfs.Remove(context.WithoutCancel(ctx), path)
+ }
+ }
+ }()
+
+ result := &PositionDeleteRewriteResult{}
+ for _, job := range jobs {
+ if err := ctx.Err(); err != nil {
+ return nil, err
+ }
+ key, group := job.key, job.group
+ supported := true
+ for _, df := range group.files {
+ src, err := tblutils.GetFile(ctx, fs, df, true)
+ if err != nil {
+ return nil, err
+ }
+ rdr, err := src.GetReader(ctx)
+ if err != nil {
+ return nil, err
+ }
+ schema, schemaErr := rdr.Schema()
+ closeErr := rdr.Close()
+ if schemaErr != nil {
+ return nil, schemaErr
+ }
+ if closeErr != nil {
+ return nil, closeErr
+ }
+ // The two-column writer cannot preserve optional
deleted rows.
+ if schema.NumFields() != 2 ||
+ len(schema.FieldIndices("file_path")) != 1 ||
+ len(schema.FieldIndices("pos")) != 1 {
+ supported = false
+
+ break
+ }
+ }
+ if !supported {
+ continue
+ }
+
+ var rows []positionDeleteRow
+ crossSpecTarget := false
+ for _, df := range group.files {
+ byPath, err := readDeletesForPaths(ctx, fs, df, nil)
+ if err != nil {
+ releasePosDeletes(byPath)
+
+ return nil, fmt.Errorf("read position deletes
from %s: %w", df.FilePath(), err)
+ }
+ for path, chunk := range byPath {
+ if !rewritePositionDeletePathInScope(df, path) {
+ continue
+ }
+ data, ok := live[path]
+ if !ok || data.sequence >
group.sequences[df.FilePath()] {
+ continue
+ }
+ if data.partition != key.partition {
+ // A path-scoped delete can target data
written with an
+ // older spec. The replacement may not
retain that scope,
+ // so leave the entire source group
untouched.
+ crossSpecTarget = true
+
+ break
+ }
+ for _, part := range chunk.Chunks() {
+ positions, ok := part.(*array.Int64)
+ if !ok {
+ releasePosDeletes(byPath)
+
+ return nil, fmt.Errorf("%w:
non-int64 position column in %s", iceberg.ErrInvalidSchema, df.FilePath())
+ }
+ if err :=
validateRewritePositionDeletePositions(df.FilePath(), positions); err != nil {
+ releasePosDeletes(byPath)
+
+ return nil, err
+ }
+ for _, pos := range
positions.Int64Values() {
+ rows = append(rows,
positionDeleteRow{path: path, position: pos})
+ }
+ }
+ }
+ releasePosDeletes(byPath)
+ if crossSpecTarget {
+ break
+ }
+ }
+ if crossSpecTarget {
+ continue
+ }
+
+ rows = sortUniquePositionDeleteRows(rows)
+ if len(rows) > 0 {
+ b := array.NewRecordBuilder(compute.GetAllocator(ctx),
PositionalDeleteArrowSchema)
+ pathBuilder := b.Field(0).(*array.StringBuilder)
+ posBuilder := b.Field(1).(*array.Int64Builder)
+ for _, row := range rows {
+ pathBuilder.Append(row.path)
+ posBuilder.Append(row.position)
+ }
+ record := b.NewRecordBatch()
+ b.Release()
+ task := WriteTask{
+ Uuid: uuid.New(), PartitionID: spec.ID(),
+ FileCount: 1, Schema:
iceberg.PositionalDeleteSchema,
+ Batches: []arrow.RecordBatch{record},
+ }
+ // Track the intended output before writing. A failed
writer
+ // may have already created a partial file.
+ produced = append(produced, writer.loc.NewDataLocation(
+
task.GenerateDataFileName(writer.invariants.extension), ""))
+ newFile, err := writer.writeFile(ctx, group.partition,
task)
+ if err != nil {
+ return nil, err
+ }
+ rewrite.AddDeleteFileWithDataSequenceNumber(newFile,
group.maxSequence)
+ result.AddedDeleteFiles++
+ result.BytesAfter += newFile.FileSizeBytes()
+ }
+ for _, df := range group.files {
+ rewrite.DeleteFile(df)
+ result.RemovedDeleteFiles++
+ result.BytesBefore += df.FileSizeBytes()
+ }
+ }
+ if result.RemovedDeleteFiles == 0 {
+ return result, nil
+ }
+ if err := rewrite.Commit(ctx); err != nil {
+ return nil, err
+ }
+
+ // This transaction still needs to commit. Do not delete staged output
+ // files when returning successfully to its caller.
+ staged = true
+
+ return result, nil
+}
+
+func splitPositionDeleteRewriteGroup(group *positionDeleteRewriteGroup, limit
int64) []*positionDeleteRewriteGroup {
+ slices.SortFunc(group.files, func(a, b iceberg.DataFile) int { return
strings.Compare(a.FilePath(), b.FilePath()) })
+ var groups []*positionDeleteRewriteGroup
+ var current *positionDeleteRewriteGroup
+ for _, file := range group.files {
+ size := file.FileSizeBytes()
+ if size < 0 || size > limit {
+ continue
+ }
+ if current == nil || size > limit-current.bytes {
+ current = &positionDeleteRewriteGroup{partition:
group.partition, sequences: group.sequences}
+ groups = append(groups, current)
+ }
+ current.files = append(current.files, file)
+ current.maxSequence = max(current.maxSequence,
group.sequences[file.FilePath()])
+ current.bytes += size
+ }
+
+ return groups
+}
+
+func sortUniquePositionDeleteRows(rows []positionDeleteRow)
[]positionDeleteRow {
+ slices.SortFunc(rows, func(a, b positionDeleteRow) int {
+ if a.path < b.path {
+ return -1
+ }
+ if a.path > b.path {
+ return 1
+ }
+ if a.position < b.position {
+ return -1
+ }
+ if a.position > b.position {
+ return 1
+ }
+
+ return 0
+ })
+
+ return slices.Compact(rows)
+}
+
+// A delete file's statistics can help planning, but do not establish an
+// explicit one-file scope for dropping rows during maintenance.
+func rewritePositionDeletePathInScope(df iceberg.DataFile, path string) bool {
+ target := explicitReferencedDataFile(df)
+
+ return target == "" || target == path
+}
+
+// Reject invalid positions before a rewrite could convert a null to zero or
+// carry a negative index into a replacement delete file.
+func validateRewritePositionDeletePositions(path string, positions
*array.Int64) error {
+ if positions.NullN() != 0 {
+ return fmt.Errorf("%w: null position in delete file %s",
iceberg.ErrInvalidSchema, path)
+ }
+ for _, pos := range positions.Int64Values() {
+ if pos < 0 {
+ return fmt.Errorf("%w: negative position %d in delete
file %s", iceberg.ErrInvalidSchema, pos, path)
+ }
+ }
+
+ return nil
+}
diff --git a/table/rewrite_position_delete_files_integration_test.go
b/table/rewrite_position_delete_files_integration_test.go
new file mode 100644
index 000000000..80bf94ff0
--- /dev/null
+++ b/table/rewrite_position_delete_files_integration_test.go
@@ -0,0 +1,341 @@
+// 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_test
+
+import (
+ "context"
+ "fmt"
+ "testing"
+
+ "github.com/apache/iceberg-go"
+ iceio "github.com/apache/iceberg-go/io"
+ "github.com/apache/iceberg-go/table"
+ "github.com/stretchr/testify/require"
+)
+
+// Embedding only IO hides the writer methods provided by LocalFS.
+type readOnlyPositionDeleteIO struct{ iceio.IO }
+
+func TestRewritePositionDeleteFilesCommit(t *testing.T) {
+ tbl, cat := newConcurrentRewriteTestTable(t)
+ schema, err := table.SchemaToArrowSchema(tbl.Schema(), nil, false,
false)
+ require.NoError(t, err)
+ a, b := tbl.Location()+"/data/a.parquet",
tbl.Location()+"/data/b.parquet"
+ writeParquetFile(t, a, schema,
`[{"id":1,"data":"a"},{"id":2,"data":"b"}]`)
+ writeParquetFile(t, b, schema, `[{"id":3,"data":"c"}]`)
+ tx := tbl.NewTransaction()
+ require.NoError(t, tx.AddDataFilesByPath(t.Context(), []string{a, b},
nil, false))
+ tbl, err = tx.Commit(t.Context())
+ require.NoError(t, err)
+ var deletes []iceberg.DataFile
+ for i := range 2 {
+ path := fmt.Sprintf("%s/data/delete-%d.parquet",
tbl.Location(), i)
+ // The second row is outside the file's referenced-data-file
scope.
+ writeParquetFile(t, path, table.PositionalDeleteArrowSchema,
fmt.Sprintf(`[{"file_path":%q,"pos":0},{"file_path":%q,"pos":0}]`, a, b))
+ builder, err :=
iceberg.NewDataFileBuilder(*iceberg.UnpartitionedSpec,
iceberg.EntryContentPosDeletes, path, iceberg.ParquetFile, nil, nil, nil, 2,
128)
+ require.NoError(t, err)
+ deletes = append(deletes, builder.ReferencedDataFile(a).Build())
+ }
+ tx = tbl.NewTransaction()
+ require.NoError(t,
tx.NewRowDelta(nil).AddDeletes(deletes...).Commit(t.Context()))
+ tbl, err = tx.Commit(t.Context())
+ require.NoError(t, err)
+ sequence := currentManifestEntry(t, tbl,
deletes[0].FilePath()).DataSequenceNumber()
+ require.ElementsMatch(t, []int64{2, 3}, scanIDs(t, tbl))
+
+ // When all inputs exceed the configured budget, the action is a no-op
+ // and must not require a filesystem that can create files.
+ readOnlyFS := func(context.Context) (iceio.IO, error) {
+ return readOnlyPositionDeleteIO{IO: iceio.LocalFS{}}, nil
+ }
+ readOnly := table.New(tbl.Identifier(), tbl.Metadata(),
tbl.MetadataLocation(), readOnlyFS, nil)
+ noOp, err := readOnly.RewritePositionDeleteFiles(t.Context(),
table.PositionDeleteRewriteOptions{
+ MinInputFiles: 1, MaxInputBytes: 1,
+ })
+ require.NoError(t, err)
+ require.Zero(t, noOp.RemovedDeleteFiles)
+ require.Equal(t, tbl.Metadata(), noOp.Table.Metadata())
+
+ result, err := tbl.RewritePositionDeleteFiles(t.Context(),
table.PositionDeleteRewriteOptions{})
+ require.NoError(t, err)
+ updated := result.Table
+ require.Equal(t, 2, result.RemovedDeleteFiles)
+ require.Equal(t, 1, result.AddedDeleteFiles)
+ require.Equal(t, int64(256), result.BytesBefore)
+ require.Positive(t, result.BytesAfter)
+ require.ElementsMatch(t, []int64{2, 3}, scanIDs(t, updated))
+ tasks, err := updated.Scan().PlanFiles(t.Context())
+ require.NoError(t, err)
+ require.Len(t, tasks, 2)
+ var replacement iceberg.DataFile
+ for _, task := range tasks {
+ if task.File.FilePath() == a {
+ require.Len(t, task.DeleteFiles, 1)
+ replacement = task.DeleteFiles[0]
+ }
+ }
+ require.NotNil(t, replacement)
+ require.Equal(t, sequence, currentManifestEntry(t, updated,
replacement.FilePath()).DataSequenceNumber())
+ require.Equal(t, int64(1), replacement.Count())
+
+ // Only one delete file remains, so the default minimum of two prevents
+ // a rewrite even if the table is exposed through read-only I/O.
+ readOnly = table.New(updated.Identifier(), updated.Metadata(),
updated.MetadataLocation(), readOnlyFS, nil)
+ noOp, err = readOnly.RewritePositionDeleteFiles(t.Context(),
table.PositionDeleteRewriteOptions{})
+ require.NoError(t, err)
+ require.Zero(t, noOp.RemovedDeleteFiles)
+ require.Equal(t, updated.Metadata(), noOp.Table.Metadata())
+
+ attemptsBefore := cat.attempts.Load()
+ _, err = tbl.RewritePositionDeleteFiles(t.Context(),
table.PositionDeleteRewriteOptions{})
+ require.ErrorIs(t, err, table.ErrCommitFailed)
+ require.Equal(t, attemptsBefore+1, cat.attempts.Load(),
+ "the pinned planning snapshot must stop a stale delete rewrite
without retrying")
+}
+
+func TestRewritePositionDeleteFilesDropsDanglingRowsFromUnscopedFiles(t
*testing.T) {
+ tbl, _ := newConcurrentRewriteTestTable(t)
+ schema, err := table.SchemaToArrowSchema(tbl.Schema(), nil, false,
false)
+ require.NoError(t, err)
+ a := tbl.Location() + "/data/a.parquet"
+ b := tbl.Location() + "/data/b.parquet"
+ dangling := tbl.Location() + "/data/removed.parquet"
+ writeParquetFile(t, a, schema,
`[{"id":1,"data":"a"},{"id":2,"data":"b"}]`)
+ writeParquetFile(t, b, schema,
`[{"id":3,"data":"c"},{"id":4,"data":"d"}]`)
+
+ tx := tbl.NewTransaction()
+ require.NoError(t, tx.AddDataFilesByPath(t.Context(), []string{a, b},
nil, false))
+ tbl, err = tx.Commit(t.Context())
+ require.NoError(t, err)
+
+ var deletes []iceberg.DataFile
+ for i := range 2 {
+ path := fmt.Sprintf("%s/data/unscoped-delete-%d.parquet",
tbl.Location(), i)
+ writeParquetFile(t, path, table.PositionalDeleteArrowSchema,
fmt.Sprintf(
+
`[{"file_path":%q,"pos":1},{"file_path":%q,"pos":0},{"file_path":%q,"pos":0}]`,
a, b, dangling))
+ builder, err :=
iceberg.NewDataFileBuilder(*iceberg.UnpartitionedSpec,
+ iceberg.EntryContentPosDeletes, path,
iceberg.ParquetFile, nil, nil, nil, 3, 128)
+ require.NoError(t, err)
+ deletes = append(deletes, builder.Build())
+ }
+
+ tx = tbl.NewTransaction()
+ require.NoError(t,
tx.NewRowDelta(nil).AddDeletes(deletes...).Commit(t.Context()))
+ tbl, err = tx.Commit(t.Context())
+ require.NoError(t, err)
+ require.ElementsMatch(t, []int64{1, 4}, scanIDs(t, tbl))
+
+ result, err := tbl.RewritePositionDeleteFiles(t.Context(),
table.PositionDeleteRewriteOptions{})
+ require.NoError(t, err)
+ updated := result.Table
+ require.Equal(t, 2, result.RemovedDeleteFiles)
+ require.Equal(t, 1, result.AddedDeleteFiles)
+ require.ElementsMatch(t, []int64{1, 4}, scanIDs(t, updated))
+
+ tasks, err := updated.Scan().PlanFiles(t.Context())
+ require.NoError(t, err)
+ require.Len(t, tasks, 2)
+ var replacement iceberg.DataFile
+ for _, task := range tasks {
+ require.Len(t, task.DeleteFiles, 1)
+ if replacement == nil {
+ replacement = task.DeleteFiles[0]
+ }
+ require.Equal(t, replacement.FilePath(),
task.DeleteFiles[0].FilePath())
+ }
+ require.Nil(t, replacement.ReferencedDataFile())
+ require.Equal(t, int64(2), replacement.Count(), "drop dangling and
duplicate delete positions")
+
+ again, err := updated.RewritePositionDeleteFiles(t.Context(),
table.PositionDeleteRewriteOptions{})
+ require.NoError(t, err)
+ require.Zero(t, again.RemovedDeleteFiles)
+ require.Equal(t, updated.Metadata(), again.Table.Metadata())
+}
+
+func TestRewritePositionDeleteFilesAcrossSequences(t *testing.T) {
+ ctx := t.Context()
+ tbl, _ := newConcurrentRewriteTestTable(t)
+ schema, err := table.SchemaToArrowSchema(tbl.Schema(), nil, false,
false)
+ require.NoError(t, err)
+ a, b := tbl.Location()+"/data/a.parquet",
tbl.Location()+"/data/b.parquet"
+ writeParquetFile(t, a, schema,
`[{"id":1,"data":"a"},{"id":2,"data":"b"}]`)
+ writeParquetFile(t, b, schema,
`[{"id":3,"data":"c"},{"id":4,"data":"d"}]`)
+
+ tx := tbl.NewTransaction()
+ require.NoError(t, tx.AddDataFilesByPath(ctx, []string{a}, nil, false))
+ tbl, err = tx.Commit(ctx)
+ require.NoError(t, err)
+
+ // The first delete predates file b. Its row targeting b must not
+ // become applicable after the rewrite raises the replacement sequence.
+ olderPath := tbl.Location() + "/data/older-delete.parquet"
+ writeParquetFile(t, olderPath, table.PositionalDeleteArrowSchema,
fmt.Sprintf(
+ `[{"file_path":%q,"pos":0},{"file_path":%q,"pos":0}]`, a, b))
+ olderBuilder, err :=
iceberg.NewDataFileBuilder(*iceberg.UnpartitionedSpec,
+ iceberg.EntryContentPosDeletes, olderPath, iceberg.ParquetFile,
nil, nil, nil, 2, 128)
+ require.NoError(t, err)
+ tx = tbl.NewTransaction()
+ require.NoError(t,
tx.NewRowDelta(nil).AddDeletes(olderBuilder.Build()).Commit(ctx))
+ tbl, err = tx.Commit(ctx)
+ require.NoError(t, err)
+
+ tx = tbl.NewTransaction()
+ require.NoError(t, tx.AddDataFilesByPath(ctx, []string{b}, nil, false))
+ tbl, err = tx.Commit(ctx)
+ require.NoError(t, err)
+
+ newerPath := tbl.Location() + "/data/newer-delete.parquet"
+ writeParquetFile(t, newerPath, table.PositionalDeleteArrowSchema,
+ fmt.Sprintf(`[{"file_path":%q,"pos":1}]`, a))
+ newerBuilder, err :=
iceberg.NewDataFileBuilder(*iceberg.UnpartitionedSpec,
+ iceberg.EntryContentPosDeletes, newerPath, iceberg.ParquetFile,
nil, nil, nil, 1, 128)
+ require.NoError(t, err)
+ newer := newerBuilder.Build()
+ tx = tbl.NewTransaction()
+ require.NoError(t, tx.NewRowDelta(nil).AddDeletes(newer).Commit(ctx))
+ tbl, err = tx.Commit(ctx)
+ require.NoError(t, err)
+
+ maxSourceSequence := currentManifestEntry(t, tbl,
newer.FilePath()).DataSequenceNumber()
+ require.ElementsMatch(t, []int64{3, 4}, scanIDs(t, tbl))
+
+ result, err := tbl.RewritePositionDeleteFiles(ctx,
table.PositionDeleteRewriteOptions{})
+ require.NoError(t, err)
+ updated := result.Table
+ require.Equal(t, 2, result.RemovedDeleteFiles, "delete files from
separate commits should compact")
+ require.Equal(t, 1, result.AddedDeleteFiles)
+ require.ElementsMatch(t, []int64{3, 4}, scanIDs(t, updated))
+
+ tasks, err := updated.Scan().PlanFiles(ctx)
+ require.NoError(t, err)
+ var replacement iceberg.DataFile
+ for _, task := range tasks {
+ if task.File.FilePath() == a {
+ require.Len(t, task.DeleteFiles, 1)
+ replacement = task.DeleteFiles[0]
+ }
+ }
+ require.NotNil(t, replacement)
+ require.Equal(t, maxSourceSequence,
+ currentManifestEntry(t, updated,
replacement.FilePath()).DataSequenceNumber())
+ require.Equal(t, int64(2), replacement.Count(),
+ "stale positions must not gain applicability from the
replacement sequence")
+}
+
+func TestRewritePositionDeleteFilesPreservesCrossSpecPathScopedDeletes(t
*testing.T) {
+ ctx := t.Context()
+ tbl, _ := newConcurrentRewriteTestTable(t)
+ schema, err := table.SchemaToArrowSchema(tbl.Schema(), nil, false,
false)
+ require.NoError(t, err)
+
+ dataPath := tbl.Location() + "/data/old-spec.parquet"
+ writeParquetFile(t, dataPath, schema,
`[{"id":1,"data":"a"},{"id":2,"data":"b"}]`)
+ tx := tbl.NewTransaction()
+ require.NoError(t, tx.AddDataFilesByPath(ctx, []string{dataPath}, nil,
false))
+ 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.
+ 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)
+ require.NoError(t, buildErr)
+ deletes = append(deletes,
builder.ReferencedDataFile(dataPath).Build())
+ }
+ tx = tbl.NewTransaction()
+ require.NoError(t,
tx.NewRowDelta(nil).AddDeletes(deletes...).Commit(ctx))
+ tbl, err = tx.Commit(ctx)
+ require.NoError(t, err)
+ require.ElementsMatch(t, []int64{2}, scanIDs(t, tbl))
+
+ // Combining these sources can lose their path-based scope because the
+ // output may be partition-scoped. Keep the original files instead.
+ result, err := tbl.RewritePositionDeleteFiles(ctx,
table.PositionDeleteRewriteOptions{})
+ require.NoError(t, err)
+ require.Zero(t, result.RemovedDeleteFiles)
+ require.Zero(t, result.AddedDeleteFiles)
+ require.Equal(t, tbl.Metadata(), result.Table.Metadata())
+ require.ElementsMatch(t, []int64{2}, scanIDs(t, result.Table))
+}
+
+func TestRewritePositionDeleteFilesStagesOnTargetBranch(t *testing.T) {
+ ctx := t.Context()
+ tbl, cat := newConcurrentRewriteTestTable(t)
+ schema, err := table.SchemaToArrowSchema(tbl.Schema(), nil, false,
false)
+ require.NoError(t, err)
+
+ dataPath := tbl.Location() + "/data/branch-data.parquet"
+ writeParquetFile(t, dataPath, schema,
"[{\"id\":1,\"data\":\"a\"},{\"id\":2,\"data\":\"b\"}]")
+ txn := tbl.NewTransaction()
+ require.NoError(t, txn.AddDataFilesByPath(ctx, []string{dataPath}, nil,
false))
+ tbl, err = txn.Commit(ctx)
+ require.NoError(t, err)
+
+ var deletes []iceberg.DataFile
+ for i := range 2 {
+ path := fmt.Sprintf("%s/data/branch-delete-%d.parquet",
tbl.Location(), i)
+ writeParquetFile(t, path, table.PositionalDeleteArrowSchema,
+ fmt.Sprintf("[{\"file_path\":%q,\"pos\":0}]", dataPath))
+ builder, buildErr :=
iceberg.NewDataFileBuilder(*iceberg.UnpartitionedSpec,
+ iceberg.EntryContentPosDeletes, path,
iceberg.ParquetFile, nil, nil, nil, 1, 128)
+ require.NoError(t, buildErr)
+ deletes = append(deletes, builder.Build())
+ }
+ txn, err = tbl.NewTransactionOnBranch("feature")
+ require.NoError(t, err)
+ require.NoError(t,
txn.NewRowDelta(nil).AddDeletes(deletes...).Commit(ctx))
+ tbl, err = txn.Commit(ctx)
+ require.NoError(t, err)
+
+ require.ElementsMatch(t, []int64{1, 2}, idsOnRef(t, tbl,
table.MainBranch))
+ require.ElementsMatch(t, []int64{2}, idsOnRef(t, tbl, "feature"))
+
+ txn, err = tbl.NewTransactionOnBranch("feature")
+ require.NoError(t, err)
+ commitsBefore := cat.attempts.Load()
+ result, err := txn.RewritePositionDeleteFiles(ctx,
table.PositionDeleteRewriteOptions{})
+ require.NoError(t, err)
+ require.Nil(t, result.Table, "transaction rewrite must not commit
itself")
+ require.Equal(t, 2, result.RemovedDeleteFiles)
+ require.Equal(t, 1, result.AddedDeleteFiles)
+ require.Equal(t, int64(256), result.BytesBefore)
+ require.Positive(t, result.BytesAfter)
+ require.Equal(t, commitsBefore, cat.attempts.Load(), "staging must not
contact the catalog")
+
+ require.NoError(t,
txn.SetProperties(iceberg.Properties{"test.position-delete-rewrite": "staged"}))
+ updated, err := txn.Commit(ctx)
+ require.NoError(t, err)
+ require.Equal(t, commitsBefore+1, cat.attempts.Load())
+ require.Equal(t, "staged",
updated.Properties()["test.position-delete-rewrite"])
+ require.ElementsMatch(t, []int64{1, 2}, idsOnRef(t, updated,
table.MainBranch))
+ require.ElementsMatch(t, []int64{2}, idsOnRef(t, updated, "feature"))
+}
diff --git a/table/rewrite_position_delete_files_test.go
b/table/rewrite_position_delete_files_test.go
new file mode 100644
index 000000000..de9c38823
--- /dev/null
+++ b/table/rewrite_position_delete_files_test.go
@@ -0,0 +1,141 @@
+// 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 (
+ "testing"
+
+ "github.com/apache/arrow-go/v18/arrow/array"
+ "github.com/apache/arrow-go/v18/arrow/memory"
+ "github.com/apache/iceberg-go"
+ "github.com/stretchr/testify/require"
+)
+
+func TestSortUniquePositionDeleteRows(t *testing.T) {
+ rows := []positionDeleteRow{
+ {path: "b.parquet", position: 4},
+ {path: "a.parquet", position: 10},
+ {path: "a.parquet", position: 1},
+ {path: "b.parquet", position: 4},
+ {path: "a.parquet", position: 1},
+ {path: "b.parquet", position: 2},
+ }
+ got := sortUniquePositionDeleteRows(rows)
+ require.Equal(t, []positionDeleteRow{
+ {path: "a.parquet", position: 1},
+ {path: "a.parquet", position: 10},
+ {path: "b.parquet", position: 2},
+ {path: "b.parquet", position: 4},
+ }, got)
+}
+
+func TestSplitPositionDeleteRewriteGroup(t *testing.T) {
+ var files []iceberg.DataFile
+ for _, path := range []string{"d", "b", "a", "c"} {
+ b, err :=
iceberg.NewDataFileBuilder(*iceberg.UnpartitionedSpec,
iceberg.EntryContentPosDeletes,
+ path, iceberg.ParquetFile, nil, nil, nil, 1, 10)
+ require.NoError(t, err)
+ files = append(files, b.Build())
+ }
+ sequences := map[string]int64{"a": 2, "b": 9, "c": 5, "d": 7}
+ group := &positionDeleteRewriteGroup{files: files, sequences: sequences}
+ groups := splitPositionDeleteRewriteGroup(group, 20)
+ require.Len(t, groups, 2)
+ require.Equal(t, "a", groups[0].files[0].FilePath())
+ require.Equal(t, "b", groups[0].files[1].FilePath())
+ require.Equal(t, int64(9), groups[0].maxSequence)
+ require.Equal(t, int64(2), groups[0].sequences["a"])
+ require.Equal(t, "c", groups[1].files[0].FilePath())
+ require.Equal(t, "d", groups[1].files[1].FilePath())
+ require.Equal(t, int64(7), groups[1].maxSequence)
+ require.Equal(t, int64(5), groups[1].sequences["c"])
+ require.Empty(t, splitPositionDeleteRewriteGroup(group, 9))
+}
+
+func TestRewritePositionDeleteFilesEmptySnapshot(t *testing.T) {
+ schema := iceberg.NewSchema(0,
+ iceberg.NestedField{ID: 1, Name: "id", Type:
iceberg.PrimitiveTypes.Int64, Required: true},
+ )
+ meta, err := NewMetadata(schema, iceberg.UnpartitionedSpec,
UnsortedSortOrder, t.TempDir(), nil)
+ require.NoError(t, err)
+ tbl := New([]string{"db", "example"}, meta, "", nil, nil)
+ result, err := tbl.RewritePositionDeleteFiles(t.Context(),
PositionDeleteRewriteOptions{})
+ require.NoError(t, err)
+ require.Zero(t, result.RemovedDeleteFiles)
+ require.Zero(t, result.AddedDeleteFiles)
+ require.Equal(t, tbl.Metadata(), result.Table.Metadata())
+
+ txn := tbl.NewTransaction()
+ staged, err := txn.RewritePositionDeleteFiles(t.Context(),
PositionDeleteRewriteOptions{})
+ require.NoError(t, err)
+ require.Zero(t, staged.RemovedDeleteFiles)
+ require.Nil(t, staged.Table)
+}
+
+func TestRewritePositionDeleteFilesRejectsV3(t *testing.T) {
+ schema := iceberg.NewSchema(0,
+ iceberg.NestedField{ID: 1, Name: "id", Type:
iceberg.PrimitiveTypes.Int64, Required: true},
+ )
+ meta, err := NewMetadata(schema, iceberg.UnpartitionedSpec,
UnsortedSortOrder, t.TempDir(),
+ iceberg.Properties{PropertyFormatVersion: "3"})
+ require.NoError(t, err)
+ tbl := New([]string{"db", "example"}, meta, "", nil, nil)
+ _, err = tbl.RewritePositionDeleteFiles(t.Context(),
PositionDeleteRewriteOptions{})
+ require.ErrorIs(t, err, iceberg.ErrNotImplemented)
+}
+
+func TestRewritePositionDeletePathInScope(t *testing.T) {
+ builder, err := iceberg.NewDataFileBuilder(*iceberg.UnpartitionedSpec,
+ iceberg.EntryContentPosDeletes, "delete.parquet",
iceberg.ParquetFile,
+ nil, nil, nil, 2, 128)
+ require.NoError(t, err)
+ builder.LowerBoundValues(map[int][]byte{filePathFieldID:
[]byte("old-metrics-path")})
+ builder.UpperBoundValues(map[int][]byte{filePathFieldID:
[]byte("old-metrics-path")})
+ require.True(t, rewritePositionDeletePathInScope(builder.Build(),
"live.parquet"),
+ "file_path statistics do not define the referenced_data_file
scope")
+
+ builder.ReferencedDataFile("target.parquet")
+ require.True(t, rewritePositionDeletePathInScope(builder.Build(),
"target.parquet"))
+ require.False(t, rewritePositionDeletePathInScope(builder.Build(),
"other.parquet"))
+}
+
+func TestValidateRewritePositionDeletePositions(t *testing.T) {
+ for _, tc := range []struct {
+ name string
+ values []int64
+ valid []bool
+ invalid bool
+ }{
+ {name: "valid", values: []int64{0, 42}},
+ {name: "negative", values: []int64{-1}, invalid: true},
+ {name: "null", values: []int64{0}, valid: []bool{false},
invalid: true},
+ } {
+ t.Run(tc.name, func(t *testing.T) {
+ builder :=
array.NewInt64Builder(memory.DefaultAllocator)
+ builder.AppendValues(tc.values, tc.valid)
+ positions := builder.NewInt64Array()
+ builder.Release()
+ err :=
validateRewritePositionDeletePositions("delete.parquet", positions)
+ positions.Release()
+ if tc.invalid {
+ require.ErrorIs(t, err,
iceberg.ErrInvalidSchema)
+ } else {
+ require.NoError(t, err)
+ }
+ })
+ }
+}
