[RLC] Case 2 — same-file deletion-vector merge (concurrent DML ⟂ DML)

spark/src/main/scala/org/apache/spark/sql/delta/ConflictChecker.scala

@@ -270,11 +270,16 @@ object WinningCommitSummary {
 }
 
 private[delta] class ConflictChecker(
-    spark: SparkSession,
+    protected val spark: SparkSession,
     initialCurrentTransactionInfo: CurrentTransactionInfo,
-    winningCommitSummary: WinningCommitSummary,
+    // A `var` (reassigned once by [[prepareRowLevelConflictResolution]]) to drop files that are
+    // resolved at the row level, mirroring how `currentTransactionInfo` is rebased. After that the
+    // file-level checks run against a summary that never mentions those files, so they need no
+    // changes.
+    protected var winningCommitSummary: WinningCommitSummary,
     isolationLevel: IsolationLevel)
-  extends DeltaLogging with ConflictCheckerPredicateElimination {
+  extends DeltaLogging with ConflictCheckerPredicateElimination
+  with RowLevelConcurrencyResolution {
 
   protected val dataPath = initialCurrentTransactionInfo.readSnapshot.dataPath
   protected val winningCommitVersion = winningCommitSummary.commitVersion
@@ -352,12 +357,24 @@ private[delta] class ConflictChecker(
     // Update the table version in newly added type widening metadata.
     updateTypeWideningMetadata()
 
+    // Row-level concurrency: set aside the "same physical file" conflicts that merging deletion
+    // vectors may resolve, so that concurrent DML touching disjoint rows of the same file no
+    // longer aborts. The file-level checks below then see only the other conflicts. Runs after
+    // row-ID reassignment so merged files keep stable base row IDs.
+    prepareRowLevelConflictResolution()
+
     // Data file checks.
     checkForAddedFilesThatShouldHaveBeenReadByCurrentTxn()
     checkForDeletedFilesAgainstCurrentTxnReadFiles()
     checkForDeletedFilesAgainstCurrentTxnDeletedFiles()
     resolveTimestampOrderingConflicts()
 
+    // Resolve the set-aside files last, so a transaction that aborts on another conflict decodes
+    // no deletion vector, scans no read and writes no merged deletion vector.
+    resolveRowLevelConflicts()
+
+    recordRowLevelConflictResolution()
+
     // Fold the winning commit into our transaction info before the next attempt: this advances
     // `preCommitLatestAMTCheckpointOpt` to the winner's inline AMT checkpoint (when the winner
     // wrote one) and clears `currentCommitAttemptAMTCheckpointOpt`.