This is an automated email from the ASF dual-hosted git repository.
laskoviymishka 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 c21d4b4e4 refactor(table)!: clarify path and DataFile replacement APIs
(#2306)
c21d4b4e4 is described below
commit c21d4b4e473b12e1e5d4fb766a88384cd2bd9ed7
Author: Minh Vu <[email protected]>
AuthorDate: Sun Oct 25 10:11:09 2026 +0100
refactor(table)!: clarify path and DataFile replacement APIs (#2306)
---
table/branch_planning_test.go | 8 ++--
table/data_file_meta.go | 2 +-
table/data_file_meta_test.go | 2 +-
table/dv_rewrite_test.go | 8 ++--
table/rewrite_files.go | 2 +-
table/rewrite_files_test.go | 4 +-
table/table_test.go | 96 +++++++++++++++++++-------------------
table/transaction.go | 27 ++++++-----
table/transaction_internal_test.go | 8 ++--
table/write_file_io_test.go | 2 +-
website/src/api.md | 2 +-
website/src/feature-status.md | 2 +-
website/src/releases.md | 2 +
13 files changed, 86 insertions(+), 79 deletions(-)
diff --git a/table/branch_planning_test.go b/table/branch_planning_test.go
index d61b117d8..cb8667ac8 100644
--- a/table/branch_planning_test.go
+++ b/table/branch_planning_test.go
@@ -463,7 +463,7 @@ func TestBranchAddFilesDuplicateCheckUsesBranchHead(t
*testing.T) {
// A table written only through a branch has no main head, so a main-based plan
// rejects the replace outright instead of performing it.
func TestReplaceOnBranchOnlyTable(t *testing.T) {
- t.Run("ReplaceDataFiles", func(t *testing.T) {
+ t.Run("ReplaceDataFilesByPath", func(t *testing.T) {
tbl := newBranchPlanningTable(t, nil)
tbl = appendRowsOnRef(t, tbl, branchPlanningRef, 1, 2)
require.Nil(t, tbl.CurrentSnapshot(), "fixture must leave main
without a head")
@@ -478,7 +478,7 @@ func TestReplaceOnBranchOnlyTable(t *testing.T) {
txn, err :=
tbl.NewTransactionOnBranchWithError(branchPlanningRef)
require.NoError(t, err)
- require.NoError(t, txn.ReplaceDataFiles(t.Context(), original,
[]string{replacement}, nil))
+ require.NoError(t, txn.ReplaceDataFilesByPath(t.Context(),
original, []string{replacement}, nil))
tbl, err = txn.Commit(t.Context())
require.NoError(t, err)
@@ -486,7 +486,7 @@ func TestReplaceOnBranchOnlyTable(t *testing.T) {
assert.Nil(t, tbl.CurrentSnapshot(), "replacing on the branch
must not create a main head")
})
- t.Run("ReplaceDataFilesWithDataFiles", func(t *testing.T) {
+ t.Run("ReplaceDataFiles", func(t *testing.T) {
tbl := newBranchPlanningTable(t, nil)
staged := buildDataFile(t,
tbl.Location()+"/data/branch-1.parquet")
@@ -500,7 +500,7 @@ func TestReplaceOnBranchOnlyTable(t *testing.T) {
replacement := buildDataFile(t,
tbl.Location()+"/data/branch-2.parquet")
txn, err =
tbl.NewTransactionOnBranchWithError(branchPlanningRef)
require.NoError(t, err)
- require.NoError(t,
txn.ReplaceDataFilesWithDataFiles(t.Context(),
+ require.NoError(t, txn.ReplaceDataFiles(t.Context(),
[]iceberg.DataFile{staged},
[]iceberg.DataFile{replacement}, nil))
tbl, err = txn.Commit(t.Context())
require.NoError(t, err)
diff --git a/table/data_file_meta.go b/table/data_file_meta.go
index 99a5fdff4..aabc710f0 100644
--- a/table/data_file_meta.go
+++ b/table/data_file_meta.go
@@ -119,7 +119,7 @@ type DataFileArgs struct {
// - [Transaction.AddDataFiles]: REQUIRED. These are
externally-written
// files, so the library cannot fabricate row IDs for them; a nil
// first_row_id is rejected at commit time.
- // - [Transaction.ReplaceDataFilesWithDataFiles]: REQUIRED for the
added
+ // - [Transaction.ReplaceDataFiles]: REQUIRED for the added
// files, except when the caller opts into rewrite/compaction
semantics
// (WithRewriteSemantics), where the manifest-list writer assigns
it by
// inheritance and it may be left nil.
diff --git a/table/data_file_meta_test.go b/table/data_file_meta_test.go
index 9ca145ed8..572cd02f4 100644
--- a/table/data_file_meta_test.go
+++ b/table/data_file_meta_test.go
@@ -1171,7 +1171,7 @@ func
TestDataFileFromMetadata_V3ReplaceDataFilesRequiresFirstRowID(t *testing.T)
require.Nil(t, replacement.FirstRowID())
tx = tbl.NewTransaction()
- err = tx.ReplaceDataFilesWithDataFiles(t.Context(),
+ err = tx.ReplaceDataFiles(t.Context(),
[]iceberg.DataFile{initial}, []iceberg.DataFile{replacement},
nil)
require.ErrorContains(t, err, "first_row_id")
require.ErrorContains(t, err, "required for v3")
diff --git a/table/dv_rewrite_test.go b/table/dv_rewrite_test.go
index 0da507fc2..028f461b6 100644
--- a/table/dv_rewrite_test.go
+++ b/table/dv_rewrite_test.go
@@ -529,11 +529,11 @@ func TestRemovingDataFileRemovesItsUnnamedDV(t
*testing.T) {
{"RewriteFiles", func(ctx context.Context, tx
*table.Transaction, dataB, dataC, _ iceberg.DataFile) error {
return
tx.NewRewrite(nil).DeleteFile(dataB).AddDataFile(dataC).Commit(ctx)
}},
- {"ReplaceDataFilesWithDataFiles", func(ctx context.Context, tx
*table.Transaction, dataB, dataC, _ iceberg.DataFile) error {
- return tx.ReplaceDataFilesWithDataFiles(ctx,
[]iceberg.DataFile{dataB}, []iceberg.DataFile{dataC}, nil)
+ {"ReplaceDataFiles", func(ctx context.Context, tx
*table.Transaction, dataB, dataC, _ iceberg.DataFile) error {
+ return tx.ReplaceDataFiles(ctx,
[]iceberg.DataFile{dataB}, []iceberg.DataFile{dataC}, nil)
}},
- {"ReplaceDataFiles", func(ctx context.Context, tx
*table.Transaction, dataB, _, _ iceberg.DataFile) error {
- return tx.ReplaceDataFiles(ctx,
[]string{dataB.FilePath()}, nil, nil)
+ {"ReplaceDataFilesByPath", func(ctx context.Context, tx
*table.Transaction, dataB, _, _ iceberg.DataFile) error {
+ return tx.ReplaceDataFilesByPath(ctx,
[]string{dataB.FilePath()}, nil, nil)
}},
{"ReplaceFiles", func(ctx context.Context, tx
*table.Transaction, dataB, dataC, eqDelete iceberg.DataFile) error {
return tx.ReplaceFiles(ctx, []iceberg.DataFile{dataB},
[]iceberg.DataFile{dataC}, []iceberg.DataFile{eqDelete}, nil)
diff --git a/table/rewrite_files.go b/table/rewrite_files.go
index c9d26b863..172510cc2 100644
--- a/table/rewrite_files.go
+++ b/table/rewrite_files.go
@@ -327,7 +327,7 @@ func (r *RewriteFiles) Commit(ctx context.Context) error {
return fmt.Errorf("%w: rewrite must have at least one file
change", ErrInvalidOperation)
}
// Adds-without-deletes would route through ReplaceFiles →
- // ReplaceDataFilesWithDataFiles → AddDataFiles, an OpAppend
+ // ReplaceDataFiles → AddDataFiles, an OpAppend
// producer that never reads cfg.rewriteSemantics. The snapshot
// would be tagged append with no rewrite validator — silently
// wrong for a rewrite. A pure delete-file expunge (only
diff --git a/table/rewrite_files_test.go b/table/rewrite_files_test.go
index 6c4e3e5f2..5b6d21d5d 100644
--- a/table/rewrite_files_test.go
+++ b/table/rewrite_files_test.go
@@ -408,7 +408,7 @@ func TestRewriteFiles_DeleteFile_RoutesByContentType(t
*testing.T) {
//
// Builder A — DeleteFile(strangerData): if routed correctly, the
// transaction has an empty deleteFilesToRemove slice and falls
- // through to ReplaceDataFilesWithDataFiles, which rejects with
+ // through to ReplaceDataFiles, which rejects with
// "cannot delete files that do not belong to the table". A
// mis-route into deleteFilesToRemove would error with "cannot
// remove delete files that do not belong to the table" instead.
@@ -502,7 +502,7 @@ func TestAddFiles_ReAddsPathDeletedByRewrite(t *testing.T) {
// TestRewriteFiles_RejectsAddsWithoutDataDeletes proves that staging
// new data files with no data files to delete is rejected up front —
// otherwise the chain would slip through ReplaceFiles →
-// ReplaceDataFilesWithDataFiles → AddDataFiles (an OpAppend producer)
+// ReplaceDataFiles → AddDataFiles (an OpAppend producer)
// and silently tag the snapshot append instead of replace, with no
// rewrite validator registered.
func TestRewriteFiles_RejectsAddsWithoutDataDeletes(t *testing.T) {
diff --git a/table/table_test.go b/table/table_test.go
index a58f70b2a..c00672a4e 100644
--- a/table/table_test.go
+++ b/table/table_test.go
@@ -1111,7 +1111,7 @@ func (m *mockedCatalog) CommitTable(ctx context.Context,
ident table.Identifier,
return meta, "", nil
}
-func (t *TableWritingTestSuite) TestReplaceDataFiles() {
+func (t *TableWritingTestSuite) TestReplaceDataFilesByPath() {
fs := iceio.LocalFS{}
files := make([]string, 0)
@@ -1169,7 +1169,7 @@ func (t *TableWritingTestSuite) TestReplaceDataFiles() {
t.writeParquet(fs, combinedFilePath, combined)
tx := tbl.NewTransaction()
- t.Require().NoError(tx.ReplaceDataFiles(ctx, files[:2],
[]string{combinedFilePath}, nil))
+ t.Require().NoError(tx.ReplaceDataFilesByPath(ctx, files[:2],
[]string{combinedFilePath}, nil))
staged, err := tx.StagedTable()
t.Require().NoError(err)
@@ -1383,7 +1383,7 @@ func (t *TableWritingTestSuite)
TestAddDataFilesAutoNameMapping() {
}
}
-func (t *TableWritingTestSuite)
TestReplaceDataFilesWithDataFilesAutoNameMapping() {
+func (t *TableWritingTestSuite) TestReplaceDataFilesAutoNameMapping() {
for _, tc := range []struct {
name string
opts []table.WriteOption
@@ -1416,7 +1416,7 @@ func (t *TableWritingTestSuite)
TestReplaceDataFilesWithDataFilesAutoNameMapping
addFile := t.mustDataFile(*iceberg.UnpartitionedSpec,
replacementPath, nil, 1, mustFileSize(t.T(), replacementPath))
tx = tbl.NewTransaction()
-
t.Require().NoError(tx.ReplaceDataFilesWithDataFiles(t.ctx,
[]iceberg.DataFile{deleteFile}, []iceberg.DataFile{addFile}, nil, tc.opts...))
+ t.Require().NoError(tx.ReplaceDataFiles(t.ctx,
[]iceberg.DataFile{deleteFile}, []iceberg.DataFile{addFile}, nil, tc.opts...))
staged, err := tx.StagedTable()
t.Require().NoError(err)
@@ -1426,7 +1426,7 @@ func (t *TableWritingTestSuite)
TestReplaceDataFilesWithDataFilesAutoNameMapping
}
}
-func (t *TableWritingTestSuite) TestReplaceDataFilesWithDataFiles() {
+func (t *TableWritingTestSuite) TestReplaceDataFiles() {
ident := table.Identifier{"default",
"replace_data_files_with_datafiles_v" + strconv.Itoa(t.formatVersion)}
tbl := t.createTable(ident, t.formatVersion,
*iceberg.UnpartitionedSpec, t.tableSchema)
@@ -1452,7 +1452,7 @@ func (t *TableWritingTestSuite)
TestReplaceDataFilesWithDataFiles() {
addFile := t.mustDataFile(*iceberg.UnpartitionedSpec, replacementPath,
nil, 1, mustFileSize(t.T(), replacementPath))
tx = tbl.NewTransaction()
- t.Require().NoError(tx.ReplaceDataFilesWithDataFiles(t.ctx,
deleteFiles, []iceberg.DataFile{addFile}, nil))
+ t.Require().NoError(tx.ReplaceDataFiles(t.ctx, deleteFiles,
[]iceberg.DataFile{addFile}, nil))
staged, err := tx.StagedTable()
t.Require().NoError(err)
@@ -1461,7 +1461,7 @@ func (t *TableWritingTestSuite)
TestReplaceDataFilesWithDataFiles() {
t.Equal("2",
staged.CurrentSnapshot().Summary.Properties["deleted-data-files"])
}
-func (t *TableWritingTestSuite)
TestReplaceDataFilesWithDataFilesDoesNotCarryEmptyManifests() {
+func (t *TableWritingTestSuite)
TestReplaceDataFilesDoesNotCarryEmptyManifests() {
ident := table.Identifier{"default",
"replace_data_files_drops_empty_manifests_v" + strconv.Itoa(t.formatVersion)}
tbl := t.createTable(ident, t.formatVersion,
*iceberg.UnpartitionedSpec, t.tableSchema)
@@ -1482,7 +1482,7 @@ func (t *TableWritingTestSuite)
TestReplaceDataFilesWithDataFilesDoesNotCarryEmp
for i := 1; i <= 5; i++ {
next := writeDataFile(fmt.Sprintf("data-%d", i))
tx = tbl.NewTransaction()
- t.Require().NoError(tx.ReplaceDataFilesWithDataFiles(t.ctx,
[]iceberg.DataFile{current}, []iceberg.DataFile{next}, nil))
+ t.Require().NoError(tx.ReplaceDataFiles(t.ctx,
[]iceberg.DataFile{current}, []iceberg.DataFile{next}, nil))
tbl, err = tx.Commit(t.ctx)
t.Require().NoError(err)
current = next
@@ -1510,7 +1510,7 @@ func (t *TableWritingTestSuite)
TestReplaceDataFilesWithDataFilesDoesNotCarryEmp
t.Equal(1, emptyCurrentManifests)
}
-func (t *TableWritingTestSuite)
TestReplaceDataFilesWithDataFilesValidatesPartitionSpecID() {
+func (t *TableWritingTestSuite) TestReplaceDataFilesValidatesPartitionSpecID()
{
ident := table.Identifier{"default",
"replace_data_files_with_datafiles_spec_validation_v" +
strconv.Itoa(t.formatVersion)}
spec := iceberg.NewPartitionSpec(
iceberg.PartitionField{SourceIDs: []int{4}, FieldID: 1000,
Transform: iceberg.IdentityTransform{}, Name: "baz"},
@@ -1535,12 +1535,12 @@ func (t *TableWritingTestSuite)
TestReplaceDataFilesWithDataFilesValidatesPartit
map[int]any{1000: int32(123)}, 1, 1)
tx = tbl.NewTransaction()
- err = tx.ReplaceDataFilesWithDataFiles(t.ctx,
[]iceberg.DataFile{deleteFile}, []iceberg.DataFile{addFile}, nil)
+ err = tx.ReplaceDataFiles(t.ctx, []iceberg.DataFile{deleteFile},
[]iceberg.DataFile{addFile}, nil)
t.Error(err)
t.ErrorContains(err, "unregistered partition spec id 999")
}
-func (t *TableWritingTestSuite)
TestReplaceDataFilesWithDataFilesValidatesPartitionData() {
+func (t *TableWritingTestSuite) TestReplaceDataFilesValidatesPartitionData() {
ident := table.Identifier{"default",
"replace_data_files_with_datafiles_partition_validation_v" +
strconv.Itoa(t.formatVersion)}
spec := iceberg.NewPartitionSpec(
iceberg.PartitionField{SourceIDs: []int{4}, FieldID: 1000,
Transform: iceberg.IdentityTransform{}, Name: "baz"},
@@ -1561,7 +1561,7 @@ func (t *TableWritingTestSuite)
TestReplaceDataFilesWithDataFilesValidatesPartit
map[int]any{}, 1, 1)
tx = tbl.NewTransaction()
- err = tx.ReplaceDataFilesWithDataFiles(t.ctx,
[]iceberg.DataFile{deleteFile}, []iceberg.DataFile{addFile}, nil)
+ err = tx.ReplaceDataFiles(t.ctx, []iceberg.DataFile{deleteFile},
[]iceberg.DataFile{addFile}, nil)
t.Error(err)
t.ErrorContains(err, "missing partition value")
}
@@ -1675,10 +1675,10 @@ func (t *TableWritingTestSuite)
TestAddDataFilesEmptySlice() {
}
// ============================================================================
-// ReplaceDataFilesWithDataFiles Error Path Tests
+// ReplaceDataFiles Error Path Tests
// ============================================================================
-func (t *TableWritingTestSuite)
TestReplaceDataFilesWithDataFilesDuplicateAddPaths() {
+func (t *TableWritingTestSuite) TestReplaceDataFilesDuplicateAddPaths() {
ident := table.Identifier{"default", "replace_data_files_dup_add_v" +
strconv.Itoa(t.formatVersion)}
tbl := t.createTable(ident, t.formatVersion,
*iceberg.UnpartitionedSpec, t.tableSchema)
@@ -1695,12 +1695,12 @@ func (t *TableWritingTestSuite)
TestReplaceDataFilesWithDataFilesDuplicateAddPat
fmt.Sprintf("%s/replace_data_files_dup_add_v%d/new.parquet",
t.location, t.formatVersion), nil, 1, 100)
tx = tbl.NewTransaction()
- err = tx.ReplaceDataFilesWithDataFiles(t.ctx,
[]iceberg.DataFile{deleteFile}, []iceberg.DataFile{addFile, addFile}, nil)
+ err = tx.ReplaceDataFiles(t.ctx, []iceberg.DataFile{deleteFile},
[]iceberg.DataFile{addFile, addFile}, nil)
t.Error(err)
- t.ErrorContains(err, "add data file paths must be unique for
ReplaceDataFilesWithDataFiles")
+ t.ErrorContains(err, "add data file paths must be unique for
ReplaceDataFiles")
}
-func (t *TableWritingTestSuite)
TestReplaceDataFilesWithDataFilesDuplicateDeletePaths() {
+func (t *TableWritingTestSuite) TestReplaceDataFilesDuplicateDeletePaths() {
ident := table.Identifier{"default", "replace_data_files_dup_del_v" +
strconv.Itoa(t.formatVersion)}
tbl := t.createTable(ident, t.formatVersion,
*iceberg.UnpartitionedSpec, t.tableSchema)
@@ -1717,12 +1717,12 @@ func (t *TableWritingTestSuite)
TestReplaceDataFilesWithDataFilesDuplicateDelete
fmt.Sprintf("%s/replace_data_files_dup_del_v%d/new.parquet",
t.location, t.formatVersion), nil, 1, 100)
tx = tbl.NewTransaction()
- err = tx.ReplaceDataFilesWithDataFiles(t.ctx,
[]iceberg.DataFile{deleteFile, deleteFile}, []iceberg.DataFile{addFile}, nil)
+ err = tx.ReplaceDataFiles(t.ctx, []iceberg.DataFile{deleteFile,
deleteFile}, []iceberg.DataFile{addFile}, nil)
t.Error(err)
- t.ErrorContains(err, "delete data file paths must be unique for
ReplaceDataFilesWithDataFiles")
+ t.ErrorContains(err, "delete data file paths must be unique for
ReplaceDataFiles")
}
-func (t *TableWritingTestSuite)
TestReplaceDataFilesWithDataFilesAddAlreadyReferenced() {
+func (t *TableWritingTestSuite) TestReplaceDataFilesAddAlreadyReferenced() {
ident := table.Identifier{"default", "replace_data_files_add_ref_v" +
strconv.Itoa(t.formatVersion)}
tbl := t.createTable(ident, t.formatVersion,
*iceberg.UnpartitionedSpec, t.tableSchema)
@@ -1740,12 +1740,12 @@ func (t *TableWritingTestSuite)
TestReplaceDataFilesWithDataFilesAddAlreadyRefer
addFile := t.mustDataFile(*iceberg.UnpartitionedSpec, existingPath2,
nil, 1, mustFileSize(t.T(), existingPath2))
tx = tbl.NewTransaction()
- err = tx.ReplaceDataFilesWithDataFiles(t.ctx,
[]iceberg.DataFile{deleteFile}, []iceberg.DataFile{addFile}, nil)
+ err = tx.ReplaceDataFiles(t.ctx, []iceberg.DataFile{deleteFile},
[]iceberg.DataFile{addFile}, nil)
t.Error(err)
t.ErrorContains(err, "cannot add files that are already referenced by
table")
}
-func (t *TableWritingTestSuite)
TestReplaceDataFilesWithDataFilesDeleteNotInTable() {
+func (t *TableWritingTestSuite) TestReplaceDataFilesDeleteNotInTable() {
ident := table.Identifier{"default",
"replace_data_files_del_not_found_v" + strconv.Itoa(t.formatVersion)}
tbl := t.createTable(ident, t.formatVersion,
*iceberg.UnpartitionedSpec, t.tableSchema)
@@ -1763,12 +1763,12 @@ func (t *TableWritingTestSuite)
TestReplaceDataFilesWithDataFilesDeleteNotInTabl
fmt.Sprintf("%s/replace_data_files_del_not_found_v%d/new.parquet", t.location,
t.formatVersion), nil, 1, 100)
tx = tbl.NewTransaction()
- err = tx.ReplaceDataFilesWithDataFiles(t.ctx,
[]iceberg.DataFile{nonExistentFile}, []iceberg.DataFile{addFile}, nil)
+ err = tx.ReplaceDataFiles(t.ctx, []iceberg.DataFile{nonExistentFile},
[]iceberg.DataFile{addFile}, nil)
t.Error(err)
t.ErrorContains(err, "cannot delete files that do not belong to the
table")
}
-func (t *TableWritingTestSuite)
TestReplaceDataFilesWithDataFilesNilDeleteFile() {
+func (t *TableWritingTestSuite) TestReplaceDataFilesNilDeleteFile() {
ident := table.Identifier{"default", "replace_data_files_nil_del_v" +
strconv.Itoa(t.formatVersion)}
tbl := t.createTable(ident, t.formatVersion,
*iceberg.UnpartitionedSpec, t.tableSchema)
@@ -1784,12 +1784,12 @@ func (t *TableWritingTestSuite)
TestReplaceDataFilesWithDataFilesNilDeleteFile()
fmt.Sprintf("%s/replace_data_files_nil_del_v%d/new.parquet",
t.location, t.formatVersion), nil, 1, 100)
tx = tbl.NewTransaction()
- err = tx.ReplaceDataFilesWithDataFiles(t.ctx, []iceberg.DataFile{nil},
[]iceberg.DataFile{addFile}, nil)
+ err = tx.ReplaceDataFiles(t.ctx, []iceberg.DataFile{nil},
[]iceberg.DataFile{addFile}, nil)
t.Error(err)
t.ErrorContains(err, "nil data file at index 0")
}
-func (t *TableWritingTestSuite) TestReplaceDataFilesWithDataFilesNilAddFile() {
+func (t *TableWritingTestSuite) TestReplaceDataFilesNilAddFile() {
ident := table.Identifier{"default", "replace_data_files_nil_add_v" +
strconv.Itoa(t.formatVersion)}
tbl := t.createTable(ident, t.formatVersion,
*iceberg.UnpartitionedSpec, t.tableSchema)
@@ -1804,12 +1804,12 @@ func (t *TableWritingTestSuite)
TestReplaceDataFilesWithDataFilesNilAddFile() {
deleteFile := t.mustDataFile(*iceberg.UnpartitionedSpec, existingPath,
nil, 1, mustFileSize(t.T(), existingPath))
tx = tbl.NewTransaction()
- err = tx.ReplaceDataFilesWithDataFiles(t.ctx,
[]iceberg.DataFile{deleteFile}, []iceberg.DataFile{nil}, nil)
+ err = tx.ReplaceDataFiles(t.ctx, []iceberg.DataFile{deleteFile},
[]iceberg.DataFile{nil}, nil)
t.Error(err)
t.ErrorContains(err, "nil data file at index 0")
}
-func (t *TableWritingTestSuite) TestReplaceDataFilesWithDataFilesNoSnapshot() {
+func (t *TableWritingTestSuite) TestReplaceDataFilesNoSnapshot() {
ident := table.Identifier{"default", "replace_data_files_no_snapshot_v"
+ strconv.Itoa(t.formatVersion)}
tbl := t.createTable(ident, t.formatVersion,
*iceberg.UnpartitionedSpec, t.tableSchema)
@@ -1819,12 +1819,12 @@ func (t *TableWritingTestSuite)
TestReplaceDataFilesWithDataFilesNoSnapshot() {
fmt.Sprintf("%s/replace_data_files_no_snapshot_v%d/add.parquet", t.location,
t.formatVersion), nil, 1, 100)
tx := tbl.NewTransaction()
- err := tx.ReplaceDataFilesWithDataFiles(t.ctx,
[]iceberg.DataFile{deleteFile}, []iceberg.DataFile{addFile}, nil)
+ err := tx.ReplaceDataFiles(t.ctx, []iceberg.DataFile{deleteFile},
[]iceberg.DataFile{addFile}, nil)
t.Error(err)
t.ErrorContains(err, "cannot replace files in a table without an
existing snapshot")
}
-func (t *TableWritingTestSuite)
TestReplaceDataFilesWithDataFilesInvalidAddContentType() {
+func (t *TableWritingTestSuite) TestReplaceDataFilesInvalidAddContentType() {
ident := table.Identifier{"default",
"replace_data_files_invalid_add_content_v" + strconv.Itoa(t.formatVersion)}
tbl := t.createTable(ident, t.formatVersion,
*iceberg.UnpartitionedSpec, t.tableSchema)
@@ -1845,17 +1845,17 @@ func (t *TableWritingTestSuite)
TestReplaceDataFilesWithDataFilesInvalidAddConte
invalidAddFile := builder.Build()
tx = tbl.NewTransaction()
- err = tx.ReplaceDataFilesWithDataFiles(t.ctx,
[]iceberg.DataFile{deleteFile}, []iceberg.DataFile{invalidAddFile}, nil)
+ err = tx.ReplaceDataFiles(t.ctx, []iceberg.DataFile{deleteFile},
[]iceberg.DataFile{invalidAddFile}, nil)
t.Error(err)
t.ErrorContains(err, "adding files other than data files is not yet
implemented")
}
-func (t *TableWritingTestSuite) TestReplaceDataFilesWithDataFilesEmptyBoth() {
+func (t *TableWritingTestSuite) TestReplaceDataFilesEmptyBoth() {
ident := table.Identifier{"default", "replace_data_files_empty_both_v"
+ strconv.Itoa(t.formatVersion)}
tbl := t.createTable(ident, t.formatVersion,
*iceberg.UnpartitionedSpec, t.tableSchema)
tx := tbl.NewTransaction()
- err := tx.ReplaceDataFilesWithDataFiles(t.ctx, []iceberg.DataFile{},
[]iceberg.DataFile{}, nil)
+ err := tx.ReplaceDataFiles(t.ctx, []iceberg.DataFile{},
[]iceberg.DataFile{}, nil)
t.NoError(err)
staged, err := tx.StagedTable()
@@ -1863,7 +1863,7 @@ func (t *TableWritingTestSuite)
TestReplaceDataFilesWithDataFilesEmptyBoth() {
t.Nil(staged.CurrentSnapshot())
}
-func (t *TableWritingTestSuite)
TestReplaceDataFilesWithDataFilesDelegatesToAddDataFiles() {
+func (t *TableWritingTestSuite) TestReplaceDataFilesDelegatesToAddDataFiles() {
ident := table.Identifier{"default", "replace_data_files_delegate_v" +
strconv.Itoa(t.formatVersion)}
tbl := t.createTable(ident, t.formatVersion,
*iceberg.UnpartitionedSpec, t.tableSchema)
@@ -1873,7 +1873,7 @@ func (t *TableWritingTestSuite)
TestReplaceDataFilesWithDataFilesDelegatesToAddD
addFile := t.mustDataFile(*iceberg.UnpartitionedSpec, filePath, nil, 1,
mustFileSize(t.T(), filePath))
tx := tbl.NewTransaction()
- t.Require().NoError(tx.ReplaceDataFilesWithDataFiles(t.ctx,
[]iceberg.DataFile{}, []iceberg.DataFile{addFile}, nil))
+ t.Require().NoError(tx.ReplaceDataFiles(t.ctx, []iceberg.DataFile{},
[]iceberg.DataFile{addFile}, nil))
staged, err := tx.StagedTable()
t.Require().NoError(err)
@@ -1882,10 +1882,10 @@ func (t *TableWritingTestSuite)
TestReplaceDataFilesWithDataFilesDelegatesToAddD
}
// ============================================================================
-// ReplaceDataFiles (string paths) Error Path Tests
+// ReplaceDataFilesByPath (string paths) Error Path Tests
// ============================================================================
-func (t *TableWritingTestSuite) TestReplaceDataFilesDuplicateDeletePaths() {
+func (t *TableWritingTestSuite)
TestReplaceDataFilesByPathDuplicateDeletePaths() {
ident := table.Identifier{"default",
"replace_data_files_dup_del_paths_v" + strconv.Itoa(t.formatVersion)}
tbl := t.createTable(ident, t.formatVersion,
*iceberg.UnpartitionedSpec, t.tableSchema)
@@ -1901,12 +1901,12 @@ func (t *TableWritingTestSuite)
TestReplaceDataFilesDuplicateDeletePaths() {
t.writeParquet(mustFS(t.T(), tbl).(iceio.WriteFileIO), newPath,
t.arrTbl)
tx = tbl.NewTransaction()
- err = tx.ReplaceDataFiles(t.ctx, []string{existingPath, existingPath},
[]string{newPath}, nil)
+ err = tx.ReplaceDataFilesByPath(t.ctx, []string{existingPath,
existingPath}, []string{newPath}, nil)
t.Error(err)
- t.ErrorContains(err, "delete file paths must be unique for
ReplaceDataFiles")
+ t.ErrorContains(err, "delete file paths must be unique for
ReplaceDataFilesByPath")
}
-func (t *TableWritingTestSuite) TestReplaceDataFilesDuplicateAddPaths() {
+func (t *TableWritingTestSuite) TestReplaceDataFilesByPathDuplicateAddPaths() {
ident := table.Identifier{"default",
"replace_data_files_dup_add_paths_v" + strconv.Itoa(t.formatVersion)}
tbl := t.createTable(ident, t.formatVersion,
*iceberg.UnpartitionedSpec, t.tableSchema)
@@ -1922,12 +1922,12 @@ func (t *TableWritingTestSuite)
TestReplaceDataFilesDuplicateAddPaths() {
t.writeParquet(mustFS(t.T(), tbl).(iceio.WriteFileIO), newPath,
t.arrTbl)
tx = tbl.NewTransaction()
- err = tx.ReplaceDataFiles(t.ctx, []string{existingPath},
[]string{newPath, newPath}, nil)
+ err = tx.ReplaceDataFilesByPath(t.ctx, []string{existingPath},
[]string{newPath, newPath}, nil)
t.Error(err)
- t.ErrorContains(err, "add file paths must be unique for
ReplaceDataFiles")
+ t.ErrorContains(err, "add file paths must be unique for
ReplaceDataFilesByPath")
}
-func (t *TableWritingTestSuite) TestReplaceDataFilesAddAlreadyReferenced() {
+func (t *TableWritingTestSuite)
TestReplaceDataFilesByPathAddAlreadyReferenced() {
ident := table.Identifier{"default",
"replace_data_files_add_already_ref_v" + strconv.Itoa(t.formatVersion)}
tbl := t.createTable(ident, t.formatVersion,
*iceberg.UnpartitionedSpec, t.tableSchema)
@@ -1942,12 +1942,12 @@ func (t *TableWritingTestSuite)
TestReplaceDataFilesAddAlreadyReferenced() {
t.Require().NoError(err)
tx = tbl.NewTransaction()
- err = tx.ReplaceDataFiles(t.ctx, []string{existingPath1},
[]string{existingPath2}, nil)
+ err = tx.ReplaceDataFilesByPath(t.ctx, []string{existingPath1},
[]string{existingPath2}, nil)
t.Error(err)
t.ErrorContains(err, "cannot add files that are already referenced by
table")
}
-func (t *TableWritingTestSuite) TestReplaceDataFilesDeleteNotInTable() {
+func (t *TableWritingTestSuite) TestReplaceDataFilesByPathDeleteNotInTable() {
ident := table.Identifier{"default",
"replace_data_files_del_not_in_table_v" + strconv.Itoa(t.formatVersion)}
tbl := t.createTable(ident, t.formatVersion,
*iceberg.UnpartitionedSpec, t.tableSchema)
@@ -1965,17 +1965,17 @@ func (t *TableWritingTestSuite)
TestReplaceDataFilesDeleteNotInTable() {
nonExistentPath :=
fmt.Sprintf("%s/replace_data_files_del_not_in_table_v%d/nonexistent.parquet",
t.location, t.formatVersion)
tx = tbl.NewTransaction()
- err = tx.ReplaceDataFiles(t.ctx, []string{nonExistentPath},
[]string{newPath}, nil)
+ err = tx.ReplaceDataFilesByPath(t.ctx, []string{nonExistentPath},
[]string{newPath}, nil)
t.Error(err)
t.ErrorContains(err, "cannot delete files that do not belong to the
table")
}
-func (t *TableWritingTestSuite) TestReplaceDataFilesNoSnapshot() {
+func (t *TableWritingTestSuite) TestReplaceDataFilesByPathNoSnapshot() {
ident := table.Identifier{"default",
"replace_data_files_no_snapshot_paths_v" + strconv.Itoa(t.formatVersion)}
tbl := t.createTable(ident, t.formatVersion,
*iceberg.UnpartitionedSpec, t.tableSchema)
tx := tbl.NewTransaction()
- err := tx.ReplaceDataFiles(t.ctx, []string{"delete.parquet"},
[]string{"add.parquet"}, nil)
+ err := tx.ReplaceDataFilesByPath(t.ctx, []string{"delete.parquet"},
[]string{"add.parquet"}, nil)
t.Error(err)
t.ErrorContains(err, "cannot replace files in a table without an
existing snapshot")
}
diff --git a/table/transaction.go b/table/transaction.go
index 2faa2ec01..e8369e6f9 100644
--- a/table/transaction.go
+++ b/table/transaction.go
@@ -927,7 +927,10 @@ func (t *Transaction) Append(ctx context.Context, rdr
array.RecordReader, snapsh
return t.apply(updates, reqs)
}
-// ReplaceFiles is actually just an overwrite operation with multiple
+// ReplaceDataFilesByPath replaces data files identified by their storage
paths.
+// Unlike ReplaceDataFiles, it loads file metadata as needed.
+//
+// It is actually an overwrite operation with multiple
// files deleted and added.
//
// TODO: technically, this could be a REPLACE operation but we aren't
performing
@@ -937,7 +940,7 @@ func (t *Transaction) Append(ctx context.Context, rdr
array.RecordReader, snapsh
// For now, we'll keep using an overwrite operation.
//
// On v3 a removed data file's live deletion vector is removed with it.
-func (t *Transaction) ReplaceDataFiles(ctx context.Context, filesToDelete,
filesToAdd []string, snapshotProps iceberg.Properties) error {
+func (t *Transaction) ReplaceDataFilesByPath(ctx context.Context,
filesToDelete, filesToAdd []string, snapshotProps iceberg.Properties) error {
meta, err := t.txnMeta()
if err != nil {
return err
@@ -962,11 +965,11 @@ func (t *Transaction) ReplaceDataFiles(ctx
context.Context, filesToDelete, files
}
if len(setToDelete) != len(filesToDelete) {
- return errors.New("delete file paths must be unique for
ReplaceDataFiles")
+ return errors.New("delete file paths must be unique for
ReplaceDataFilesByPath")
}
if len(setToAdd) != len(filesToAdd) {
- return errors.New("add file paths must be unique for
ReplaceDataFiles")
+ return errors.New("add file paths must be unique for
ReplaceDataFilesByPath")
}
s := t.planningSnapshot(meta)
@@ -1508,7 +1511,7 @@ func withDataSequenceNumber(seq int64) WriteOption {
// 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
+// and [Transaction.ReplaceDataFiles] will set the name mapping if
// one does not already exist. This option is useful when working with catalogs
// (such as Databricks Unity Catalog) that reject the name mapping property.
func WithoutAutoNameMapping() WriteOption {
@@ -1644,7 +1647,7 @@ func (t *Transaction) AddDataFiles(ctx context.Context,
dataFiles []iceberg.Data
return t.apply(updates, reqs)
}
-// ReplaceDataFilesWithDataFiles replaces files using pre-built DataFile
objects.
+// ReplaceDataFiles replaces files using pre-built DataFile objects.
// This avoids scanning files to extract schema and statistics - the caller
provides
// DataFile objects directly with all required metadata.
//
@@ -1668,7 +1671,7 @@ func (t *Transaction) AddDataFiles(ctx context.Context,
dataFiles []iceberg.Data
// - Files are written via a separate I/O path and metadata is already known
// - Avoiding file scanning improves performance or reliability
// - Working with storage systems where immediate file reads may be
unreliable
-func (t *Transaction) ReplaceDataFilesWithDataFiles(ctx context.Context,
filesToDelete, filesToAdd []iceberg.DataFile, snapshotProps iceberg.Properties,
opts ...WriteOption) error {
+func (t *Transaction) ReplaceDataFiles(ctx context.Context, filesToDelete,
filesToAdd []iceberg.DataFile, snapshotProps iceberg.Properties, opts
...WriteOption) error {
meta, err := t.txnMeta()
if err != nil {
return err
@@ -1686,7 +1689,7 @@ func (t *Transaction) ReplaceDataFilesWithDataFiles(ctx
context.Context, filesTo
o(&cfg)
}
- setToAdd, err := t.validateDataFilesToAdd(filesToAdd,
"ReplaceDataFilesWithDataFiles", !cfg.rewriteSemantics)
+ setToAdd, err := t.validateDataFilesToAdd(filesToAdd,
"ReplaceDataFiles", !cfg.rewriteSemantics)
if err != nil {
return err
}
@@ -1694,16 +1697,16 @@ func (t *Transaction) ReplaceDataFilesWithDataFiles(ctx
context.Context, filesTo
setToDelete := make(map[string]struct{}, len(filesToDelete))
for i, df := range filesToDelete {
if df == nil {
- return fmt.Errorf("nil data file at index %d for
ReplaceDataFilesWithDataFiles", i)
+ return fmt.Errorf("nil data file at index %d for
ReplaceDataFiles", i)
}
path := df.FilePath()
if path == "" {
- return errors.New("delete data file paths must be
non-empty for ReplaceDataFilesWithDataFiles")
+ return errors.New("delete data file paths must be
non-empty for ReplaceDataFiles")
}
if _, ok := setToDelete[path]; ok {
- return errors.New("delete data file paths must be
unique for ReplaceDataFilesWithDataFiles")
+ return errors.New("delete data file paths must be
unique for ReplaceDataFiles")
}
setToDelete[path] = struct{}{}
}
@@ -1843,7 +1846,7 @@ func (t *Transaction) replaceFiles(ctx context.Context,
dataFilesToDelete, dataF
// 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...)
+ return t.ReplaceDataFiles(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",
diff --git a/table/transaction_internal_test.go
b/table/transaction_internal_test.go
index 4a26c07b1..0b09ed835 100644
--- a/table/transaction_internal_test.go
+++ b/table/transaction_internal_test.go
@@ -1174,9 +1174,11 @@ func
TestTransactionEntryPointsRejectNilAndBrokenTransaction(t *testing.T) {
{"Append", func(txn *Transaction) error { return
txn.Append(ctx, nil, nil) }},
{"AddDataFiles", func(txn *Transaction) error { return
txn.AddDataFiles(ctx, []iceberg.DataFile{dataFile}, nil) }},
{"AddFiles", func(txn *Transaction) error { return
txn.AddFiles(ctx, []string{"file://a.parquet"}, nil, false) }},
- {"ReplaceDataFiles", func(txn *Transaction) error { return
txn.ReplaceDataFiles(ctx, []string{"a"}, []string{"b"}, nil) }},
- {"ReplaceDataFilesWithDataFiles", func(txn *Transaction) error {
- return txn.ReplaceDataFilesWithDataFiles(ctx,
[]iceberg.DataFile{dataFile}, nil, nil)
+ {"ReplaceDataFilesByPath", func(txn *Transaction) error {
+ return txn.ReplaceDataFilesByPath(ctx, []string{"a"},
[]string{"b"}, nil)
+ }},
+ {"ReplaceDataFiles", func(txn *Transaction) error {
+ return txn.ReplaceDataFiles(ctx,
[]iceberg.DataFile{dataFile}, nil, nil)
}},
{"ReplaceFiles", func(txn *Transaction) error {
return txn.ReplaceFiles(ctx,
[]iceberg.DataFile{dataFile}, nil, []iceberg.DataFile{dataFile}, nil)
diff --git a/table/write_file_io_test.go b/table/write_file_io_test.go
index 3acd52a91..13e3af85b 100644
--- a/table/write_file_io_test.go
+++ b/table/write_file_io_test.go
@@ -150,7 +150,7 @@ func TestWritePathsReturnErrorForReadOnlyIO(t *testing.T) {
{
name: "replace data files with data files add-only
path",
run: func(ctx context.Context, tbl *Table) error {
- return
tbl.NewTransaction().ReplaceDataFilesWithDataFiles(ctx, nil, []iceberg.DataFile{
+ return
tbl.NewTransaction().ReplaceDataFiles(ctx, nil, []iceberg.DataFile{
newWriteIORequiredDataFileWithPath(t,
"file:///tmp/read-only-write/replace-data-files-add.parquet"),
}, nil)
},
diff --git a/website/src/api.md b/website/src/api.md
index 0f00ddd37..b1c62e370 100644
--- a/website/src/api.md
+++ b/website/src/api.md
@@ -362,7 +362,7 @@ if err != nil { /* ... */ }
newTbl, err := txn.Commit(ctx)
```
-`ReplaceDataFiles(ctx, filesToDelete, filesToAdd, snapshotProps)` and
`ReplaceDataFilesWithDataFiles(ctx, filesToDelete, dataFilesToAdd,
snapshotProps, opts...)` are also available on `*Transaction` for swapping
files atomically.
+`ReplaceDataFilesByPath(ctx, filesToDelete, filesToAdd, snapshotProps)` and
`ReplaceDataFiles(ctx, filesToDelete, dataFilesToAdd, snapshotProps, opts...)`
are also available on `*Transaction` for swapping files atomically.
### Transactions
diff --git a/website/src/feature-status.md b/website/src/feature-status.md
index 5802da8e0..9021bcbbf 100644
--- a/website/src/feature-status.md
+++ b/website/src/feature-status.md
@@ -39,7 +39,7 @@ All V1 features are supported. V1 is the format-version
baseline.
| Equality deletes | Supported (read + write). Write via
`Transaction.WriteEqualityDeletes`; row-level commits via
`Transaction.NewRowDelta`; delete-only CDC via
`Transaction.CommitEqualityDeletes` |
| Partition spec evolution | Supported |
| Sort order enforcement on write | Supported (PR
[#1157](https://github.com/apache/iceberg-go/pull/1157), closes
[#833](https://github.com/apache/iceberg-go/issues/833)) |
-| `ReplaceDataFiles` using `OpReplace` | Pending
([#841](https://github.com/apache/iceberg-go/issues/841)) |
+| `ReplaceDataFilesByPath` using `OpReplace` | Pending
([#841](https://github.com/apache/iceberg-go/issues/841)) |
### V3
diff --git a/website/src/releases.md b/website/src/releases.md
index 61ae67689..074e1453f 100644
--- a/website/src/releases.md
+++ b/website/src/releases.md
@@ -59,6 +59,8 @@ Per-release notes (highlights, breaking changes,
contributors) are published on
- Option constructors: `WithCleanupLocation`, `WithCleanupFilesOlderThan`,
`WithCleanupDryRun`, `WithCleanupDeleteFunc`, `WithCleanupMaxConcurrency`,
`WithPrefixMismatchMode`, `WithEqualSchemes`, `WithEqualAuthorities`. The four
generic names (`WithLocation`, `WithFilesOlderThan`, `WithDryRun`,
`WithDeleteFunc`) gained the `WithCleanup` prefix for consistency.
Relatedly, the previously unexported `isGCEnabled` is now
`table.IsGCEnabled`, so the new package can gate destructive operations on the
`gc.enabled` table property.
+- **Data-file replacement methods renamed.** Use `ReplaceDataFilesByPath` for
string paths. Use `ReplaceDataFiles` for prebuilt `DataFile` values instead of
`ReplaceDataFilesWithDataFiles`.
+
- `maintenance.PurgeFiles` (previously `Table.PurgeFiles`) now calls
`IO.Remove` concurrently, with up to 32 calls in flight, when the filesystem
does not implement `BulkRemovableIO`. Custom IO implementations, including
those registered with `io.Register`, must make `Remove` safe for concurrent use
and synchronize any shared mutable state. The bulk deletion path is unchanged.
- **Partition and sort source accessors renamed.** Replace
`PartitionField.SourceID()` and `SortField.SourceID()` with `FirstSourceID()`.
Their behavior and serialized fields are unchanged.
