badalprasadsingh commented on code in PR #2542:
URL: https://github.com/apache/iceberg-go/pull/2542#discussion_r4554455557
##########
table/transaction_delete.go:
##########
@@ -177,6 +181,28 @@ func (t *Transaction) performMergeOnReadDeletion(ctx
context.Context, snapshotPr
dvs.removeDataFile(updater, df)
}
+ // A whole-file drop takes the file's deletion vector with it. Reject
any
+ // delete a concurrent commit adds against those files, so the drop can
+ // replay on a CAS conflict without stranding a peer's DV. The validator
+ // treats any concurrent equality delete as a conflict, so it covers
only
+ // files that carry a DV: drops of other files keep replaying as before.
+ //
+ // Only a delete that drops whole files and nothing else may replay. A
+ // partial delete adds a DV to a file that stays live, and nothing
checks
+ // that file for a concurrent DV, so replaying it could leave two live
DVs
+ // on one file. Such deletes keep the no-replay behavior the DV removal
+ // gives them.
+ var droppedWithDV []iceberg.DataFile
+ for _, df := range filesToDelete {
+ if _, ok := dvs[df.FilePath()]; ok {
+ droppedWithDV = append(droppedWithDV, df)
+ }
+ }
+ if len(droppedWithDV) > 0 && len(withPartialDeletions) == 0 {
+ t.addValidator(rewriteValidator(droppedWithDV))
+ updater.markRemovedDVsValidated()
+ }
Review Comment:
P1: here, when a delete drops a file with a dv and one without: it now
replays but checks only the file with the dv, a peer's new dv on the other file
then stays live after that file is gone-> somehting which the spec forbids
better: check every dropped v3 file:
like this:
```go
if meta.formatVersion >= 3 && len(filesToDelete) > 0 {
t.addValidator(rejectConcurrentDVs(filesToDelete))
if len(withPartialDeletions) == 0 {
updater.markRemovedDVsValidated()
}
}
```
`rejectConcurrentDVs` would be a new DV-only check, like the one #2434 added
below for partial deletes, not rewriteValidator, which also rejects equality
deletes.
##########
table/table.go:
##########
@@ -610,8 +610,8 @@ type commitOpts struct {
// noReplay makes a CAS conflict terminal: doCommit returns the
// ErrCommitFailed error instead of entering refresh-and-replay.
- // Set for commits carrying delete-file removals; see the flag site in
- // snapshotProducer.commitManifests for the rationale.
+ // Set for commits carrying delete-file removals that a replay could
+ // get wrong; see the flag site in snapshotProducer.commitManifests.
Review Comment:
these still say a commit removing a dv is never replayed:
- OverwriteTableat line356
- Overwrite at lin389
- Delete at line422
- the two Transaction overwrites (transaction.go: line2481, line2552)
- Transaction.Delete (transaction_delete.go: line313)
better: drop that from the overwrites and limit it to mor deletes that also
delete part of a file
##########
table/no_replay_removals_test.go:
##########
@@ -461,20 +465,303 @@ func TestCoWDeleteRemovingDVFailsInsteadOfReplaying(t
*testing.T) {
attemptsBefore := cat.attempts.Load()
_, err = staleTxn.Commit(ctx)
- assert.Equal(t, attemptsBefore+tc.wantAttempts,
cat.attempts.Load())
+ require.NoError(t, err)
+ assert.Equal(t, attemptsBefore+2, cat.attempts.Load(),
+ "expected exactly one conflict followed by one
successful replay")
committed, loadErr := cat.LoadTable(ctx,
table.Identifier{"db", "no_replay_removals"})
require.NoError(t, loadErr)
- if tc.seededDVs > 0 {
- require.Error(t, err)
- assert.ErrorIs(t, err, table.ErrCommitFailed)
- assert.Equal(t, []int64{1, 2, 3, 4, 5, 6, 8, 9,
10, 100}, idsInTable(t, committed),
- "only the peer's append landed; the
stale delete of id 2 must not have")
- } else {
- require.NoError(t, err)
- assert.Equal(t, []int64{1, 3, 4, 5, 6, 7, 8, 9,
10, 100}, idsInTable(t, committed),
- "the delete replays over the peer's
append, so both writers land")
- }
+ assert.Equal(t, tc.wantIDs, idsInTable(t, committed),
+ "the delete replays over the peer's append, so
both writers land")
+ assert.Zero(t, liveDVCount(t, committed),
+ "the rewritten file must not keep the removed
deletion vector")
})
}
}
+
+// seedTwoFileTable returns a v3 merge-on-read table with ids 1-10 in one data
+// file and ids 100 and 101 in a second, so a filter on id >= 100 drops the
+// second file whole.
+func seedTwoFileTable(t *testing.T) (*table.Table, *concurrentTestCatalog) {
+ t.Helper()
+
+ tbl, cat := newNoReplayV3Table(t)
+ tbl = appendTenRows(t, tbl)
+ tbl = appendRowsOnRef(t, tbl, table.MainBranch, 100, 101)
+
+ return tbl, cat
+}
+
+// dataFilePathFor returns the path of the single data file holding id.
+func dataFilePathFor(t *testing.T, tbl *table.Table, id int64) string {
+ t.Helper()
+
+ tasks, err :=
tbl.Scan(table.WithRowFilter(iceberg.EqualTo(iceberg.Reference("id"),
id))).PlanFiles(t.Context())
+ require.NoError(t, err)
+ require.Len(t, tasks, 1)
+
+ return tasks[0].File.FilePath()
+}
+
+func switchToCopyOnWrite(t *testing.T, tbl *table.Table) *table.Table {
+ t.Helper()
+
+ tx := tbl.NewTransaction()
+ require.NoError(t,
tx.SetProperties(iceberg.Properties{table.WriteDeleteModeKey:
table.WriteModeCopyOnWrite}))
+ tbl, err := tx.Commit(t.Context())
+ require.NoError(t, err)
+
+ return tbl
+}
+
+func loadNoReplayTable(t *testing.T, cat *concurrentTestCatalog) *table.Table {
+ t.Helper()
+
+ committed, err := cat.LoadTable(t.Context(), table.Identifier{"db",
"no_replay_removals"})
+ require.NoError(t, err)
+
+ return committed
+}
+
+var idAtLeast100 = iceberg.GreaterThanEqual(iceberg.Reference("id"),
int64(100))
+
+// Why: #2195. A copy-on-write delete that drops a whole file removes the
file's
+// deletion vector with it, under rewriteValidator over the removed file, so a
+// concurrent unrelated append must not make it fail.
+// Condition: the second file carries a DV; a stale CoW delete drops that file
+// while a peer appends.
+// Assertion: the delete lands on the replay (two CommitTable attempts), the
+// peer's row survives, and no live DV is left.
+func TestCoWWholeFileDropWithDVReplaysOverConcurrentAppend(t *testing.T) {
+ ctx := t.Context()
+ tbl, cat := seedTwoFileTable(t)
+ tbl = mergeOnReadDelete(t, tbl, 101)
+ require.Equal(t, 1, liveDVCount(t, tbl))
+ tbl = switchToCopyOnWrite(t, tbl)
+
+ staleTxn := tbl.NewTransaction()
+ require.NoError(t, staleTxn.Delete(ctx, idAtLeast100, nil))
+ appendRowsOnRef(t, tbl, table.MainBranch, 50)
+
+ attemptsBefore := cat.attempts.Load()
+ _, err := staleTxn.Commit(ctx)
+ require.NoError(t, err)
+ assert.Equal(t, attemptsBefore+2, cat.attempts.Load(), "expected one
conflict and one successful replay")
+
+ committed := loadNoReplayTable(t, cat)
+ assert.Equal(t, []int64{1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 50},
idsInTable(t, committed))
+ assert.Zero(t, liveDVCount(t, committed))
+}
+
+// Why: #2195. Overwrite shares the copy-on-write deletion path, so the same
+// replay applies when its filter drops a file that carries a DV.
+// Condition: the second file carries a DV; a stale overwrite of id >= 100
+// races a peer append.
+// Assertion: the overwrite lands on the replay with both writers' rows.
+func TestCoWOverwriteRemovingDVReplaysOverConcurrentAppend(t *testing.T) {
+ ctx := t.Context()
+ tbl, cat := seedTwoFileTable(t)
+ tbl = mergeOnReadDelete(t, tbl, 101)
+ tbl = switchToCopyOnWrite(t, tbl)
+
+ staleTxn := tbl.NewTransaction()
+ rdr := idRecordReader(t, tbl, 150)
+ defer rdr.Release()
+ require.NoError(t, staleTxn.Overwrite(ctx, rdr, nil,
table.WithOverwriteFilter(idAtLeast100)))
+ appendRowsOnRef(t, tbl, table.MainBranch, 50)
+
+ attemptsBefore := cat.attempts.Load()
+ _, err := staleTxn.Commit(ctx)
+ require.NoError(t, err)
+ assert.Equal(t, attemptsBefore+2, cat.attempts.Load(), "expected one
conflict and one successful replay")
+
+ committed := loadNoReplayTable(t, cat)
+ assert.Equal(t, []int64{1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 50, 150},
idsInTable(t, committed))
+ assert.Zero(t, liveDVCount(t, committed))
+}
+
+// Why: #2195. A merge-on-read delete that drops a whole file also takes the
+// file's DV with it. It now registers rewriteValidator over that file, so it
+// can replay over an unrelated append.
+// Condition: the second file carries a DV; a stale MoR delete of id >= 100
+// races a peer append.
+// Assertion: the delete lands on the replay and no DV references the dropped
+// file.
+func TestMoRWholeFileDropWithDVReplaysOverConcurrentAppend(t *testing.T) {
+ ctx := t.Context()
+ tbl, cat := seedTwoFileTable(t)
+ tbl = mergeOnReadDelete(t, tbl, 101)
+ dropped := dataFilePathFor(t, tbl, 100)
+
+ staleTxn := tbl.NewTransaction()
+ require.NoError(t, staleTxn.Delete(ctx, idAtLeast100, nil))
+ appendRowsOnRef(t, tbl, table.MainBranch, 50)
+
+ attemptsBefore := cat.attempts.Load()
+ _, err := staleTxn.Commit(ctx)
+ require.NoError(t, err)
+ assert.Equal(t, attemptsBefore+2, cat.attempts.Load(), "expected one
conflict and one successful replay")
+
+ committed := loadNoReplayTable(t, cat)
+ assert.Equal(t, []int64{1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 50},
idsInTable(t, committed))
+ assert.Empty(t, deleteEntriesReferencing(t, committed,
map[string]struct{}{dropped: {}}),
+ "no DV may reference the dropped file")
+}
+
+// Why: #2195. Replaying a whole-file drop is only safe because the validator
+// rejects a peer's new DV on the dropped file. Without it, the stale DV
removal
+// would replay as a no-op and leave the peer's DV orphaned on a removed file.
+// Condition: the second file carries a DV; a peer supersedes it with its own
+// merged DV while a stale MoR delete drops the file.
+// Assertion: the stale commit is rejected without committing, and the table
+// is the peer's, with exactly one live DV on the second file.
+func TestMoRWholeFileDropRejectsConcurrentDVOnDroppedFile(t *testing.T) {
Review Comment:
add the mixed case next to this test: drop a file with a dv and one without
while a peer gives the second file its first dv
assert the commit fails with `ErrConflictingDeleteFiles` and no live dv
references a removed file
##########
website/src/concurrent-writes.md:
##########
@@ -143,7 +143,10 @@ the transaction is unusable.
After a conflict, calling `Commit` again refreshes the table and replays staged
changes even with retries disabled. Each call gets the configured retry
budget. A commit that removes delete files and cannot be replayed resubmits
-only if its branch has not moved.
+only if its branch has not moved. Copy-on-write deletes and overwrites, and
+merge-on-read deletes that only drop whole files, still replay when they remove
+a deletion vector together with its data file: they reject any delete a
+concurrent commit adds to the data files whose deletion vectors they remove.
Review Comment:
these commits used to fail at once with -> ErrCommitFailed
now they retry, and when a peer removed the same file they fail with ->
ErrCommitDiverged -> something which does not match ErrCommitFailed so a retry
loop keyed on ErrCommitFailed stops
better: add an unreleased compatibility note in `releases.md`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
