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 6e83162d8 test(table): pin copy-on-write abort when a concurrent
commit removed the same file (#2140)
6e83162d8 is described below
commit 6e83162d8bc4d405a322fc465d78348a27a4c1bf
Author: Revanth <[email protected]>
AuthorDate: Sun Oct 11 08:52:26 2026 -0500
test(table): pin copy-on-write abort when a concurrent commit removed the
same file (#2140)
* test(table): pin copy-on-write abort when a concurrent commit removed the
same file
Signed-off-by: Revanth Ch <[email protected]>
* test(table): pin copy-on-write divergence to checkRemovedFiles and verify
surviving rows
* docs(table): note copy-on-write shares rewriteValidator with
RewriteDataFiles
---------
Signed-off-by: Revanth Ch <[email protected]>
---
table/cow_delete_conflict_test.go | 60 ++++++++++++++++++++++++++++++++-------
table/transaction.go | 11 +++----
2 files changed, 56 insertions(+), 15 deletions(-)
diff --git a/table/cow_delete_conflict_test.go
b/table/cow_delete_conflict_test.go
index ec4c227cd..693681be3 100644
--- a/table/cow_delete_conflict_test.go
+++ b/table/cow_delete_conflict_test.go
@@ -75,7 +75,10 @@ func cowTestRecords(t *testing.T, rowsJSON string)
array.RecordReader {
require.NoError(t, err)
t.Cleanup(data.Release)
- return array.NewTableReader(data, -1)
+ rdr := array.NewTableReader(data, -1)
+ t.Cleanup(rdr.Release)
+
+ return rdr
}
// stageCopyOnWrite stages, without committing, a copy-on-write removal of the
@@ -104,15 +107,8 @@ var (
cowFormatVersions = []string{"2", "3"}
cowIsolations = []table.IsolationLevel{table.IsolationSerializable,
table.IsolationSnapshot}
cowOps = []string{"delete", "overwrite"}
-)
-// TestCopyOnWriteConflict_ConcurrentDeleteOnRemovedFile proves a copy-on-write
-// Delete or Overwrite is rejected, at every isolation level, when a concurrent
-// commit added a delete against a data file it removes. Without the validator,
-// refresh-and-replay swaps the original file for a rewrite built from the
stale
-// snapshot, dropping the concurrent delete and resurrecting its row (#2090).
-func TestCopyOnWriteConflict_ConcurrentDeleteOnRemovedFile(t *testing.T) {
- removals := []struct {
+ cowRemovals = []struct {
name string
filter iceberg.BooleanExpression
}{
@@ -121,11 +117,18 @@ func
TestCopyOnWriteConflict_ConcurrentDeleteOnRemovedFile(t *testing.T) {
// id>=1 matches every row, so the file is dropped outright.
{"fully deleted",
iceberg.GreaterThanEqual(iceberg.Reference("id"), int64(1))},
}
+)
+// TestCopyOnWriteConflict_ConcurrentDeleteOnRemovedFile proves a copy-on-write
+// Delete or Overwrite is rejected, at every isolation level, when a concurrent
+// commit added a delete against a data file it removes. Without the validator,
+// refresh-and-replay swaps the original file for a rewrite built from the
stale
+// snapshot, dropping the concurrent delete and resurrecting its row (#2090).
+func TestCopyOnWriteConflict_ConcurrentDeleteOnRemovedFile(t *testing.T) {
for _, version := range cowFormatVersions {
for _, isolation := range cowIsolations {
for _, op := range cowOps {
- for _, removal := range removals {
+ for _, removal := range cowRemovals {
name := fmt.Sprintf("v%s/%s/%s/%s",
version, isolation, op, removal.name)
t.Run(name, func(t *testing.T) {
ctx := context.Background()
@@ -198,3 +201,40 @@ func
TestCopyOnWriteConflict_ConcurrentDeleteOnUntouchedFileCommits(t *testing.T
}
}
}
+
+// TestCopyOnWriteConflict_ConcurrentRemovalOfSameFileDiverges pins the other
+// half of copy-on-write conflict detection: when a concurrent commit already
+// removed a data file this commit also removes, the stale commit aborts
+// terminally with ErrCommitDiverged (the retry rebuild's checkRemovedFiles)
+// rather than rebuilding the file from the stale snapshot, and the table keeps
+// the concurrent result.
+func TestCopyOnWriteConflict_ConcurrentRemovalOfSameFileDiverges(t *testing.T)
{
+ for _, version := range cowFormatVersions {
+ for _, isolation := range cowIsolations {
+ for _, op := range cowOps {
+ for _, removal := range cowRemovals {
+ name := fmt.Sprintf("v%s/%s/%s/%s",
version, isolation, op, removal.name)
+ t.Run(name, func(t *testing.T) {
+ ctx := context.Background()
+ tbl := appendTenRows(t,
newCoWConflictTestTable(t, version, isolation))
+
+ txn := stageCopyOnWrite(t, tbl,
op, removal.filter)
+
+ // A concurrent copy-on-write
delete of id==4 rewrites the
+ // same data file first.
+ _, err := stageCopyOnWrite(t,
tbl, "delete", iceberg.EqualTo(iceberg.Reference("id"), int64(4))).Commit(ctx)
+ require.NoError(t, err)
+
+ _, err = txn.Commit(ctx)
+ require.ErrorIs(t, err,
table.ErrCommitDiverged)
+ require.ErrorContains(t, err,
"no longer on the branch head")
+ require.NotErrorIs(t, err,
table.ErrCommitFailed)
+
+ require.NoError(t,
tbl.Refresh(ctx))
+ require.Equal(t, []int64{1, 2,
3, 5, 6, 7, 8, 9, 10}, idsInTable(t, tbl))
+ })
+ }
+ }
+ }
+ }
+}
diff --git a/table/transaction.go b/table/transaction.go
index b6a1bb706..9516a3b92 100644
--- a/table/transaction.go
+++ b/table/transaction.go
@@ -2496,12 +2496,13 @@ func (t *Transaction) performCopyOnWriteDeletion(ctx
context.Context, operation
// without this check a refresh-and-replay would swap the original file
// for a rewrite built from the stale snapshot, dropping the concurrent
// deletes and resurrecting their rows. Mirrors Java's copy-on-write
- // validateNoConflictingDeletes: no isolation gating.
- removed := append(slices.Clip(filesToDelete), filesToRewrite...)
+ // validateNoConflictingDeletes: no isolation gating. The check is
+ // shared with RewriteDataFiles via rewriteValidator, so changes there
+ // apply here too. A concurrent removal of one of these files is caught
+ // earlier, by the retry rebuild's checkRemovedFiles
(ErrCommitDiverged).
+ removed := slices.Concat(filesToDelete, filesToRewrite)
if len(removed) > 0 {
- t.addValidator(func(cc *conflictContext) error {
- return validateNoNewDeletesForRewrittenFiles(cc,
removed)
- })
+ t.addValidator(rewriteValidator(removed))
}
return updater, wfs, nil
