You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iceberg.apache.org by ze...@apache.org on 2026/08/17 21:12:27 UTC
(iceberg-go) branch main updated: feat(table): allow RewriteFiles to add delete files (#1677)
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 f63550438 feat(table): allow RewriteFiles to add delete files (#1677)
f63550438 is described below
commit f63550438b966b7c49cd9ed91a6bdfe42b9bf837
Author: Minh Vu <vu...@gmail.com>
AuthorDate: Mon Aug 17 23:12:22 2026 +0200
feat(table): allow RewriteFiles to add delete files (#1677)
* feat(table): allow rewrites to add delete files
* fix(table): preserve delete rewrite sequence semantics
* fix(table): keep delete rewrite builder compatibility
* fix(table): keep automatic rewrite deletes separate
* fix(table): validate deletion vector rewrites
* fix(table): preserve rewrite delete applicability
* fix(table): validate rewrite sequence and delete conflicts
* fix(table): satisfy rewrite files lint
* fix(table): address rewrite review cleanup
---
table/dv_rewrite_test.go | 126 ++++++++++
table/replace_files_test.go | 552 +++++++++++++++++++++++++++++++++++++++++++-
table/rewrite_data_files.go | 15 ++
table/rewrite_files.go | 202 +++++++++++++---
table/rewrite_files_test.go | 168 +++++++++++++-
table/snapshot_producers.go | 113 +++++++--
table/transaction.go | 542 ++++++++++++++++++++++++++++++++++++++++++-
7 files changed, 1655 insertions(+), 63 deletions(-)
diff --git a/table/dv_rewrite_test.go b/table/dv_rewrite_test.go
index 2e7aca4d8..ad9d4f54a 100644
--- a/table/dv_rewrite_test.go
+++ b/table/dv_rewrite_test.go
@@ -19,6 +19,7 @@ package table_test
import (
"context"
+ "fmt"
"path/filepath"
"testing"
@@ -174,6 +175,131 @@ func TestRewriteFilesApplyResultRemovesDeletionVectors(t *testing.T) {
"the coordinator path must preserve survivors' _row_id")
}
+func TestRewriteFiles_ExplicitDeleteRewriteIgnoresAutomaticDVRemoval(t *testing.T) {
+ ctx := context.Background()
+ tbl, dvTarget := seedV3TableWithDV(t)
+ tbl = appendEqualityDelete(t, tbl, []int{1}, `[{"id": 99}]`)
+
+ var oldEquality iceberg.DataFile
+ var oldEqualitySequence int64
+ manifests, err := tbl.CurrentSnapshot().Manifests(iceio.LocalFS{})
+ require.NoError(t, err)
+ for _, manifest := range manifests {
+ if manifest.ManifestContent() != iceberg.ManifestContentDeletes {
+ continue
+ }
+ for entry, err := range manifest.Entries(iceio.LocalFS{}, false) {
+ require.NoError(t, err)
+ if entry.Status() == iceberg.EntryStatusDELETED || entry.DataFile().ContentType() != iceberg.EntryContentEqDeletes {
+ continue
+ }
+ oldEquality = entry.DataFile()
+ oldEqualitySequence = entry.SequenceNum()
+ }
+ }
+ require.NotNil(t, oldEquality)
+
+ groups, rewritten := planDVCompaction(t, tbl, dvTarget)
+ tx := tbl.NewTransaction()
+ rewrite := tx.NewRewrite(nil)
+ for _, group := range groups {
+ result, err := table.ExecuteCompactionGroup(ctx, tbl, group)
+ require.NoError(t, err)
+ rewrite.ApplyResult(result)
+ }
+
+ newEqualityPath := tbl.Location() + "/data/replacement-equality-delete.parquet"
+ newEquality := newEqDeleteFile(t, newEqualityPath)
+ rewrite.DeleteFile(oldEquality).
+ AddDeleteFileWithDataSequenceNumber(newEquality, oldEqualitySequence)
+ require.NoError(t, rewrite.Commit(ctx),
+ "automatic DV cleanup must not be treated as part of the explicit equality-delete rewrite")
+ tbl, err = tx.Commit(ctx)
+ require.NoError(t, err)
+
+ assert.Equal(t, oldEqualitySequence, deleteFileSequence(t, tbl, newEqualityPath))
+ assert.Empty(t, deleteEntriesReferencing(t, tbl, rewritten))
+ assertRowCount(t, tbl, 3)
+}
+
+func TestRewriteFiles_DataSequenceNumberKeepsDVApplicableToReplacement(t *testing.T) {
+ tbl := newReplaceFilesTestTable(t)
+ arrowSc, err := table.SchemaToArrowSchema(tbl.Schema(), nil, false, false)
+ require.NoError(t, err)
+
+ oldDataPath := tbl.Location() + "/data/old-data.parquet"
+ writeParquetFile(t, oldDataPath, arrowSc, `[{"id":1,"data":"a"},{"id":2,"data":"b"}]`)
+ tx := tbl.NewTransaction()
+ require.NoError(t, tx.AddFiles(t.Context(), []string{oldDataPath}, nil, false))
+ tbl, err = tx.Commit(t.Context())
+ require.NoError(t, err)
+
+ tasks, err := tbl.Scan().PlanFiles(t.Context())
+ require.NoError(t, err)
+ require.Len(t, tasks, 1)
+ oldData := tasks[0].File
+ oldDataSequence := fileDataSequence(t, tbl, oldDataPath)
+
+ oldDeletePath := tbl.Location() + "/data/old-pos-delete.parquet"
+ writeParquetFile(t, oldDeletePath, table.PositionalDeleteArrowSchema,
+ fmt.Sprintf(`[{"file_path":%q,"pos":0}]`, oldDataPath))
+ oldDeleteBuilder, err := iceberg.NewDataFileBuilder(
+ *iceberg.UnpartitionedSpec, iceberg.EntryContentPosDeletes,
+ oldDeletePath, iceberg.ParquetFile, nil, nil, nil, 1, 128)
+ require.NoError(t, err)
+ oldDelete := oldDeleteBuilder.Build()
+ tx = tbl.NewTransaction()
+ require.NoError(t, tx.NewRowDelta(nil).AddDeletes(oldDelete).Commit(t.Context()))
+ tbl, err = tx.Commit(t.Context())
+ require.NoError(t, err)
+ oldDeleteSequence := fileDataSequence(t, tbl, oldDeletePath)
+
+ tx = tbl.NewTransaction()
+ require.NoError(t, tx.UpgradeFormatVersion(3))
+ tbl, err = tx.Commit(t.Context())
+ require.NoError(t, err)
+
+ newDataPath := tbl.Location() + "/data/replacement-data.parquet"
+ writeParquetFile(t, newDataPath, arrowSc, `[{"id":1,"data":"a"},{"id":2,"data":"b"}]`)
+ newDataBuilder, err := iceberg.NewDataFileBuilder(
+ *iceberg.UnpartitionedSpec, iceberg.EntryContentData,
+ newDataPath, iceberg.ParquetFile, nil, nil, nil, 2, 256)
+ require.NoError(t, err)
+ newData := newDataBuilder.FirstRowID(0).Build()
+
+ writer := dv.NewDVWriter(iceio.LocalFS{}, unpartitionedSpecByID)
+ require.NoError(t, writer.Add(newDataPath, []int64{0}, 0, nil))
+ newDVs, err := writer.Flush(t.Context(), tbl.Location()+"/data/replacement-dv.puffin")
+ require.NoError(t, err)
+ require.Len(t, newDVs, 1)
+
+ tx = tbl.NewTransaction()
+ rewrite := tx.NewRewrite(nil).
+ DeleteFile(oldData).
+ DeleteFile(oldDelete).
+ AddDataFile(newData).
+ AddDeleteFileWithDataSequenceNumber(newDVs[0], oldDeleteSequence)
+ err = rewrite.Commit(t.Context())
+ require.Error(t, err)
+ assert.Contains(t, err.Error(), "does not apply to referenced data file")
+
+ tx = tbl.NewTransaction()
+ rewrite = tx.NewRewrite(nil).
+ DeleteFile(oldData).
+ DeleteFile(oldDelete).
+ AddDataFile(newData).
+ AddDeleteFileWithDataSequenceNumber(newDVs[0], oldDeleteSequence).
+ DataSequenceNumber(oldDataSequence)
+ require.NoError(t, rewrite.Commit(t.Context()))
+ tbl, err = tx.Commit(t.Context())
+ require.NoError(t, err)
+
+ assert.Equal(t, oldDataSequence, fileDataSequence(t, tbl, newDataPath))
+ assert.Equal(t, oldDeleteSequence, fileDataSequence(t, tbl, newDVs[0].FilePath()))
+ assert.Equal(t, []int64{2}, scanIDs(t, tbl),
+ "the deletion vector must remain applicable to the replacement data file")
+}
+
// TestRewriteDataFilesPreservesSiblingDeletionVector pins the granularity of DV
// removal: one Puffin file holds DV blobs for two data files (a shared path),
// but only one of those data files is rewritten. The rewritten file's DV must
diff --git a/table/replace_files_test.go b/table/replace_files_test.go
index 7c4b41d37..0313ec0bd 100644
--- a/table/replace_files_test.go
+++ b/table/replace_files_test.go
@@ -21,16 +21,23 @@ import (
"context"
"fmt"
"path/filepath"
+ "strconv"
"testing"
+ "github.com/apache/arrow-go/v18/arrow/array"
"github.com/apache/iceberg-go"
iceio "github.com/apache/iceberg-go/io"
"github.com/apache/iceberg-go/table"
+ "github.com/apache/iceberg-go/table/dv"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
func newReplaceFilesTestTable(t *testing.T) *table.Table {
+ return newReplaceFilesTestTableVersion(t, 2)
+}
+
+func newReplaceFilesTestTableVersion(t *testing.T, version int) *table.Table {
t.Helper()
location := filepath.ToSlash(t.TempDir())
@@ -42,7 +49,7 @@ func newReplaceFilesTestTable(t *testing.T) *table.Table {
meta, err := table.NewMetadata(schema, iceberg.UnpartitionedSpec,
table.UnsortedSortOrder, location,
- iceberg.Properties{table.PropertyFormatVersion: "2"})
+ iceberg.Properties{table.PropertyFormatVersion: strconv.Itoa(version)})
require.NoError(t, err)
return table.New(
@@ -55,6 +62,48 @@ func newReplaceFilesTestTable(t *testing.T) *table.Table {
)
}
+func newPartitionedReplaceFilesTestTable(t *testing.T) (*table.Table, iceberg.PartitionSpec) {
+ t.Helper()
+
+ location := filepath.ToSlash(t.TempDir())
+ schema := iceberg.NewSchema(0,
+ iceberg.NestedField{ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int64, Required: true},
+ )
+ spec := iceberg.NewPartitionSpec(iceberg.PartitionField{
+ SourceIDs: []int{1}, FieldID: 1000, Name: "id", Transform: iceberg.IdentityTransform{},
+ })
+ meta, err := table.NewMetadata(schema, &spec, table.UnsortedSortOrder, location,
+ iceberg.Properties{table.PropertyFormatVersion: "2"})
+ require.NoError(t, err)
+
+ return table.New(
+ table.Identifier{"db", "partitioned_replace_files_test"},
+ meta, location+"/metadata/v1.metadata.json",
+ func(context.Context) (iceio.IO, error) { return iceio.LocalFS{}, nil },
+ &rowDeltaCatalog{metadata: meta},
+ ), spec
+}
+
+func newRewriteDeletionVector(t *testing.T, path, ref string, offset, length *int64) iceberg.DataFile {
+ t.Helper()
+
+ builder, err := iceberg.NewDataFileBuilder(
+ *iceberg.UnpartitionedSpec, iceberg.EntryContentPosDeletes,
+ path, iceberg.PuffinFile, nil, nil, nil, 1, 128)
+ require.NoError(t, err)
+ if ref != "" {
+ builder.ReferencedDataFile(ref)
+ }
+ if offset != nil {
+ builder.ContentOffset(*offset)
+ }
+ if length != nil {
+ builder.ContentSizeInBytes(*length)
+ }
+
+ return builder.Build()
+}
+
func TestReplaceFiles_DataAndDeleteFiles(t *testing.T) {
tbl := newReplaceFilesTestTable(t)
@@ -139,6 +188,507 @@ func TestReplaceFiles_DataAndDeleteFiles(t *testing.T) {
assert.Equal(t, table.OpOverwrite, snap.Summary.Operation)
}
+func TestReplaceFilesWithDeleteFilesPreservesDataSequenceNumber(t *testing.T) {
+ tbl := newReplaceFilesTestTable(t)
+ arrowSc, err := table.SchemaToArrowSchema(tbl.Schema(), nil, false, false)
+ require.NoError(t, err)
+
+ dataPath := tbl.Location() + "/data/data.parquet"
+ writeParquetFile(t, dataPath, arrowSc, `[{"id":1,"data":"a"},{"id":2,"data":"b"}]`)
+ tx := tbl.NewTransaction()
+ require.NoError(t, tx.AddFiles(t.Context(), []string{dataPath}, nil, false))
+ tbl, err = tx.Commit(t.Context())
+ require.NoError(t, err)
+
+ oldDeletePath := tbl.Location() + "/data/old-pos-delete.parquet"
+ writeParquetFile(t, oldDeletePath, table.PositionalDeleteArrowSchema,
+ fmt.Sprintf(`[{"file_path":%q,"pos":0}]`, dataPath))
+ oldDeleteBuilder, err := iceberg.NewDataFileBuilder(
+ *iceberg.UnpartitionedSpec, iceberg.EntryContentPosDeletes,
+ oldDeletePath, iceberg.ParquetFile, nil, nil, nil, 1, 128)
+ require.NoError(t, err)
+ tx = tbl.NewTransaction()
+ require.NoError(t, tx.NewRowDelta(nil).AddDeletes(oldDeleteBuilder.Build()).Commit(t.Context()))
+ tbl, err = tx.Commit(t.Context())
+ require.NoError(t, err)
+
+ oldDelete := oldDeleteBuilder.Build()
+ oldSequence := deleteFileSequence(t, tbl, oldDelete.FilePath())
+ newDeletePath := tbl.Location() + "/data/new-pos-delete.parquet"
+ writeParquetFile(t, newDeletePath, table.PositionalDeleteArrowSchema,
+ fmt.Sprintf(`[{"file_path":%q,"pos":0}]`, dataPath))
+ newDeleteBuilder, err := iceberg.NewDataFileBuilder(
+ *iceberg.UnpartitionedSpec, iceberg.EntryContentPosDeletes,
+ newDeletePath, iceberg.ParquetFile, nil, nil, nil, 1, 128)
+ require.NoError(t, err)
+
+ tx = tbl.NewTransaction()
+ require.NoError(t, tx.ReplaceFilesWithDeleteFiles(t.Context(), nil, nil,
+ []iceberg.DataFile{oldDelete},
+ []table.DeleteFileAddition{{
+ File: newDeleteBuilder.Build(),
+ DataSequenceNumber: oldSequence,
+ }}, nil))
+ tbl, err = tx.Commit(t.Context())
+ require.NoError(t, err)
+
+ assert.Equal(t, oldSequence, deleteFileSequence(t, tbl, newDeletePath),
+ "rewritten delete files must retain the replaced data sequence number")
+ assert.NotEqual(t, tbl.CurrentSnapshot().SequenceNumber, oldSequence,
+ "the new snapshot sequence must not replace the delete data sequence")
+ assert.Equal(t, []int64{2}, scanIDs(t, tbl),
+ "the replacement must preserve delete applicability")
+}
+
+func TestReplaceFilesWithDeleteFilesRejectsDVOnV2(t *testing.T) {
+ tbl := newReplaceFilesTestTable(t)
+ ref := tbl.Location() + "/data/data.parquet"
+ dvBuilder, err := iceberg.NewDataFileBuilder(
+ *iceberg.UnpartitionedSpec, iceberg.EntryContentPosDeletes,
+ tbl.Location()+"/data/delete-vector.puffin", iceberg.PuffinFile,
+ nil, nil, nil, 1, 1)
+ require.NoError(t, err)
+ dv := dvBuilder.ReferencedDataFile(ref).Build()
+ oldDelete := newPosDeleteFile(t, tbl.Location()+"/data/old-pos-delete.parquet")
+
+ tx := tbl.NewTransaction()
+ err = tx.ReplaceFilesWithDeleteFiles(t.Context(), nil, nil,
+ []iceberg.DataFile{oldDelete},
+ []table.DeleteFileAddition{{File: dv, DataSequenceNumber: 0}}, nil)
+ require.Error(t, err)
+ assert.Contains(t, err.Error(), "requires table format version >= 3")
+}
+
+func TestReplaceFilesWithDeleteFilesRejectsPositionDeleteOnV3(t *testing.T) {
+ tbl := newReplaceFilesTestTableVersion(t, 3)
+ tx := tbl.NewTransaction()
+ err := tx.ReplaceFilesWithDeleteFiles(t.Context(), nil, nil,
+ []iceberg.DataFile{newPosDeleteFile(t, "old-pos-delete.parquet")},
+ []table.DeleteFileAddition{{
+ File: newPosDeleteFile(t, "new-pos-delete.parquet"),
+ DataSequenceNumber: 0,
+ }}, nil)
+ require.Error(t, err)
+ assert.Contains(t, err.Error(), "must be a deletion vector for v3 table")
+}
+
+func TestReplaceFilesWithDeleteFilesValidatesDeletionVectorMetadata(t *testing.T) {
+ tbl := newReplaceFilesTestTableVersion(t, 3)
+ oldDelete := newPosDeleteFile(t, "old-pos-delete.parquet")
+ ref := "data.parquet"
+ offset := int64(8)
+ length := int64(16)
+ negativeOffset := int64(-1)
+ zeroLength := int64(0)
+
+ tests := []struct {
+ name string
+ file iceberg.DataFile
+ errSubstr string
+ }{
+ {
+ name: "missing referenced data file",
+ file: newRewriteDeletionVector(t, "missing-ref.puffin", "", &offset, &length),
+ errSubstr: "missing referenced_data_file",
+ },
+ {
+ name: "missing content offset",
+ file: newRewriteDeletionVector(t, "missing-offset.puffin", ref, nil, &length),
+ errSubstr: "missing content_offset",
+ },
+ {
+ name: "negative content offset",
+ file: newRewriteDeletionVector(t, "negative-offset.puffin", ref, &negativeOffset, &length),
+ errSubstr: "invalid content_offset -1",
+ },
+ {
+ name: "missing content size",
+ file: newRewriteDeletionVector(t, "missing-size.puffin", ref, &offset, nil),
+ errSubstr: "missing content_size_in_bytes",
+ },
+ {
+ name: "nonpositive content size",
+ file: newRewriteDeletionVector(t, "zero-size.puffin", ref, &offset, &zeroLength),
+ errSubstr: "invalid content_size_in_bytes 0",
+ },
+ }
+
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ tx := tbl.NewTransaction()
+ err := tx.ReplaceFilesWithDeleteFiles(t.Context(), nil, nil,
+ []iceberg.DataFile{oldDelete},
+ []table.DeleteFileAddition{{File: tt.file, DataSequenceNumber: 0}}, nil)
+ require.Error(t, err)
+ assert.Contains(t, err.Error(), tt.errSubstr)
+ })
+ }
+}
+
+func TestReplaceFilesWithDeleteFilesValidatesDeletionVectorIdentity(t *testing.T) {
+ tbl := newReplaceFilesTestTableVersion(t, 3)
+ oldDelete := newPosDeleteFile(t, "old-pos-delete.parquet")
+ offsetA, offsetB := int64(8), int64(24)
+ length := int64(16)
+
+ t.Run("duplicate referenced data file", func(t *testing.T) {
+ tx := tbl.NewTransaction()
+ err := tx.ReplaceFilesWithDeleteFiles(t.Context(), nil, nil,
+ []iceberg.DataFile{oldDelete},
+ []table.DeleteFileAddition{
+ {File: newRewriteDeletionVector(t, "a.puffin", "data.parquet", &offsetA, &length), DataSequenceNumber: 0},
+ {File: newRewriteDeletionVector(t, "b.puffin", "data.parquet", &offsetB, &length), DataSequenceNumber: 0},
+ }, nil)
+ require.Error(t, err)
+ assert.Contains(t, err.Error(), "must reference distinct data files")
+ })
+
+ t.Run("duplicate blob identity", func(t *testing.T) {
+ tx := tbl.NewTransaction()
+ err := tx.ReplaceFilesWithDeleteFiles(t.Context(), nil, nil,
+ []iceberg.DataFile{oldDelete},
+ []table.DeleteFileAddition{
+ {File: newRewriteDeletionVector(t, "shared.puffin", "data-a.parquet", &offsetA, &length), DataSequenceNumber: 0},
+ {File: newRewriteDeletionVector(t, "shared.puffin", "data-b.parquet", &offsetA, &length), DataSequenceNumber: 0},
+ }, nil)
+ require.Error(t, err)
+ assert.Contains(t, err.Error(), "blob identity must be unique")
+ })
+
+ t.Run("distinct blobs may share a Puffin path", func(t *testing.T) {
+ sharedTbl := appendTwoDataFiles(t, newReplaceFilesTestTable(t))
+ sourceDelete := newPosDeleteFile(t, sharedTbl.Location()+"/data/source-pos-delete.parquet")
+ tx := sharedTbl.NewTransaction()
+ require.NoError(t, tx.NewRowDelta(nil).AddDeletes(sourceDelete).Commit(t.Context()))
+ sharedTbl, err := tx.Commit(t.Context())
+ require.NoError(t, err)
+ sequence := deleteFileSequence(t, sharedTbl, sourceDelete.FilePath())
+
+ tx = sharedTbl.NewTransaction()
+ require.NoError(t, tx.UpgradeFormatVersion(3))
+ sharedTbl, err = tx.Commit(t.Context())
+ require.NoError(t, err)
+ tasks, err := sharedTbl.Scan().PlanFiles(t.Context())
+ require.NoError(t, err)
+ require.Len(t, tasks, 2)
+
+ sharedPath := sharedTbl.Location() + "/data/shared.puffin"
+ tx = sharedTbl.NewTransaction()
+ err = tx.ReplaceFilesWithDeleteFiles(t.Context(), nil, nil,
+ []iceberg.DataFile{sourceDelete},
+ []table.DeleteFileAddition{
+ {File: newRewriteDeletionVector(t, sharedPath, tasks[0].File.FilePath(), &offsetA, &length), DataSequenceNumber: sequence},
+ {File: newRewriteDeletionVector(t, sharedPath, tasks[1].File.FilePath(), &offsetB, &length), DataSequenceNumber: sequence},
+ }, nil)
+ require.NoError(t, err, "distinct DV blobs in one Puffin container must be accepted")
+ })
+}
+
+func TestReplaceFilesWithDeleteFilesRejectsPartialPositionDeleteToDVRewrite(t *testing.T) {
+ tbl := newReplaceFilesTestTable(t)
+ arrowSc, err := table.SchemaToArrowSchema(tbl.Schema(), nil, false, false)
+ require.NoError(t, err)
+
+ dataPath := tbl.Location() + "/data/data.parquet"
+ writeParquetFile(t, dataPath, arrowSc, `[{"id":1,"data":"a"},{"id":2,"data":"b"}]`)
+ tx := tbl.NewTransaction()
+ require.NoError(t, tx.AddFiles(t.Context(), []string{dataPath}, nil, false))
+ tbl, err = tx.Commit(t.Context())
+ require.NoError(t, err)
+
+ filePathField, ok := iceberg.PositionalDeleteSchema.FindFieldByName("file_path")
+ require.True(t, ok)
+ bound, err := iceberg.StringLiteral(dataPath).MarshalBinary()
+ require.NoError(t, err)
+ oldDeletes := make([]iceberg.DataFile, 0, 2)
+ for i := range 2 {
+ builder, err := iceberg.NewDataFileBuilder(
+ *iceberg.UnpartitionedSpec, iceberg.EntryContentPosDeletes,
+ fmt.Sprintf("%s/data/old-pos-delete-%d.parquet", tbl.Location(), i),
+ iceberg.ParquetFile, nil, nil, nil, 1, 128)
+ require.NoError(t, err)
+ if i == 0 {
+ builder.
+ LowerBoundValues(map[int][]byte{filePathField.ID: bound}).
+ UpperBoundValues(map[int][]byte{filePathField.ID: bound})
+ }
+ oldDeletes = append(oldDeletes, builder.Build())
+ }
+ tx = tbl.NewTransaction()
+ require.NoError(t, tx.NewRowDelta(nil).AddDeletes(oldDeletes...).Commit(t.Context()))
+ tbl, err = tx.Commit(t.Context())
+ require.NoError(t, err)
+
+ tx = tbl.NewTransaction()
+ require.NoError(t, tx.UpgradeFormatVersion(3))
+ tbl, err = tx.Commit(t.Context())
+ require.NoError(t, err)
+
+ sequence := deleteFileSequence(t, tbl, oldDeletes[0].FilePath())
+ offset, length := int64(8), int64(16)
+ replacement := newRewriteDeletionVector(t, tbl.Location()+"/data/replacement.puffin", dataPath, &offset, &length)
+
+ tx = tbl.NewTransaction()
+ err = tx.ReplaceFilesWithDeleteFiles(t.Context(), nil, nil,
+ []iceberg.DataFile{oldDeletes[0]},
+ []table.DeleteFileAddition{{File: replacement, DataSequenceNumber: sequence}}, nil)
+ require.Error(t, err)
+ assert.Contains(t, err.Error(), "requires replacing all applicable position-delete files")
+ assert.Contains(t, err.Error(), oldDeletes[1].FilePath())
+ assert.Contains(t, err.Error(), "partition-scoped")
+
+ tx = tbl.NewTransaction()
+ require.NoError(t, tx.ReplaceFilesWithDeleteFiles(t.Context(), nil, nil,
+ oldDeletes,
+ []table.DeleteFileAddition{{File: replacement, DataSequenceNumber: sequence}}, nil),
+ "the DV rewrite should be accepted once every applicable position delete is replaced")
+}
+
+func TestReplaceFilesWithDeleteFilesRejectsDeletionVectorPartitionMismatch(t *testing.T) {
+ tbl, spec := newPartitionedReplaceFilesTestTable(t)
+ dataPath := tbl.Location() + "/data/data.parquet"
+ dataPartition := map[int]any{1000: int64(10)}
+ dataBuilder, err := iceberg.NewDataFileBuilder(
+ spec, iceberg.EntryContentData, dataPath, iceberg.ParquetFile,
+ dataPartition, nil, nil, 1, 128)
+ require.NoError(t, err)
+ tx := tbl.NewTransaction()
+ require.NoError(t, tx.AddDataFiles(t.Context(), []iceberg.DataFile{dataBuilder.Build()}, nil))
+ tbl, err = tx.Commit(t.Context())
+ require.NoError(t, err)
+
+ oldDeletePath := tbl.Location() + "/data/old-pos-delete.parquet"
+ oldDeleteBuilder, err := iceberg.NewDataFileBuilder(
+ spec, iceberg.EntryContentPosDeletes, oldDeletePath, iceberg.ParquetFile,
+ dataPartition, nil, nil, 1, 128)
+ require.NoError(t, err)
+ oldDelete := oldDeleteBuilder.Build()
+ tx = tbl.NewTransaction()
+ require.NoError(t, tx.NewRowDelta(nil).AddDeletes(oldDelete).Commit(t.Context()))
+ tbl, err = tx.Commit(t.Context())
+ require.NoError(t, err)
+ deleteSequence := fileDataSequence(t, tbl, oldDeletePath)
+
+ tx = tbl.NewTransaction()
+ require.NoError(t, tx.UpgradeFormatVersion(3))
+ tbl, err = tx.Commit(t.Context())
+ require.NoError(t, err)
+
+ offset, length := int64(8), int64(16)
+ dvBuilder, err := iceberg.NewDataFileBuilder(
+ spec, iceberg.EntryContentPosDeletes,
+ tbl.Location()+"/data/replacement-dv.puffin", iceberg.PuffinFile,
+ map[int]any{1000: int64(11)}, nil, nil, 1, 128)
+ require.NoError(t, err)
+ replacement := dvBuilder.
+ ReferencedDataFile(dataPath).
+ ContentOffset(offset).
+ ContentSizeInBytes(length).
+ Build()
+
+ tx = tbl.NewTransaction()
+ err = tx.ReplaceFilesWithDeleteFiles(t.Context(), nil, nil,
+ []iceberg.DataFile{oldDelete},
+ []table.DeleteFileAddition{{File: replacement, DataSequenceNumber: deleteSequence}}, nil)
+ require.Error(t, err)
+ assert.Contains(t, err.Error(), "partition spec or values")
+ assert.Contains(t, err.Error(), dataPath)
+}
+
+func TestReplaceFilesWithDeleteFilesIgnoresOlderSurvivingPositionDelete(t *testing.T) {
+ tbl, spec := newPartitionedReplaceFilesTestTable(t)
+ partition := map[int]any{1000: int64(10)}
+
+ dataAPath := tbl.Location() + "/data/data-a.parquet"
+ dataABuilder, err := iceberg.NewDataFileBuilder(
+ spec, iceberg.EntryContentData, dataAPath, iceberg.ParquetFile,
+ partition, nil, nil, 1, 128)
+ require.NoError(t, err)
+ tx := tbl.NewTransaction()
+ require.NoError(t, tx.AddDataFiles(t.Context(), []iceberg.DataFile{dataABuilder.Build()}, nil))
+ tbl, err = tx.Commit(t.Context())
+ require.NoError(t, err)
+
+ oldPartitionDeletePath := tbl.Location() + "/data/old-partition-delete.parquet"
+ oldPartitionDeleteBuilder, err := iceberg.NewDataFileBuilder(
+ spec, iceberg.EntryContentPosDeletes, oldPartitionDeletePath, iceberg.ParquetFile,
+ partition, nil, nil, 1, 128)
+ require.NoError(t, err)
+ oldPartitionDelete := oldPartitionDeleteBuilder.Build()
+ tx = tbl.NewTransaction()
+ require.NoError(t, tx.NewRowDelta(nil).AddDeletes(oldPartitionDelete).Commit(t.Context()))
+ tbl, err = tx.Commit(t.Context())
+ require.NoError(t, err)
+ oldPartitionDeleteSequence := fileDataSequence(t, tbl, oldPartitionDeletePath)
+
+ dataBPath := tbl.Location() + "/data/data-b.parquet"
+ dataBBuilder, err := iceberg.NewDataFileBuilder(
+ spec, iceberg.EntryContentData, dataBPath, iceberg.ParquetFile,
+ partition, nil, nil, 1, 128)
+ require.NoError(t, err)
+ tx = tbl.NewTransaction()
+ require.NoError(t, tx.AddDataFiles(t.Context(), []iceberg.DataFile{dataBBuilder.Build()}, nil))
+ tbl, err = tx.Commit(t.Context())
+ require.NoError(t, err)
+ dataBSequence := fileDataSequence(t, tbl, dataBPath)
+ require.Greater(t, dataBSequence, oldPartitionDeleteSequence)
+
+ filePathField, ok := iceberg.PositionalDeleteSchema.FindFieldByName("file_path")
+ require.True(t, ok)
+ bound, err := iceberg.StringLiteral(dataBPath).MarshalBinary()
+ require.NoError(t, err)
+ newPositionDeletePath := tbl.Location() + "/data/new-position-delete.parquet"
+ newPositionDeleteBuilder, err := iceberg.NewDataFileBuilder(
+ spec, iceberg.EntryContentPosDeletes, newPositionDeletePath, iceberg.ParquetFile,
+ partition, nil, nil, 1, 128)
+ require.NoError(t, err)
+ newPositionDelete := newPositionDeleteBuilder.
+ LowerBoundValues(map[int][]byte{filePathField.ID: bound}).
+ UpperBoundValues(map[int][]byte{filePathField.ID: bound}).
+ Build()
+ tx = tbl.NewTransaction()
+ require.NoError(t, tx.NewRowDelta(nil).AddDeletes(newPositionDelete).Commit(t.Context()))
+ tbl, err = tx.Commit(t.Context())
+ require.NoError(t, err)
+ newPositionDeleteSequence := fileDataSequence(t, tbl, newPositionDeletePath)
+
+ tx = tbl.NewTransaction()
+ require.NoError(t, tx.UpgradeFormatVersion(3))
+ tbl, err = tx.Commit(t.Context())
+ require.NoError(t, err)
+
+ offset, length := int64(8), int64(16)
+ dvBuilder, err := iceberg.NewDataFileBuilder(
+ spec, iceberg.EntryContentPosDeletes,
+ tbl.Location()+"/data/replacement-dv.puffin", iceberg.PuffinFile,
+ partition, nil, nil, 1, 128)
+ require.NoError(t, err)
+ replacement := dvBuilder.
+ ReferencedDataFile(dataBPath).
+ ContentOffset(offset).
+ ContentSizeInBytes(length).
+ Build()
+
+ tx = tbl.NewTransaction()
+ require.NoError(t, tx.ReplaceFilesWithDeleteFiles(t.Context(), nil, nil,
+ []iceberg.DataFile{newPositionDelete},
+ []table.DeleteFileAddition{{File: replacement, DataSequenceNumber: newPositionDeleteSequence}}, nil),
+ "an older position delete cannot apply to the newer target data file")
+ tbl, err = tx.Commit(t.Context())
+ require.NoError(t, err)
+ assert.Equal(t, newPositionDeleteSequence, fileDataSequence(t, tbl, replacement.FilePath()))
+}
+
+func TestReplaceFilesWithDeleteFilesRejectsSurvivingDeletionVector(t *testing.T) {
+ tbl, target := seedV3TableWithDV(t)
+ tasks, err := tbl.Scan().PlanFiles(t.Context())
+ require.NoError(t, err)
+
+ var sibling string
+ for _, task := range tasks {
+ if task.File.FilePath() != target {
+ sibling = task.File.FilePath()
+
+ break
+ }
+ }
+ require.NotEmpty(t, sibling)
+
+ writer := dv.NewDVWriter(iceio.LocalFS{}, unpartitionedSpecByID)
+ require.NoError(t, writer.Add(sibling, []int64{0}, 0, nil))
+ siblingDVs, err := writer.Flush(t.Context(), tbl.Location()+"/data/sibling-dv.puffin")
+ require.NoError(t, err)
+ require.Len(t, siblingDVs, 1)
+ tx := tbl.NewTransaction()
+ require.NoError(t, tx.NewRowDelta(nil).AddDeletes(siblingDVs...).Commit(t.Context()))
+ tbl, err = tx.Commit(t.Context())
+ require.NoError(t, err)
+
+ offset, length := int64(8), int64(16)
+ replacement := newRewriteDeletionVector(t, tbl.Location()+"/data/replacement-dv.puffin", target, &offset, &length)
+ sequence := deleteFileSequence(t, tbl, siblingDVs[0].FilePath())
+ tx = tbl.NewTransaction()
+ err = tx.ReplaceFilesWithDeleteFiles(t.Context(), nil, nil,
+ []iceberg.DataFile{siblingDVs[0]},
+ []table.DeleteFileAddition{{File: replacement, DataSequenceNumber: sequence}}, nil)
+ require.Error(t, err)
+ assert.Contains(t, err.Error(), "already exists and must be replaced")
+}
+
+func TestReplaceFilesWithDeleteFilesAllowsDroppedEqualityField(t *testing.T) {
+ tbl := newReplaceFilesTestTable(t)
+ oldDeleteBuilder, err := iceberg.NewDataFileBuilder(
+ *iceberg.UnpartitionedSpec, iceberg.EntryContentEqDeletes,
+ tbl.Location()+"/data/old-equality-delete.parquet", iceberg.ParquetFile,
+ nil, nil, nil, 1, 128)
+ require.NoError(t, err)
+ oldDelete := oldDeleteBuilder.EqualityFieldIDs([]int{2}).Build()
+
+ tx := tbl.NewTransaction()
+ require.NoError(t, tx.NewRowDelta(nil).AddDeletes(oldDelete).Commit(t.Context()))
+ tbl, err = tx.Commit(t.Context())
+ require.NoError(t, err)
+ sequence := deleteFileSequence(t, tbl, oldDelete.FilePath())
+
+ tx = tbl.NewTransaction()
+ require.NoError(t, tx.UpdateSchema(true, false).DeleteColumn([]string{"data"}).Commit())
+ tbl, err = tx.Commit(t.Context())
+ require.NoError(t, err)
+ _, currentHasDroppedField := tbl.Schema().FindFieldByID(2)
+ require.False(t, currentHasDroppedField)
+
+ newDeleteBuilder, err := iceberg.NewDataFileBuilder(
+ *iceberg.UnpartitionedSpec, iceberg.EntryContentEqDeletes,
+ tbl.Location()+"/data/new-equality-delete.parquet", iceberg.ParquetFile,
+ nil, nil, nil, 1, 128)
+ require.NoError(t, err)
+ newDelete := newDeleteBuilder.EqualityFieldIDs([]int{2}).Build()
+ tx = tbl.NewTransaction()
+ require.NoError(t, tx.ReplaceFilesWithDeleteFiles(t.Context(), nil, nil,
+ []iceberg.DataFile{oldDelete},
+ []table.DeleteFileAddition{{File: newDelete, DataSequenceNumber: sequence}}, nil))
+}
+
+func deleteFileSequence(t *testing.T, tbl *table.Table, path string) int64 {
+ return fileDataSequence(t, tbl, path)
+}
+
+func fileDataSequence(t *testing.T, tbl *table.Table, path string) int64 {
+ t.Helper()
+ snap := tbl.CurrentSnapshot()
+ require.NotNil(t, snap)
+ manifests, err := snap.Manifests(iceio.LocalFS{})
+ require.NoError(t, err)
+ for _, manifest := range manifests {
+ for entry, err := range manifest.Entries(iceio.LocalFS{}, false) {
+ require.NoError(t, err)
+ if entry.DataFile().FilePath() == path {
+ return entry.SequenceNum()
+ }
+ }
+ }
+ t.Fatalf("file %q not found in current snapshot", path)
+
+ return -1
+}
+
+func scanIDs(t *testing.T, tbl *table.Table) []int64 {
+ t.Helper()
+ _, records, err := tbl.Scan().ToArrowRecords(t.Context())
+ require.NoError(t, err)
+ var ids []int64
+ for record, err := range records {
+ require.NoError(t, err)
+ values := record.Column(record.Schema().FieldIndices("id")[0]).(*array.Int64)
+ for i := range values.Len() {
+ ids = append(ids, values.Value(i))
+ }
+ record.Release()
+ }
+
+ return ids
+}
+
func TestReplaceFiles_DelegatesToReplaceDataFilesWhenNoDeleteFiles(t *testing.T) {
tbl := newReplaceFilesTestTable(t)
diff --git a/table/rewrite_data_files.go b/table/rewrite_data_files.go
index 1eeba63f3..9e04f1385 100644
--- a/table/rewrite_data_files.go
+++ b/table/rewrite_data_files.go
@@ -503,10 +503,25 @@ func accumulateGroupMetrics(r *RewriteResult, gr CompactionGroupResult) {
// Always runs — no isolation gating, because rewrite is a structural
// operation, not a user-facing isolation choice.
func rewriteValidator(rewrittenFiles []iceberg.DataFile) conflictValidatorFunc {
+ return rewriteValidatorWithReferencedDataFiles(rewrittenFiles, nil)
+}
+
+// rewriteValidatorWithReferencedDataFiles extends rewrite validation to data
+// files targeted by deletion vectors added by the rewrite. Existing targets
+// must remain live, and no concurrent delete may be added for them before the
+// rewrite commits; otherwise the rewrite could create a duplicate DV or leave
+// a position delete surviving beside the new DV.
+func rewriteValidatorWithReferencedDataFiles(
+ rewrittenFiles []iceberg.DataFile,
+ referencedDataFilePaths []string,
+) conflictValidatorFunc {
return func(cc *conflictContext) error {
if cc == nil {
return nil
}
+ if err := validateDataFilesExist(cc, referencedDataFilePaths); err != nil {
+ return err
+ }
return validateNoNewDeletesForRewrittenFiles(cc, rewrittenFiles)
}
diff --git a/table/rewrite_files.go b/table/rewrite_files.go
index ef4e5dd64..0c6835f72 100644
--- a/table/rewrite_files.go
+++ b/table/rewrite_files.go
@@ -57,20 +57,22 @@ import (
// [RewriteFiles.Commit] drains it. The builder is single-use; once
// Commit has been called, a second call returns an error regardless
// of whether the first call succeeded.
-//
-// Adding new delete files (e.g., rewriting position deletes into
-// deletion vectors) is not yet supported; [RewriteFiles.AddDataFile]
-// rejects pos/eq-delete inputs at insertion time. Add the support to
-// the underlying [Transaction.ReplaceFiles] before lifting that
-// restriction.
type RewriteFiles struct {
- txn *Transaction
- dataFilesToDelete []iceberg.DataFile
- dataFilesToAdd []iceberg.DataFile
- deleteFilesToRemove []iceberg.DataFile
- snapshotProps iceberg.Properties
- err error
- committed bool
+ txn *Transaction
+ dataFilesToDelete []iceberg.DataFile
+ dataFilesToAdd []iceberg.DataFile
+ deleteFilesToAdd []rewriteDeleteFileAddition
+ deleteFilesToRemove []iceberg.DataFile
+ autoDeleteFilesToRemove []iceberg.DataFile
+ dataSequenceNumber *int64
+ snapshotProps iceberg.Properties
+ err error
+ committed bool
+}
+
+type rewriteDeleteFileAddition struct {
+ file iceberg.DataFile
+ dataSequenceNumber *int64
}
// NewRewrite returns a [RewriteFiles] builder bound to this transaction.
@@ -120,12 +122,33 @@ func (r *RewriteFiles) DeleteFile(df iceberg.DataFile) *RewriteFiles {
return r
}
-// AddDataFile queues a new data file. Adding delete files is not yet
-// supported by the underlying snapshot machinery; a pos/eq-delete here
-// stages an error that is returned from the next [RewriteFiles.Commit]
-// call. The error names the offending file path so callers driving the
-// builder via [RewriteFiles.Apply] can identify it without tracking
-// queue order.
+// removeAutomaticDeleteFile queues a delete file that became removable as a
+// consequence of rewriting data files. These files are removed in the same
+// snapshot, but they are not considered source files for an explicit delete
+// file rewrite.
+func (r *RewriteFiles) removeAutomaticDeleteFile(df iceberg.DataFile) *RewriteFiles {
+ if r.err != nil {
+ return r
+ }
+ if df == nil {
+ r.err = fmt.Errorf("%w: automatic delete removal got nil data file", ErrInvalidOperation)
+
+ return r
+ }
+
+ switch df.ContentType() {
+ case iceberg.EntryContentPosDeletes, iceberg.EntryContentEqDeletes:
+ r.autoDeleteFilesToRemove = append(r.autoDeleteFilesToRemove, df)
+ default:
+ r.err = fmt.Errorf("%w: automatic delete removal got unsupported content type %s (%s)",
+ ErrInvalidOperation, df.ContentType(), df.FilePath())
+ }
+
+ return r
+}
+
+// AddDataFile queues a new data file. Use [RewriteFiles.AddDeleteFile] for
+// positional deletes, equality deletes, or deletion vectors.
func (r *RewriteFiles) AddDataFile(df iceberg.DataFile) *RewriteFiles {
if r.err != nil {
return r
@@ -147,10 +170,112 @@ func (r *RewriteFiles) AddDataFile(df iceberg.DataFile) *RewriteFiles {
return r
}
+// DataSequenceNumber configures the data sequence number used for every data
+// file added by this rewrite. This is useful when replacement data files must
+// retain the applicability of delete files committed before the rewrite
+// snapshot.
+func (r *RewriteFiles) DataSequenceNumber(seq int64) *RewriteFiles {
+ if r.err != nil {
+ return r
+ }
+ if seq < 0 {
+ r.err = fmt.Errorf("%w: invalid rewrite data sequence number %d", ErrInvalidOperation, seq)
+
+ return r
+ }
+
+ seqCopy := seq
+ r.dataSequenceNumber = &seqCopy
+
+ return r
+}
+
+// AddDeleteFile queues a new positional delete, equality delete, or deletion
+// vector for this rewrite. When existing delete files are replaced, the new
+// file inherits the maximum data sequence number of those files.
+func (r *RewriteFiles) AddDeleteFile(df iceberg.DataFile) *RewriteFiles {
+ if r.err != nil {
+ return r
+ }
+ if df == nil {
+ r.err = fmt.Errorf("%w: AddDeleteFile got nil data file", ErrInvalidOperation)
+
+ return r
+ }
+
+ switch df.ContentType() {
+ case iceberg.EntryContentPosDeletes, iceberg.EntryContentEqDeletes:
+ r.deleteFilesToAdd = append(r.deleteFilesToAdd, rewriteDeleteFileAddition{file: df})
+ default:
+ r.err = fmt.Errorf("%w: AddDeleteFile only supports delete files; got content type %s (%s)",
+ ErrInvalidOperation, df.ContentType(), df.FilePath())
+ }
+
+ return r
+}
+
+// AddDeleteFileWithDataSequenceNumber is a convenience form of
+// [RewriteFiles.AddDeleteFile]. The caller must assign the maximum data
+// sequence number of the exact source delete files represented by df.
+func (r *RewriteFiles) AddDeleteFileWithDataSequenceNumber(df iceberg.DataFile, seq int64) *RewriteFiles {
+ if r.err != nil {
+ return r
+ }
+ if df == nil {
+ return r.AddDeleteFile(df)
+ }
+ if seq < 0 {
+ r.err = fmt.Errorf("%w: AddDeleteFile got invalid data sequence number %d for %s",
+ ErrInvalidOperation, seq, df.FilePath())
+
+ return r
+ }
+
+ switch df.ContentType() {
+ case iceberg.EntryContentPosDeletes, iceberg.EntryContentEqDeletes:
+ seqCopy := seq
+ r.deleteFilesToAdd = append(r.deleteFilesToAdd, rewriteDeleteFileAddition{
+ file: df,
+ dataSequenceNumber: &seqCopy,
+ })
+ default:
+ r.err = fmt.Errorf("%w: AddDeleteFile only supports delete files; got content type %s (%s)",
+ ErrInvalidOperation, df.ContentType(), df.FilePath())
+ }
+
+ return r
+}
+
+// AddFile queues a file according to its content type. It is useful for
+// coordinators that carry data and delete files in one result slice.
+func (r *RewriteFiles) AddFile(df iceberg.DataFile) *RewriteFiles {
+ if r.err != nil {
+ return r
+ }
+ if df == nil {
+ r.err = fmt.Errorf("%w: AddFile got nil data file", ErrInvalidOperation)
+
+ return r
+ }
+
+ switch df.ContentType() {
+ case iceberg.EntryContentData:
+ return r.AddDataFile(df)
+ case iceberg.EntryContentPosDeletes, iceberg.EntryContentEqDeletes:
+ return r.AddDeleteFile(df)
+ default:
+ r.err = fmt.Errorf("%w: AddFile got unsupported content type %s (%s)",
+ ErrInvalidOperation, df.ContentType(), df.FilePath())
+
+ return r
+ }
+}
+
// Apply is a bulk shortcut that routes three slices onto this builder:
-// every entry in deletes and safeDeletes is queued via
+// every entry in deletes is queued via
// [RewriteFiles.DeleteFile] (which routes data vs. delete files by
-// content type), and every entry in adds via [RewriteFiles.AddDataFile].
+// content type), every entry in adds via [RewriteFiles.AddFile], and every
+// entry in safeDeletes as an automatic delete-file removal.
//
// Deprecated: use [RewriteFiles.ApplyResult], which also carries
// SafeDeletionVectors. Apply has no slot for them, so a coordinator wiring
@@ -165,19 +290,19 @@ func (r *RewriteFiles) Apply(deletes, adds, safeDeletes []iceberg.DataFile) *Rew
r.DeleteFile(df)
}
for _, df := range adds {
- r.AddDataFile(df)
+ r.AddFile(df)
}
for _, df := range safeDeletes {
- r.DeleteFile(df)
+ r.removeAutomaticDeleteFile(df)
}
return r
}
// ApplyResult queues a worker's [CompactionGroupResult] onto this builder,
-// routing OldDataFiles (DeleteFile), NewDataFiles (AddDataFile), SafePosDeletes
-// and SafeDeletionVectors (DeleteFile). Prefer it over [RewriteFiles.Apply],
-// which cannot carry SafeDeletionVectors.
+// routing OldDataFiles (DeleteFile), NewDataFiles (AddFile), SafePosDeletes
+// and SafeDeletionVectors as automatic delete-file removals. Prefer it over
+// [RewriteFiles.Apply], which cannot carry SafeDeletionVectors.
//
// Typical distributed-coordinator pattern:
//
@@ -191,15 +316,14 @@ func (r *RewriteFiles) ApplyResult(gr CompactionGroupResult) *RewriteFiles {
r.DeleteFile(df)
}
for _, df := range gr.NewDataFiles {
- r.AddDataFile(df)
+ r.AddFile(df)
}
- // DVs route through DeleteFile like any pos-delete; ReplaceFiles then
- // re-identifies them by referenced data file.
+ // DVs are automatic removals keyed by their referenced data file.
for _, df := range gr.SafePosDeletes {
- r.DeleteFile(df)
+ r.removeAutomaticDeleteFile(df)
}
for _, df := range gr.SafeDeletionVectors {
- r.DeleteFile(df)
+ r.removeAutomaticDeleteFile(df)
}
return r
@@ -211,7 +335,7 @@ func (r *RewriteFiles) ApplyResult(gr CompactionGroupResult) *RewriteFiles {
// Commit is single-shot: any second call returns an error regardless
// of whether the first call succeeded, and neither re-stages the
// rewrite nor re-registers the conflict validator. Returns an error
-// if any file passed to [RewriteFiles.AddDataFile] or
+// if any file passed to [RewriteFiles.AddFile], [RewriteFiles.AddDataFile], or
// [RewriteFiles.DeleteFile] had an unsupported content type, if the
// builder has no file changes, or if the underlying
// [Transaction.ReplaceFiles] call fails.
@@ -224,7 +348,9 @@ func (r *RewriteFiles) Commit(ctx context.Context) error {
if r.err != nil {
return r.err
}
- if len(r.dataFilesToDelete) == 0 && len(r.dataFilesToAdd) == 0 && len(r.deleteFilesToRemove) == 0 {
+ if len(r.dataFilesToDelete) == 0 && len(r.dataFilesToAdd) == 0 &&
+ len(r.deleteFilesToAdd) == 0 && len(r.deleteFilesToRemove) == 0 &&
+ len(r.autoDeleteFilesToRemove) == 0 {
return fmt.Errorf("%w: rewrite must have at least one file change", ErrInvalidOperation)
}
// Adds-without-deletes would route through ReplaceFiles →
@@ -237,11 +363,17 @@ func (r *RewriteFiles) Commit(ctx context.Context) error {
return fmt.Errorf("%w: rewrite must delete at least one data file when adding data files", ErrInvalidOperation)
}
- if err := r.txn.ReplaceFiles(ctx, r.dataFilesToDelete, r.dataFilesToAdd, r.deleteFilesToRemove, r.snapshotProps, withRewriteSemantics()); err != nil {
+ opts := []WriteOption{withRewriteSemantics()}
+ if r.dataSequenceNumber != nil {
+ opts = append(opts, withDataSequenceNumber(*r.dataSequenceNumber))
+ }
+ if err := r.txn.replaceFiles(ctx, r.dataFilesToDelete, r.dataFilesToAdd,
+ r.deleteFilesToRemove, r.autoDeleteFilesToRemove, r.deleteFilesToAdd,
+ r.snapshotProps, opts...); err != nil {
return err
}
- if len(r.dataFilesToDelete) > 0 {
+ if len(r.dataFilesToDelete) > 0 && len(r.deleteFilesToAdd) == 0 {
r.txn.addValidator(rewriteValidator(r.dataFilesToDelete))
}
diff --git a/table/rewrite_files_test.go b/table/rewrite_files_test.go
index 5ee289e47..d205d62ae 100644
--- a/table/rewrite_files_test.go
+++ b/table/rewrite_files_test.go
@@ -29,6 +29,7 @@ import (
"context"
"fmt"
"path/filepath"
+ "strconv"
"sync/atomic"
"testing"
"time"
@@ -78,6 +79,10 @@ func (c *concurrentTestCatalog) CommitTable(_ context.Context, _ table.Identifie
}
func newConcurrentRewriteTestTable(t *testing.T) (*table.Table, *concurrentTestCatalog) {
+ return newConcurrentRewriteTestTableVersion(t, 2)
+}
+
+func newConcurrentRewriteTestTableVersion(t *testing.T, version int) (*table.Table, *concurrentTestCatalog) {
t.Helper()
location := filepath.ToSlash(t.TempDir())
@@ -87,7 +92,7 @@ func newConcurrentRewriteTestTable(t *testing.T) (*table.Table, *concurrentTestC
)
meta, err := table.NewMetadata(schema, iceberg.UnpartitionedSpec, table.UnsortedSortOrder, location,
iceberg.Properties{
- table.PropertyFormatVersion: "2",
+ table.PropertyFormatVersion: strconv.Itoa(version),
table.CommitNumRetriesKey: "2",
table.CommitMinRetryWaitMsKey: "1",
table.CommitMaxRetryWaitMsKey: "2",
@@ -180,6 +185,39 @@ func TestRewriteFiles_EmptyCommit_Errors(t *testing.T) {
assert.Contains(t, err.Error(), "at least one file change")
}
+func TestRewriteFiles_RejectsNegativeDataSequenceNumber(t *testing.T) {
+ tbl := newRewriteTestTable(t)
+ rewrite := tbl.NewTransaction().NewRewrite(nil).DataSequenceNumber(-1)
+
+ err := rewrite.Commit(t.Context())
+ require.Error(t, err)
+ assert.ErrorIs(t, err, table.ErrInvalidOperation)
+ assert.Contains(t, err.Error(), "invalid rewrite data sequence number")
+}
+
+func TestRewriteFiles_RejectsDataSequenceNumberAboveSnapshot(t *testing.T) {
+ tbl := newRewriteTestTable(t)
+ oldData := newDataFile(t, tbl.Location()+"/data/old-data.parquet")
+ tx := tbl.NewTransaction()
+ require.NoError(t, tx.AddDataFiles(t.Context(), []iceberg.DataFile{oldData}, nil))
+ tbl, err := tx.Commit(t.Context())
+ require.NoError(t, err)
+
+ snapshot := tbl.CurrentSnapshot()
+ require.NotNil(t, snapshot)
+ nextSnapshotSequence := snapshot.SequenceNumber + 1
+ replacement := newDataFile(t, tbl.Location()+"/data/replacement-data.parquet")
+ tx = tbl.NewTransaction()
+ rewrite := tx.NewRewrite(nil).
+ DeleteFile(oldData).
+ AddDataFile(replacement).
+ DataSequenceNumber(nextSnapshotSequence + 1)
+ err = rewrite.Commit(t.Context())
+ require.Error(t, err)
+ assert.ErrorIs(t, err, table.ErrInvalidOperation)
+ assert.Contains(t, err.Error(), "cannot be greater than new snapshot sequence number")
+}
+
func TestRewriteFiles_AddDataFile_RejectsNonDataFile(t *testing.T) {
tbl := newRewriteTestTable(t)
tx := tbl.NewTransaction()
@@ -195,6 +233,30 @@ func TestRewriteFiles_AddDataFile_RejectsNonDataFile(t *testing.T) {
"error must name the offending file path")
}
+func TestRewriteFiles_AddFile_PreservesFirstError(t *testing.T) {
+ tbl := newRewriteTestTable(t)
+ tx := tbl.NewTransaction()
+
+ err := tx.NewRewrite(nil).
+ AddDataFile(newPosDeleteFile(t, "first-error.parquet")).
+ AddFile(nil).
+ Commit(t.Context())
+ require.Error(t, err)
+ assert.Contains(t, err.Error(), "AddDataFile only supports data files")
+ assert.NotContains(t, err.Error(), "AddFile got nil")
+}
+
+func TestRewriteFiles_RejectsAddOnlyDeleteFile(t *testing.T) {
+ tbl := newRewriteTestTable(t)
+ deleteFile := newPosDeleteFile(t, tbl.Location()+"/data/pos-delete.parquet")
+ tx := tbl.NewTransaction()
+ rewrite := tx.NewRewrite(nil).AddDeleteFile(deleteFile)
+ err := rewrite.Commit(t.Context())
+ require.Error(t, err)
+ assert.ErrorIs(t, err, table.ErrInvalidOperation)
+ assert.Contains(t, err.Error(), "requires replacing at least one existing delete file")
+}
+
func TestRewriteFiles_Commit_SingleShot(t *testing.T) {
tbl := newRewriteTestTable(t)
@@ -240,6 +302,59 @@ func TestRewriteFiles_Commit_SingleShot(t *testing.T) {
assert.Contains(t, err.Error(), "already called")
}
+func TestRewriteFiles_PreservesIndependentPositionDeleteSequences(t *testing.T) {
+ tbl := newRewriteTestTable(t)
+ arrowSc, err := table.SchemaToArrowSchema(tbl.Schema(), nil, false, false)
+ require.NoError(t, err)
+
+ dataPath := tbl.Location() + "/data/independent-seq-data.parquet"
+ writeParquetFile(t, dataPath, arrowSc,
+ `[{"id": 1, "data": "a"}, {"id": 2, "data": "b"}]`)
+ tx := tbl.NewTransaction()
+ require.NoError(t, tx.AddFiles(t.Context(), []string{dataPath}, nil, false))
+ tbl, err = tx.Commit(t.Context())
+ require.NoError(t, err)
+
+ oldDeletes := make([]iceberg.DataFile, 0, 2)
+ oldSequences := make([]int64, 0, 2)
+ for i := range 2 {
+ path := tbl.Location() + fmt.Sprintf("/data/independent-old-%d.parquet", i)
+ writeParquetFile(t, path, table.PositionalDeleteArrowSchema,
+ fmt.Sprintf(`[{"file_path": %q, "pos": %d}]`, dataPath, i))
+ builder := newPosDeleteFile(t, path)
+ tx = tbl.NewTransaction()
+ require.NoError(t, tx.NewRowDelta(nil).AddDeletes(builder).Commit(t.Context()))
+ tbl, err = tx.Commit(t.Context())
+ require.NoError(t, err)
+ oldDeletes = append(oldDeletes, builder)
+ oldSequences = append(oldSequences, deleteFileSequence(t, tbl, path))
+ }
+ require.NotEqual(t, oldSequences[0], oldSequences[1])
+
+ newDeletes := make([]iceberg.DataFile, 0, 2)
+ for i := range 2 {
+ path := tbl.Location() + fmt.Sprintf("/data/independent-new-%d.parquet", i)
+ writeParquetFile(t, path, table.PositionalDeleteArrowSchema,
+ fmt.Sprintf(`[{"file_path": %q, "pos": %d}]`, dataPath, i))
+ newDeletes = append(newDeletes, newPosDeleteFile(t, path))
+ }
+
+ tx = tbl.NewTransaction()
+ rewrite := tx.NewRewrite(nil)
+ for i := range oldDeletes {
+ rewrite.DeleteFile(oldDeletes[i]).AddDeleteFileWithDataSequenceNumber(newDeletes[i], oldSequences[i])
+ }
+ require.NoError(t, rewrite.Commit(t.Context()))
+ tbl, err = tx.Commit(t.Context())
+ require.NoError(t, err)
+
+ for i, df := range newDeletes {
+ assert.Equal(t, oldSequences[i], deleteFileSequence(t, tbl, df.FilePath()))
+ }
+ assert.Empty(t, scanIDs(t, tbl),
+ "each replacement must retain the applicability of its corresponding old position delete")
+}
+
// TestRewriteFiles_Commit_SingleShot_AfterFailure pins down the
// stricter contract: a builder is dead even when the first Commit
// failed before reaching ReplaceFiles. Without this guard, an
@@ -679,6 +794,57 @@ func TestRewriteFiles_RejectsConcurrentEqDelete(t *testing.T) {
"only the stale-assertion attempt landed; the retry never reached CommitTable")
}
+// TestRewriteFiles_RejectsConcurrentDeleteForAddedDVTarget covers a pure
+// delete-file rewrite. The rewrite does not replace a data file, but it adds
+// a deletion vector for one, so the target still needs rewrite conflict
+// validation. A concurrent DV for the same target must not be allowed to
+// commit alongside it.
+func TestRewriteFiles_RejectsConcurrentDeleteForAddedDVTarget(t *testing.T) {
+ tbl, cat := newConcurrentRewriteTestTableVersion(t, 2)
+
+ dataPath := tbl.Location() + "/data/dv-target.parquet"
+ dataFile := newDataFile(t, dataPath)
+ tx := tbl.NewTransaction()
+ require.NoError(t, tx.AddDataFiles(t.Context(), []iceberg.DataFile{dataFile}, nil))
+ tbl, err := tx.Commit(t.Context())
+ require.NoError(t, err)
+
+ oldDelete := newPosDeleteFile(t, tbl.Location()+"/data/old-pos-delete.parquet")
+ tx = tbl.NewTransaction()
+ require.NoError(t, tx.NewRowDelta(nil).AddDeletes(oldDelete).Commit(t.Context()))
+ tbl, err = tx.Commit(t.Context())
+ require.NoError(t, err)
+ oldDeleteSequence := fileDataSequence(t, tbl, oldDelete.FilePath())
+
+ tx = tbl.NewTransaction()
+ require.NoError(t, tx.UpgradeFormatVersion(3))
+ tbl, err = tx.Commit(t.Context())
+ require.NoError(t, err)
+
+ firstOffset, secondOffset, length := int64(8), int64(24), int64(16)
+ leaderDV := newRewriteDeletionVector(t, tbl.Location()+"/data/leader.puffin", dataPath, &firstOffset, &length)
+ leaderTxn := tbl.NewTransaction()
+ rewrite := leaderTxn.NewRewrite(nil).
+ DeleteFile(oldDelete).
+ AddDeleteFileWithDataSequenceNumber(leaderDV, oldDeleteSequence)
+ require.NoError(t, rewrite.Commit(t.Context()),
+ "the conflict must be checked when the transaction commits")
+
+ peerDV := newRewriteDeletionVector(t, tbl.Location()+"/data/peer.puffin", dataPath, &secondOffset, &length)
+ peerTxn := tbl.NewTransaction()
+ require.NoError(t, peerTxn.NewRowDelta(nil).AddDeletes(peerDV).Commit(t.Context()))
+ _, err = peerTxn.Commit(t.Context())
+ require.NoError(t, err, "the peer commit advances the catalog")
+
+ beforeLeader := cat.attempts.Load()
+ _, err = leaderTxn.Commit(t.Context())
+ require.Error(t, err)
+ assert.ErrorIs(t, err, table.ErrCommitFailed,
+ "a concurrent delete for an added DV target must be rejected before replay")
+ assert.Equal(t, int32(1), cat.attempts.Load()-beforeLeader,
+ "the retry must stop in conflict validation before a second catalog commit")
+}
+
// TestRewriteFiles_RejectsConcurrentPartitionScopedPosDelete covers the
// gap fixed by validateNoNewDeletesForRewrittenFiles' partition
// fallback: a concurrent pos-delete with neither ReferencedDataFile nor
diff --git a/table/snapshot_producers.go b/table/snapshot_producers.go
index ab79975e3..6e612054c 100644
--- a/table/snapshot_producers.go
+++ b/table/snapshot_producers.go
@@ -569,19 +569,30 @@ func (m *mergeAppendFiles) needsValidation() bool { return false }
type snapshotProducer struct {
producerImpl
- commitUuid uuid.UUID
- io iceio.WriteFileIO
- txn *Transaction
- op Operation
- snapshotID int64
- parentSnapshotID int64
- addedFiles []iceberg.DataFile
- addedDeleteFiles []iceberg.DataFile
- manifestCount atomic.Int32
- deletedFiles map[string]iceberg.DataFile
- deletedDeleteFiles map[string]iceberg.DataFile
- deletedDVsByRef map[string]iceberg.DataFile
- snapshotProps iceberg.Properties
+ commitUuid uuid.UUID
+ io iceio.WriteFileIO
+ txn *Transaction
+ op Operation
+ snapshotID int64
+ parentSnapshotID int64
+ addedFiles []iceberg.DataFile
+ addedDataSequenceNumber *int64
+ addedDeleteFiles []deleteFileAddition
+ manifestCount atomic.Int32
+ deletedFiles map[string]iceberg.DataFile
+ deletedDeleteFiles map[string]iceberg.DataFile
+ deletedDVsByRef map[string]iceberg.DataFile
+ snapshotProps iceberg.Properties
+}
+
+// deleteFileAddition carries the data sequence number of a rewritten delete
+// file. A nil sequence means the file is a normal delete added by the current
+// snapshot and should inherit that snapshot's sequence number. Rewrite paths
+// use an explicit sequence so replacing delete files cannot change which data
+// files the delete applies to.
+type deleteFileAddition struct {
+ file iceberg.DataFile
+ dataSequenceNumber *int64
}
func createSnapshotProducer(op Operation, txn *Transaction, fs iceio.WriteFileIO, commitUUID *uuid.UUID, snapshotProps iceberg.Properties) *snapshotProducer {
@@ -629,8 +640,24 @@ func (sp *snapshotProducer) appendDataFile(df iceberg.DataFile) *snapshotProduce
return sp
}
+func (sp *snapshotProducer) setNewDataFilesDataSequenceNumber(seq int64) *snapshotProducer {
+ seqCopy := seq
+ sp.addedDataSequenceNumber = &seqCopy
+
+ return sp
+}
+
func (sp *snapshotProducer) appendDeleteFile(df iceberg.DataFile) *snapshotProducer {
- sp.addedDeleteFiles = append(sp.addedDeleteFiles, df)
+ sp.addedDeleteFiles = append(sp.addedDeleteFiles, deleteFileAddition{file: df})
+
+ return sp
+}
+
+func (sp *snapshotProducer) appendDeleteFileWithDataSequenceNumber(df iceberg.DataFile, seq int64) *snapshotProducer {
+ sp.addedDeleteFiles = append(sp.addedDeleteFiles, deleteFileAddition{
+ file: df,
+ dataSequenceNumber: &seq,
+ })
return sp
}
@@ -771,7 +798,7 @@ func (sp *snapshotProducer) addedContentManifests() ([]iceberg.ManifestFile, err
}
if len(sp.addedDeleteFiles) > 0 {
- g.Go(sp.manifestProducer(iceberg.ManifestContentDeletes, sp.addedDeleteFiles, &positionDeleteManifests))
+ g.Go(sp.deleteManifestProducer(&positionDeleteManifests))
}
if err := g.Wait(); err != nil {
@@ -781,6 +808,27 @@ func (sp *snapshotProducer) addedContentManifests() ([]iceberg.ManifestFile, err
return slices.Concat(addedManifests, positionDeleteManifests), nil
}
+func (sp *snapshotProducer) deleteManifestProducer(output *[]iceberg.ManifestFile) func() error {
+ return func() error {
+ groups := make(map[int][]deleteFileAddition)
+ for _, addition := range sp.addedDeleteFiles {
+ specID := int(addition.file.SpecID())
+ groups[specID] = append(groups[specID], addition)
+ }
+
+ for _, specID := range slices.Sorted(maps.Keys(groups)) {
+ additions := groups[specID]
+ mf, err := sp.writeAddedDeleteManifest(specID, additions)
+ if err != nil {
+ return err
+ }
+ *output = append(*output, mf)
+ }
+
+ return nil
+ }
+}
+
// assembleManifests recomputes the parent-dependent manifests against parent and
// combines them with the already-written added-content manifests, then applies
// producer-specific post-processing (manifest merging for merge-append, a no-op
@@ -938,7 +986,7 @@ func (sp *snapshotProducer) writeAddedManifest(content iceberg.ManifestContent,
for _, df := range files {
err := wr.Add(iceberg.NewManifestEntry(iceberg.EntryStatusADDED, &sp.snapshotID,
- nil, nil, df))
+ sp.addedDataSequenceNumber, nil, df))
if err != nil {
return nil, err
}
@@ -953,6 +1001,35 @@ func (sp *snapshotProducer) writeAddedManifest(content iceberg.ManifestContent,
return wr.ToManifestFile(path, counter.Count, iceberg.WithManifestFileContent(content))
}
+func (sp *snapshotProducer) writeAddedDeleteManifest(specID int, additions []deleteFileAddition) (_ iceberg.ManifestFile, retErr error) {
+ wr, path, counter, out, err := sp.newManifestWriter(sp.spec(specID), iceberg.WithManifestWriterContent(iceberg.ManifestContentDeletes))
+ if err != nil {
+ return nil, err
+ }
+ defer internal.CheckedClose(out, &retErr)
+ writerClosed := false
+ defer func() {
+ if !writerClosed {
+ internal.CheckedClose(wr, &retErr)
+ }
+ }()
+
+ for _, addition := range additions {
+ err := wr.Add(iceberg.NewManifestEntry(iceberg.EntryStatusADDED, &sp.snapshotID,
+ addition.dataSequenceNumber, nil, addition.file))
+ if err != nil {
+ return nil, err
+ }
+ }
+
+ writerClosed = true
+ if err := wr.Close(); err != nil {
+ return nil, err
+ }
+
+ return wr.ToManifestFile(path, counter.Count, iceberg.WithManifestFileContent(iceberg.ManifestContentDeletes))
+}
+
func (sp *snapshotProducer) summary(props iceberg.Properties) (Summary, error) {
delta, err := sp.accumulateSummaryDelta(nil)
if err != nil {
@@ -1019,8 +1096,8 @@ func (sp *snapshotProducer) accumulateSummaryDelta(countDeleteRemoval func(icebe
return nil, err
}
}
- for _, df := range sp.addedDeleteFiles {
- if err = ssc.addFile(df, currentSchema, sp.spec(int(df.SpecID()))); err != nil {
+ for _, addition := range sp.addedDeleteFiles {
+ if err = ssc.addFile(addition.file, currentSchema, sp.spec(int(addition.file.SpecID()))); err != nil {
return nil, err
}
}
diff --git a/table/transaction.go b/table/transaction.go
index 31b7dcef0..d2ff1df97 100644
--- a/table/transaction.go
+++ b/table/transaction.go
@@ -1092,6 +1092,257 @@ func (t *Transaction) validateDataFilesToAdd(dataFiles []iceberg.DataFile, opera
return setToAdd, nil
}
+type deletionVectorBlobKey struct {
+ path string
+ offset int64
+ length int64
+}
+
+type deleteFilesToAddSet struct {
+ paths map[string]struct{}
+ regularPaths map[string]struct{}
+ dvPaths map[string]struct{}
+ dvBlobs map[deletionVectorBlobKey]struct{}
+ dvsByRef map[string]rewriteDeleteFileAddition
+}
+
+// validateDeleteFilesToAdd performs metadata-only validation for delete files
+// supplied to a rewrite. Delete files may use an older partition spec, so the
+// partition values are checked against the spec carried by each file rather
+// than only against the table's current spec.
+func (t *Transaction) validateDeleteFilesToAdd(deleteFiles []rewriteDeleteFileAddition, operation string) (*deleteFilesToAddSet, error) {
+ meta, err := t.txnMeta()
+ if err != nil {
+ return nil, err
+ }
+ setToAdd := &deleteFilesToAddSet{
+ paths: make(map[string]struct{}, len(deleteFiles)),
+ regularPaths: make(map[string]struct{}, len(deleteFiles)),
+ dvPaths: make(map[string]struct{}, len(deleteFiles)),
+ dvBlobs: make(map[deletionVectorBlobKey]struct{}, len(deleteFiles)),
+ dvsByRef: make(map[string]rewriteDeleteFileAddition, len(deleteFiles)),
+ }
+ if len(deleteFiles) == 0 {
+ return setToAdd, nil
+ }
+ if meta.formatVersion < 2 {
+ return nil, fmt.Errorf("delete files require table format version >= 2, got v%d", meta.formatVersion)
+ }
+
+ for i, addition := range deleteFiles {
+ df := addition.file
+ if df == nil {
+ return nil, fmt.Errorf("nil delete file at index %d for %s", i, operation)
+ }
+ if addition.dataSequenceNumber != nil && *addition.dataSequenceNumber < 0 {
+ return nil, fmt.Errorf("delete file %s has invalid data sequence number %d for %s",
+ df.FilePath(), *addition.dataSequenceNumber, operation)
+ }
+ path := df.FilePath()
+ if path == "" {
+ return nil, fmt.Errorf("delete file path cannot be empty for %s", operation)
+ }
+ setToAdd.paths[path] = struct{}{}
+
+ switch df.ContentType() {
+ case iceberg.EntryContentPosDeletes, iceberg.EntryContentEqDeletes:
+ default:
+ return nil, fmt.Errorf("adding non-delete file %s has content type %s for %s", path, df.ContentType(), operation)
+ }
+
+ spec, err := meta.GetSpecByID(int(df.SpecID()))
+ if err != nil {
+ return nil, fmt.Errorf("delete file %s references unknown partition spec id %d for %s: %w", path, df.SpecID(), operation, err)
+ }
+ if spec == nil {
+ return nil, fmt.Errorf("delete file %s references unknown partition spec id %d for %s", path, df.SpecID(), operation)
+ }
+ if err := validateDataFilePartitionData(df, spec); err != nil {
+ return nil, fmt.Errorf("delete file %s has invalid partition data for %s: %w", path, operation, err)
+ }
+
+ if df.ContentType() == iceberg.EntryContentEqDeletes {
+ eqIDs := df.EqualityFieldIDs()
+ if len(eqIDs) == 0 {
+ return nil, fmt.Errorf("equality delete file must have non-empty EqualityFieldIDs: %s", path)
+ }
+ if err := validateRewriteEqualityFieldIDs(meta.schemaList, eqIDs); err != nil {
+ return nil, fmt.Errorf("invalid equality delete file %s: %w", path, err)
+ }
+ }
+
+ if !IsDeletionVector(df) {
+ if meta.formatVersion >= 3 && df.ContentType() == iceberg.EntryContentPosDeletes {
+ return nil, fmt.Errorf("position delete file %s must be a deletion vector for v%d table for %s",
+ path, meta.formatVersion, operation)
+ }
+ if _, ok := setToAdd.regularPaths[path]; ok {
+ return nil, fmt.Errorf("add delete file paths must be unique for %s", operation)
+ }
+ if _, ok := setToAdd.dvPaths[path]; ok {
+ return nil, fmt.Errorf("delete file path %s cannot identify both a deletion vector container and a regular delete file for %s",
+ path, operation)
+ }
+ setToAdd.regularPaths[path] = struct{}{}
+
+ continue
+ }
+
+ if IsDeletionVector(df) {
+ if meta.formatVersion < 3 {
+ return nil, fmt.Errorf("deletion vector %s requires table format version >= 3 for %s",
+ path, operation)
+ }
+ ref := df.ReferencedDataFile()
+ if ref == nil || *ref == "" {
+ return nil, fmt.Errorf("deletion vector to add is missing referenced_data_file for %s", operation)
+ }
+ offset := df.ContentOffset()
+ if offset == nil {
+ return nil, fmt.Errorf("deletion vector %s is missing content_offset for %s", path, operation)
+ }
+ if *offset < 0 {
+ return nil, fmt.Errorf("deletion vector %s has invalid content_offset %d for %s", path, *offset, operation)
+ }
+ length := df.ContentSizeInBytes()
+ if length == nil {
+ return nil, fmt.Errorf("deletion vector %s is missing content_size_in_bytes for %s", path, operation)
+ }
+ if *length <= 0 {
+ return nil, fmt.Errorf("deletion vector %s has invalid content_size_in_bytes %d for %s", path, *length, operation)
+ }
+
+ blob := deletionVectorBlobKey{path: path, offset: *offset, length: *length}
+ if _, ok := setToAdd.dvBlobs[blob]; ok {
+ return nil, fmt.Errorf("deletion vector blob identity must be unique for %s: %s at offset %d with length %d",
+ operation, path, *offset, *length)
+ }
+ if _, ok := setToAdd.dvsByRef[*ref]; ok {
+ return nil, fmt.Errorf("deletion vectors to add must reference distinct data files for %s: %s",
+ operation, *ref)
+ }
+ if _, ok := setToAdd.regularPaths[path]; ok {
+ return nil, fmt.Errorf("delete file path %s cannot identify both a deletion vector container and a regular delete file for %s",
+ path, operation)
+ }
+ setToAdd.dvPaths[path] = struct{}{}
+ setToAdd.dvBlobs[blob] = struct{}{}
+ setToAdd.dvsByRef[*ref] = addition
+ }
+ }
+
+ return setToAdd, nil
+}
+
+// validateRewriteEqualityFieldIDs accepts fields retained only in schema
+// history. Rewriting an existing equality delete must preserve its original
+// key even after a key column has been dropped from the current schema.
+func validateRewriteEqualityFieldIDs(schemas []*iceberg.Schema, fieldIDs []int) error {
+ for _, id := range fieldIDs {
+ var field iceberg.NestedField
+ found := false
+ for _, schema := range schemas {
+ if candidate, ok := schema.FindFieldByID(id); ok {
+ field = candidate
+ found = true
+
+ break
+ }
+ }
+ if !found {
+ return fmt.Errorf("%w: field ID %d not found in table schema history", iceberg.ErrInvalidSchema, id)
+ }
+ if isFloatingPointType(field.Type) {
+ return fmt.Errorf("%w: equality field ID %d (%s) has unsupported floating-point type %s: floating-point columns cannot be used as equality delete keys",
+ iceberg.ErrInvalidSchema, id, field.Name, field.Type)
+ }
+ }
+
+ return nil
+}
+
+type rewriteFileState struct {
+ file iceberg.DataFile
+ dataSequenceNumber int64
+}
+
+// validateAddedDeletionVectorTargets guarantees that each new DV has the same
+// partition as its target, applies to the target by sequence number, and cannot
+// suppress a surviving position-delete file that may still apply. File-scoped
+// deletes are matched by referenced_data_file or file_path bounds;
+// partition-scoped deletes use the same spec/partition tuple fallback as scan
+// and rewrite conflict planning.
+func validateAddedDeletionVectorTargets(
+ dvsByRef map[string]rewriteDeleteFileAddition,
+ liveDataFiles map[string]rewriteFileState,
+ survivingPositionDeletes []rewriteFileState,
+ defaultDeleteSequenceNumber int64,
+) error {
+ for ref, addition := range dvsByRef {
+ dataState, ok := liveDataFiles[ref]
+ if !ok {
+ return fmt.Errorf("%w: deletion vector references data file that will not exist in the rewritten snapshot: %s",
+ ErrInvalidOperation, ref)
+ }
+ dataFile := dataState.file
+
+ partitionKey, err := canonicalPartitionKey(dataFile.SpecID(), dataFile.Partition())
+ if err != nil {
+ return fmt.Errorf("building partition key for deletion vector target %s (spec %d): %w",
+ ref, dataFile.SpecID(), err)
+ }
+ dvPartitionKey, err := canonicalPartitionKey(addition.file.SpecID(), addition.file.Partition())
+ if err != nil {
+ return fmt.Errorf("building partition key for deletion vector %s (spec %d): %w",
+ addition.file.FilePath(), addition.file.SpecID(), err)
+ }
+ if dvPartitionKey != partitionKey {
+ return fmt.Errorf("%w: deletion vector %s has partition spec or values that do not match referenced data file %s",
+ ErrInvalidOperation, addition.file.FilePath(), ref)
+ }
+
+ dvSequenceNumber := defaultDeleteSequenceNumber
+ if addition.dataSequenceNumber != nil {
+ dvSequenceNumber = *addition.dataSequenceNumber
+ }
+ if dataState.dataSequenceNumber < 0 {
+ return fmt.Errorf("%w: referenced data file %s has no data sequence number",
+ ErrInvalidOperation, ref)
+ }
+ if dataState.dataSequenceNumber > dvSequenceNumber {
+ return fmt.Errorf("%w: deletion vector %s with data sequence number %d does not apply to referenced data file %s with data sequence number %d",
+ ErrInvalidOperation, addition.file.FilePath(), dvSequenceNumber, ref, dataState.dataSequenceNumber)
+ }
+
+ for _, deleteState := range survivingPositionDeletes {
+ if deleteState.dataSequenceNumber >= 0 && dataState.dataSequenceNumber > deleteState.dataSequenceNumber {
+ continue
+ }
+ deleteFile := deleteState.file
+ if target := referencedDataFilePath(deleteFile); target != "" {
+ if target == ref {
+ return fmt.Errorf("%w: adding a deletion vector for %s requires replacing all applicable position-delete files; %s would survive",
+ ErrInvalidOperation, ref, deleteFile.FilePath())
+ }
+
+ continue
+ }
+
+ deletePartitionKey, err := canonicalPartitionKey(deleteFile.SpecID(), deleteFile.Partition())
+ if err != nil {
+ return fmt.Errorf("building partition key for surviving position-delete file %s (spec %d): %w",
+ deleteFile.FilePath(), deleteFile.SpecID(), err)
+ }
+ if deletePartitionKey == partitionKey {
+ return fmt.Errorf("%w: adding a deletion vector for %s requires replacing all applicable position-delete files; partition-scoped file %s would survive",
+ ErrInvalidOperation, ref, deleteFile.FilePath())
+ }
+ }
+ }
+
+ return nil
+}
+
// WriteOption is an option for methods that operate on pre-built DataFile objects.
type WriteOption func(*dataFileCfg)
@@ -1099,6 +1350,7 @@ type dataFileCfg struct {
skipAutoNameMapping bool
skipDuplicateCheck bool
rewriteSemantics bool
+ dataSequenceNumber *int64
}
// withRewriteSemantics marks an overwrite/replace operation as a
@@ -1117,6 +1369,13 @@ func withRewriteSemantics() WriteOption {
}
}
+func withDataSequenceNumber(seq int64) WriteOption {
+ return func(cfg *dataFileCfg) {
+ seqCopy := seq
+ cfg.dataSequenceNumber = &seqCopy
+ }
+}
+
// WithoutAutoNameMapping disables the automatic setting of the schema name
// mapping in table properties. By default, methods like [Transaction.AddDataFiles]
// and [Transaction.ReplaceDataFilesWithDataFiles] will set the name mapping if
@@ -1367,6 +1626,9 @@ func (t *Transaction) ReplaceDataFilesWithDataFiles(ctx context.Context, filesTo
// mergeOverwrite guarantees an *overwriteFiles producerImpl.
updater.producerImpl.(*overwriteFiles).skipDefaultValidator = true
}
+ if cfg.dataSequenceNumber != nil {
+ updater.setNewDataFilesDataSequenceNumber(*cfg.dataSequenceNumber)
+ }
for _, df := range markedForDeletion {
updater.deleteDataFile(df)
@@ -1384,29 +1646,82 @@ func (t *Transaction) ReplaceDataFilesWithDataFiles(ctx context.Context, filesTo
return t.apply(updates, reqs)
}
+// DeleteFileAddition is a delete file added by a rewrite. The data sequence
+// number must be copied from the delete files being replaced so the new file
+// has the same applicability boundary as the old files. When one rewrite
+// contains multiple replacement groups, callers are responsible for assigning
+// each output the maximum sequence number of the exact source files it replaces;
+// the transaction can validate membership in the overall source sequence set,
+// but the grouping is not represented by this API.
+type DeleteFileAddition struct {
+ File iceberg.DataFile
+ DataSequenceNumber int64
+}
+
// ReplaceFiles atomically replaces data files and removes associated delete files
// in a single snapshot. This is the commit primitive for compaction: old data files
// are replaced with new (compacted) data files, and delete files that are fully
// applied are removed.
func (t *Transaction) ReplaceFiles(ctx context.Context, dataFilesToDelete, dataFilesToAdd, deleteFilesToRemove []iceberg.DataFile, snapshotProps iceberg.Properties, opts ...WriteOption) error {
+ return t.replaceFiles(ctx, dataFilesToDelete, dataFilesToAdd, deleteFilesToRemove, nil, nil, snapshotProps, opts...)
+}
+
+// ReplaceFilesWithDeleteFiles atomically replaces data files, adds new delete
+// files, and removes superseded delete files in one snapshot. Every added delete
+// file must carry the data sequence number of the delete files it replaces.
+// This is intended for rewrites that change delete-file representation, such as
+// rewriting position deletes into deletion vectors.
+func (t *Transaction) ReplaceFilesWithDeleteFiles(ctx context.Context, dataFilesToDelete, dataFilesToAdd, deleteFilesToRemove []iceberg.DataFile, deleteFilesToAdd []DeleteFileAddition, snapshotProps iceberg.Properties, opts ...WriteOption) error {
+ additions := make([]rewriteDeleteFileAddition, len(deleteFilesToAdd))
+ for i, addition := range deleteFilesToAdd {
+ seq := addition.DataSequenceNumber
+ additions[i] = rewriteDeleteFileAddition{
+ file: addition.File,
+ dataSequenceNumber: &seq,
+ }
+ }
+
+ return t.replaceFiles(ctx, dataFilesToDelete, dataFilesToAdd, deleteFilesToRemove, nil, additions, snapshotProps, opts...)
+}
+
+func (t *Transaction) replaceFiles(ctx context.Context, dataFilesToDelete, dataFilesToAdd, deleteFilesToRemove, autoDeleteFilesToRemove []iceberg.DataFile, deleteFilesToAdd []rewriteDeleteFileAddition, snapshotProps iceberg.Properties, opts ...WriteOption) error {
meta, err := t.txnMeta()
if err != nil {
return err
}
- // Delegate data file replacement to existing logic.
- if len(deleteFilesToRemove) == 0 {
- return t.ReplaceDataFilesWithDataFiles(ctx, dataFilesToDelete, dataFilesToAdd, snapshotProps, opts...)
- }
-
var cfg dataFileCfg
for _, o := range opts {
o(&cfg)
}
+ nextSequenceNumber := meta.nextSequenceNumber()
+ if cfg.dataSequenceNumber != nil && *cfg.dataSequenceNumber > nextSequenceNumber {
+ return fmt.Errorf("%w: data sequence number %d cannot be greater than new snapshot sequence number %d",
+ ErrInvalidOperation, *cfg.dataSequenceNumber, nextSequenceNumber)
+ }
+
+ // Delegate data file replacement to existing logic.
+ if len(deleteFilesToRemove) == 0 && len(autoDeleteFilesToRemove) == 0 && len(deleteFilesToAdd) == 0 {
+ return t.ReplaceDataFilesWithDataFiles(ctx, dataFilesToDelete, dataFilesToAdd, snapshotProps, opts...)
+ }
+ if len(deleteFilesToAdd) > 0 && len(deleteFilesToRemove) == 0 {
+ return fmt.Errorf("%w: adding delete files requires replacing at least one existing delete file",
+ ErrInvalidOperation)
+ }
+
setToAdd, err := t.validateDataFilesToAdd(dataFilesToAdd, "ReplaceFiles", !cfg.rewriteSemantics)
if err != nil {
return err
}
+ setDeleteFilesToAdd, err := t.validateDeleteFilesToAdd(deleteFilesToAdd, "ReplaceFiles")
+ if err != nil {
+ return err
+ }
+ for path := range setDeleteFilesToAdd.paths {
+ if _, ok := setToAdd[path]; ok {
+ return fmt.Errorf("data and delete file paths must be unique for ReplaceFiles: %s", path)
+ }
+ }
setToDelete := make(map[string]struct{}, len(dataFilesToDelete))
for i, df := range dataFilesToDelete {
@@ -1450,6 +1765,39 @@ func (t *Transaction) ReplaceFiles(ctx context.Context, dataFilesToDelete, dataF
}
setDeleteFilesToRemove[path] = struct{}{}
}
+ autoSetDeleteFilesToRemove := make(map[string]struct{}, len(autoDeleteFilesToRemove))
+ autoDVRefsToRemove := make(map[string]struct{}, len(autoDeleteFilesToRemove))
+ for i, df := range autoDeleteFilesToRemove {
+ if df == nil {
+ return fmt.Errorf("nil automatic delete file at index %d for ReplaceFiles", i)
+ }
+ path := df.FilePath()
+ if path == "" {
+ return errors.New("automatic delete file paths must be non-empty for ReplaceFiles")
+ }
+ if IsDeletionVector(df) {
+ ref := df.ReferencedDataFile()
+ if ref == nil {
+ return errors.New("automatic deletion vector to remove is missing referenced_data_file for ReplaceFiles")
+ }
+ if _, ok := dvRefsToRemove[*ref]; ok {
+ return errors.New("delete vectors to remove must reference distinct data files for ReplaceFiles")
+ }
+ if _, ok := autoDVRefsToRemove[*ref]; ok {
+ return errors.New("automatic deletion vectors to remove must reference distinct data files for ReplaceFiles")
+ }
+ autoDVRefsToRemove[*ref] = struct{}{}
+
+ continue
+ }
+ if _, ok := setDeleteFilesToRemove[path]; ok {
+ return errors.New("delete file paths must be unique for ReplaceFiles")
+ }
+ if _, ok := autoSetDeleteFilesToRemove[path]; ok {
+ return errors.New("automatic delete file paths must be unique for ReplaceFiles")
+ }
+ autoSetDeleteFilesToRemove[path] = struct{}{}
+ }
s := t.planningSnapshot(meta)
if s == nil {
@@ -1469,42 +1817,200 @@ func (t *Transaction) ReplaceFiles(ctx context.Context, dataFilesToDelete, dataF
// that all files to delete/remove actually exist in the table.
markedDataForDeletion := make([]iceberg.DataFile, 0, len(setToDelete))
markedDeleteForRemoval := make([]iceberg.DataFile, 0, len(setDeleteFilesToRemove))
+ markedAutoDeleteForRemoval := make([]iceberg.DataFile, 0, len(autoSetDeleteFilesToRemove))
markedDVsForRemoval := make(map[string]iceberg.DataFile, len(dvRefsToRemove))
- for df, err := range s.dataFiles(fs, nil) {
+ removedDeleteSequenceNumbers := make([]int64, 0, len(deleteFilesToRemove))
+ removedDeleteContents := make(map[iceberg.ManifestEntryContent]struct{})
+ liveDataFiles := make(map[string]rewriteFileState)
+ survivingPositionDeletes := make([]rewriteFileState, 0)
+ for entry, err := range s.entries(fs, -1) {
if err != nil {
return err
}
+ df := entry.DataFile()
path := df.FilePath()
isData := df.ContentType() == iceberg.EntryContentData
+ isLive := entry.Status() != iceberg.EntryStatusDELETED
+ if isData && isLive {
+ liveDataFiles[path] = rewriteFileState{file: df, dataSequenceNumber: entry.SequenceNum()}
+ }
if _, ok := setToDelete[path]; ok && isData {
markedDataForDeletion = append(markedDataForDeletion, df)
}
if !isData {
if _, ok := setDeleteFilesToRemove[path]; ok {
markedDeleteForRemoval = append(markedDeleteForRemoval, df)
+ if seq := entry.SequenceNum(); seq >= 0 {
+ removedDeleteSequenceNumbers = append(removedDeleteSequenceNumbers, seq)
+ removedDeleteContents[df.ContentType()] = struct{}{}
+ } else {
+ return fmt.Errorf("delete file %s has no data sequence number in the current snapshot", path)
+ }
+ } else if _, ok := autoSetDeleteFilesToRemove[path]; ok {
+ markedAutoDeleteForRemoval = append(markedAutoDeleteForRemoval, df)
} else if ref := iceberginternal.BorrowedDataFileReferencedDataFile(df); IsDeletionVector(df) && ref != nil {
if _, ok := dvRefsToRemove[*ref]; ok {
markedDVsForRemoval[*ref] = df
+ if seq := entry.SequenceNum(); seq >= 0 {
+ removedDeleteSequenceNumbers = append(removedDeleteSequenceNumbers, seq)
+ removedDeleteContents[df.ContentType()] = struct{}{}
+ } else {
+ return fmt.Errorf("deletion vector %s has no data sequence number in the current snapshot", path)
+ }
+ } else if _, ok := autoDVRefsToRemove[*ref]; ok {
+ markedDVsForRemoval[*ref] = df
}
}
}
if _, ok := setToAdd[path]; ok {
return fmt.Errorf("cannot add files that are already referenced by table, files: %s", path)
}
+ if _, ok := setDeleteFilesToAdd.regularPaths[path]; ok {
+ return fmt.Errorf("cannot add files that are already referenced by table, files: %s", path)
+ }
+ if _, ok := setDeleteFilesToAdd.dvPaths[path]; ok && isLive {
+ if !IsDeletionVector(df) {
+ return fmt.Errorf("cannot add deletion vector container path already referenced by a non-DV file: %s", path)
+ }
+ if offset, length := df.ContentOffset(), df.ContentSizeInBytes(); offset != nil && length != nil {
+ blob := deletionVectorBlobKey{path: path, offset: *offset, length: *length}
+ if _, duplicate := setDeleteFilesToAdd.dvBlobs[blob]; duplicate {
+ return fmt.Errorf("cannot add deletion vector blob already referenced by table: %s at offset %d with length %d",
+ path, *offset, *length)
+ }
+ }
+ }
+
+ if !isLive || isData {
+ continue
+ }
+ if IsDeletionVector(df) {
+ ref := df.ReferencedDataFile()
+ if ref == nil {
+ continue
+ }
+ if _, addingReplacement := setDeleteFilesToAdd.dvsByRef[*ref]; !addingReplacement {
+ continue
+ }
+ if _, explicitlyRemoved := dvRefsToRemove[*ref]; explicitlyRemoved {
+ continue
+ }
+ if _, automaticallyRemoved := autoDVRefsToRemove[*ref]; automaticallyRemoved {
+ continue
+ }
+
+ return fmt.Errorf("%w: deletion vector for data file %s already exists and must be replaced",
+ ErrInvalidOperation, *ref)
+ }
+ if df.ContentType() == iceberg.EntryContentPosDeletes {
+ if _, removed := setDeleteFilesToRemove[path]; !removed {
+ if _, automaticallyRemoved := autoSetDeleteFilesToRemove[path]; !automaticallyRemoved {
+ survivingPositionDeletes = append(survivingPositionDeletes, rewriteFileState{
+ file: df, dataSequenceNumber: entry.SequenceNum(),
+ })
+ }
+ }
+ }
}
+ addedDataSequenceNumber := nextSequenceNumber
+ if cfg.dataSequenceNumber != nil {
+ addedDataSequenceNumber = *cfg.dataSequenceNumber
+ }
+ for _, df := range dataFilesToAdd {
+ liveDataFiles[df.FilePath()] = rewriteFileState{
+ file: df, dataSequenceNumber: addedDataSequenceNumber,
+ }
+ }
+ for path := range setToDelete {
+ delete(liveDataFiles, path)
+ }
if len(markedDataForDeletion) != len(setToDelete) {
return errors.New("cannot delete data files that do not belong to the table")
}
if len(markedDeleteForRemoval) != len(setDeleteFilesToRemove) {
return errors.New("cannot remove delete files that do not belong to the table")
}
+ if len(markedAutoDeleteForRemoval) != len(autoSetDeleteFilesToRemove) {
+ return errors.New("cannot remove automatic delete files that do not belong to the table")
+ }
// Keyed by referenced data file, so duplicate DV entries for one ref collapse
// to one slot; equality then means every requested ref exists in the table.
- if len(markedDVsForRemoval) != len(dvRefsToRemove) {
+ if len(markedDVsForRemoval) != len(dvRefsToRemove)+len(autoDVRefsToRemove) {
return errors.New("cannot remove deletion vectors that do not belong to the table")
}
+ deleteSequenceNumber := int64(-1)
+ if len(deleteFilesToAdd) > 0 {
+ if len(removedDeleteSequenceNumbers) == 0 {
+ return fmt.Errorf("%w: added delete files must replace existing delete files",
+ ErrInvalidOperation)
+ }
+ if len(removedDeleteContents) != 1 {
+ return fmt.Errorf("%w: delete-file rewrites cannot combine position and equality delete files",
+ ErrInvalidOperation)
+ }
+
+ maxSequenceNumber := removedDeleteSequenceNumbers[0]
+ var sourceContent iceberg.ManifestEntryContent
+ for content := range removedDeleteContents {
+ sourceContent = content
+ }
+ for _, seq := range removedDeleteSequenceNumbers[1:] {
+ maxSequenceNumber = max(maxSequenceNumber, seq)
+ }
+ deleteSequenceNumber = maxSequenceNumber
+ removedDeleteSeqs := make(map[int64]struct{}, len(removedDeleteSequenceNumbers))
+ for _, seq := range removedDeleteSequenceNumbers {
+ removedDeleteSeqs[seq] = struct{}{}
+ }
+ if sourceContent == iceberg.EntryContentEqDeletes && len(removedDeleteSeqs) != 1 {
+ return fmt.Errorf("%w: equality delete files with different data sequence numbers cannot be merged",
+ ErrInvalidOperation)
+ }
+ for _, addition := range deleteFilesToAdd {
+ if addition.file.ContentType() != sourceContent {
+ return fmt.Errorf("%w: cannot rewrite %s delete files as %s delete files",
+ ErrInvalidOperation, sourceContent, addition.file.ContentType())
+ }
+ if addition.dataSequenceNumber != nil {
+ if sourceContent == iceberg.EntryContentEqDeletes {
+ if *addition.dataSequenceNumber != maxSequenceNumber {
+ return fmt.Errorf("%w: equality delete file %s has data sequence number %d, expected %d",
+ ErrInvalidOperation, addition.file.FilePath(), *addition.dataSequenceNumber, maxSequenceNumber)
+ }
+ } else if _, ok := removedDeleteSeqs[*addition.dataSequenceNumber]; !ok {
+ return fmt.Errorf("%w: delete file %s has data sequence number %d, expected one of the replaced delete sequence numbers",
+ ErrInvalidOperation, addition.file.FilePath(), *addition.dataSequenceNumber)
+ }
+ }
+ }
+ }
+ if err := validateAddedDeletionVectorTargets(setDeleteFilesToAdd.dvsByRef, liveDataFiles,
+ survivingPositionDeletes, deleteSequenceNumber); err != nil {
+ return err
+ }
+
+ var rewriteValidationFiles []iceberg.DataFile
+ var referencedDataFilePaths []string
+ if cfg.rewriteSemantics && len(deleteFilesToAdd) > 0 {
+ rewriteValidationFiles = slices.Clone(dataFilesToDelete)
+ for ref := range setDeleteFilesToAdd.dvsByRef {
+ if _, added := setToAdd[ref]; added {
+ continue
+ }
+
+ state, ok := liveDataFiles[ref]
+ if !ok {
+ return fmt.Errorf("%w: deletion vector target %s is not live in the rewrite",
+ ErrInvalidOperation, ref)
+ }
+ rewriteValidationFiles = append(rewriteValidationFiles, state.file)
+ referencedDataFilePaths = append(referencedDataFilePaths, ref)
+ }
+ slices.Sort(referencedDataFilePaths)
+ }
+
if !cfg.skipAutoNameMapping {
if err := t.ensureNameMapping(); err != nil {
return err
@@ -1522,6 +2028,9 @@ func (t *Transaction) ReplaceFiles(ctx context.Context, dataFilesToDelete, dataF
// mergeOverwrite guarantees an *overwriteFiles producerImpl.
updater.producerImpl.(*overwriteFiles).skipDefaultValidator = true
}
+ if cfg.dataSequenceNumber != nil {
+ updater.setNewDataFilesDataSequenceNumber(*cfg.dataSequenceNumber)
+ }
for _, df := range markedDataForDeletion {
updater.deleteDataFile(df)
@@ -1529,9 +2038,19 @@ func (t *Transaction) ReplaceFiles(ctx context.Context, dataFilesToDelete, dataF
for _, df := range dataFilesToAdd {
updater.appendDataFile(df)
}
+ for _, addition := range deleteFilesToAdd {
+ sequenceNumber := deleteSequenceNumber
+ if addition.dataSequenceNumber != nil {
+ sequenceNumber = *addition.dataSequenceNumber
+ }
+ updater.appendDeleteFileWithDataSequenceNumber(addition.file, sequenceNumber)
+ }
for _, df := range markedDeleteForRemoval {
updater.removeDeleteFile(df)
}
+ for _, df := range markedAutoDeleteForRemoval {
+ updater.removeDeleteFile(df)
+ }
for _, df := range markedDVsForRemoval {
updater.removeDeletionVector(df)
}
@@ -1541,7 +2060,14 @@ func (t *Transaction) ReplaceFiles(ctx context.Context, dataFilesToDelete, dataF
return err
}
- return t.apply(updates, reqs)
+ if err := t.apply(updates, reqs); err != nil {
+ return err
+ }
+ if len(rewriteValidationFiles) > 0 {
+ t.addValidator(rewriteValidatorWithReferencedDataFiles(rewriteValidationFiles, referencedDataFilePaths))
+ }
+
+ return nil
}
type AddFilesOption func(addFilesOp *addFilesOperation)