From e3562830d1f66d4835bc1f0d0e039f9ad87eaaeb Mon Sep 17 00:00:00 2001 From: Eunjin Song Date: Tue, 7 Jul 2026 15:46:11 -0700 Subject: [PATCH 01/11] [Spark] Row-level concurrency for concurrent DML via deletion-vector merge POC for issue #7057. Adds a config-gated conflict-resolution phase to ConflictChecker that resolves 'same physical file' conflicts between concurrent DML. It decodes both transactions' deletion vectors and, when the newly-deleted rows are disjoint, merges them (dv_win UNION dv_cur), writes a new DV file, and rebases the losing transaction onto the winner's post-image. The delete/read, delete/delete, and append checks skip resolved paths; rewrite-only DML (DELETE/UPDATE) winners also skip the append check. DV-only (row tracking not required). Gated by spark.databricks.delta.rowLevelConcurrency.enabled (default off). Adds RowLevelConcurrencySuite. Co-Authored-By: Claude Opus 4.8 --- .../spark/sql/delta/ConflictChecker.scala | 234 +++++++++++++- .../sql/delta/sources/DeltaSQLConf.scala | 13 + .../sql/delta/RowLevelConcurrencySuite.scala | 297 ++++++++++++++++++ 3 files changed, 538 insertions(+), 6 deletions(-) create mode 100644 spark/src/test/scala/org/apache/spark/sql/delta/RowLevelConcurrencySuite.scala diff --git a/spark/src/main/scala/org/apache/spark/sql/delta/ConflictChecker.scala b/spark/src/main/scala/org/apache/spark/sql/delta/ConflictChecker.scala index d5cd7be1beb..b77bca2f9f1 100644 --- a/spark/src/main/scala/org/apache/spark/sql/delta/ConflictChecker.scala +++ b/spark/src/main/scala/org/apache/spark/sql/delta/ConflictChecker.scala @@ -17,24 +17,28 @@ package org.apache.spark.sql.delta // scalastyle:off import.ordering.noEmptyLine +import java.util.UUID import java.util.concurrent.TimeUnit import scala.collection.mutable -import org.apache.spark.sql.delta.DeltaOperations.{OP_SET_TBLPROPERTIES, ROW_TRACKING_BACKFILL_OPERATION_NAME, ROW_TRACKING_UNBACKFILL_OPERATION_NAME} +import org.apache.spark.sql.delta.DeltaOperations.{OP_DELETE, OP_SET_TBLPROPERTIES, OP_UPDATE, ROW_TRACKING_BACKFILL_OPERATION_NAME, ROW_TRACKING_UNBACKFILL_OPERATION_NAME} import org.apache.spark.sql.delta.RowId.RowTrackingMetadataDomain import org.apache.spark.sql.delta.actions._ import org.apache.spark.sql.delta.catalog.DeltaTableV2 +import org.apache.spark.sql.delta.commands.DeletionVectorUtils +import org.apache.spark.sql.delta.deletionvectors.{RoaringBitmapArray, RoaringBitmapArrayFormat} import org.apache.spark.sql.delta.logging.DeltaLogKeys import org.apache.spark.sql.delta.metering.DeltaLogging import org.apache.spark.sql.delta.sources.DeltaSourceUtils import org.apache.spark.sql.delta.sources.DeltaSQLConf +import org.apache.spark.sql.delta.storage.dv.DeletionVectorStore import org.apache.spark.sql.delta.util.DeltaSparkPlanUtils.CheckDeterministicOptions import org.apache.spark.sql.delta.util.FileNames import io.delta.storage.commit.UpdatedActions import io.delta.storage.commit.uccommitcoordinator.UCCommitCoordinatorClient import io.delta.storage.commit.uniform.UniformMetadata -import org.apache.hadoop.fs.FileStatus +import org.apache.hadoop.fs.{FileStatus, Path} import org.apache.spark.internal.{MDC, MessageWithContext} import org.apache.spark.sql.{DataFrame, SparkSession} @@ -229,6 +233,13 @@ private[delta] class ConflictChecker( protected var currentTransactionInfo: CurrentTransactionInfo = initialCurrentTransactionInfo + /** + * Paths of files whose "same physical file" conflict with the winning transaction was resolved at + * the row level by [[resolveRowLevelConflicts]] (deletion vectors merged). The file-level delete + * and append checks skip these paths, since they have already been reconciled. + */ + private val rowLevelResolvedPaths = mutable.Set.empty[String] + protected def recordSkippedPhase(phase: String): Unit = timingStats += phase -> 0 /** @@ -300,6 +311,12 @@ private[delta] class ConflictChecker( // Update the table version in newly added type widening metadata. updateTypeWideningMetadata() + // Row-level concurrency: try to resolve "same physical file" conflicts by merging deletion + // vectors before the file-level checks run, so that concurrent DML touching disjoint rows of + // the same file no longer aborts. Runs after row-ID reassignment so merged files keep stable + // base row IDs. + resolveRowLevelConflicts() + // Data file checks. checkForAddedFilesThatShouldHaveBeenReadByCurrentTxn() checkForDeletedFilesAgainstCurrentTxnReadFiles() @@ -1117,6 +1134,204 @@ private[delta] class ConflictChecker( false } + /** Whether row-level concurrency resolution is enabled and applicable to this table. */ + private lazy val rowLevelConcurrencyEnabled: Boolean = + spark.conf.get(DeltaSQLConf.DELTA_ROW_LEVEL_CONCURRENCY_ENABLED) && + DeletionVectorUtils.deletionVectorsWritable( + currentTransactionInfo.protocol, currentTransactionInfo.metadata) + + /** The operation name of the winning commit, if available. */ + private lazy val winningOperationName: Option[String] = + winningCommitSummary.commitInfo.map(_.operation) + + /** + * Whether the winning commit is a row-level DML that only rewrites or removes existing rows + * (DELETE or UPDATE), i.e. it introduces no net-new logical rows. Files added by such a commit + * are rewrites of rows that the current transaction already observed in its read snapshot (or + * re-adds of files masked by a deletion vector), so they cannot introduce phantom rows for the + * current transaction. MERGE is intentionally excluded because it may insert net-new rows. + */ + private lazy val winningCommitIsRewriteOnlyRowLevelDml: Boolean = { + val usesDeletionVectors = winningCommitSummary.addedFiles.exists(_.deletionVector != null) + usesDeletionVectors && winningOperationName.exists { + case OP_DELETE | OP_UPDATE => true + case _ => false + } + } + + /** + * Whether a file added by the winning transaction can be skipped in the added-files (append) + * conflict check thanks to row-level concurrency resolution. This is true when: + * 1. the file's "same physical file" conflict was already reconciled by merging deletion + * vectors ([[rowLevelResolvedPaths]]), or + * 2. the winning commit is a rewrite-only row-level DML (DELETE/UPDATE) whose added files + * cannot introduce phantom rows (see [[winningCommitIsRewriteOnlyRowLevelDml]]). + */ + private def canSkipAddedFileForRowLevelConcurrency(addFile: AddFile): Boolean = { + if (!rowLevelConcurrencyEnabled) return false + rowLevelResolvedPaths.contains(addFile.path) || winningCommitIsRewriteOnlyRowLevelDml + } + + /** + * Resolves "same physical file" conflicts with the winning transaction at the row level. + * + * A DV-based DELETE/UPDATE emits, for each touched file `P`, a `RemoveFile(P)` (tombstone of the + * pre-image) and an `AddFile(P)` carrying a larger deletion vector. When both the winning and the + * current transaction touch the same file `P` this way, the two operations are logically + * independent as long as they mark *different* rows deleted. For every such shared file we: + * 1. decode the winning DV, the current DV and their common base DV (from the pre-image + * `RemoveFile`) as [[RoaringBitmapArray]]s; + * 2. check whether the newly-deleted rows are disjoint, i.e. `(dv_win INTERSECT dv_cur) MINUS + * dv_base` is empty. If they overlap, this is a genuine row-level conflict and we leave the + * file for the standard checks to abort; + * 3. on disjoint sets, merge the DVs (`dv_win UNION dv_cur`), persist a new DV file, + * and rebase the current transaction onto the winner's post-image: the current `AddFile(P)` + * now carries the merged DV and the current `RemoveFile(P)` now tombstones the winner's + * `AddFile(P)`. + * + * Resolved paths are recorded in [[rowLevelResolvedPaths]] and skipped by the file-level delete + * and append checks. Row identity is preserved for free: the merged file is the same physical + * file, so its base row ID is unchanged and [[reassignOverlappingRowIds]] (already run) leaves it + * alone. Deletion vectors index physical row positions within one immutable Parquet file, so the + * merge needs no row tracking. + */ + private def resolveRowLevelConflicts(): Unit = { + if (!rowLevelConcurrencyEnabled) return + + // Winning transaction's DV updates: path present in both an AddFile (with a DV) and a + // RemoveFile of the winning commit. + val winningRemovedPaths = winningCommitSummary.removedFiles.map(_.path).toSet + val winningDvUpdates: Map[String, AddFile] = winningCommitSummary.addedFiles.iterator + .filter(a => a.deletionVector != null && winningRemovedPaths.contains(a.path)) + .map(a => a.path -> a) + .toMap + if (winningDvUpdates.isEmpty) return + + // Current transaction's DV updates, indexed by path. + val currentAddByPath = currentTransactionInfo.actions.collect { + case a: AddFile if a.deletionVector != null => a.path -> a + }.toMap + val currentRemoveByPath = currentTransactionInfo.actions.collect { + case r: RemoveFile => r.path -> r + }.toMap + + val sharedPaths = winningDvUpdates.keySet + .intersect(currentAddByPath.keySet) + .intersect(currentRemoveByPath.keySet) + if (sharedPaths.isEmpty) return + + recordTime("resolved-row-level-conflicts") { + val dvStore = DeletionVectorStore.createInstance(deltaLog.newDeltaHadoopConf()) + val tablePath = deltaLog.dataPath + + // path -> (rebased AddFile, rebased RemoveFile) + val replacements = mutable.Map.empty[String, (AddFile, RemoveFile)] + for (path <- sharedPaths) { + val winningAdd = winningDvUpdates(path) + val currentAdd = currentAddByPath(path) + val currentRemove = currentRemoveByPath(path) + + def bitmapOf(dv: DeletionVectorDescriptor): RoaringBitmapArray = + readDeletionVectorOrEmpty(dvStore, dv, tablePath) + val baseBitmap = bitmapOf(currentRemove.deletionVector) + val winningBitmap = bitmapOf(winningAdd.deletionVector) + val currentBitmap = bitmapOf(currentAdd.deletionVector) + + // `baseBitmap` is the DV of `P` at the current txn's read time (carried on its RemoveFile); + // for a 2-way conflict it equals the winner's pre-image too. Both `dv_win` and `dv_cur` are + // supersets of it (a DV only grows), so the newly-deleted rows are `dv \ base` on each side + // and their overlap is `(dv_win INTERSECT dv_cur) MINUS base`. If empty, the two txns + // touched disjoint rows and the schedule `current ; winner` is a valid serialization under + // both WriteSerializable and Serializable (the winner's rewrites/deletes are of rows the + // current txn did not touch), so merging is safe. If non-empty, the same row was touched by + // both -> genuine conflict, left for the standard checks. (For 3+ way chains `base` becomes + // previous winner's DV rather than the original pre-image; the merge stays correct and the + // overlap test stays conservative.) + val newlyDeletedOverlap = winningBitmap.copy() + newlyDeletedOverlap.and(currentBitmap) + newlyDeletedOverlap.andNot(baseBitmap) + + if (newlyDeletedOverlap.isEmpty) { + // Disjoint: merge the deletion vectors and rebase onto the winner's post-image. + val mergedBitmap = winningBitmap.copy() + mergedBitmap.merge(currentBitmap) + val mergedDescriptor = writeMergedDeletionVector(dvStore, tablePath, mergedBitmap) + // Keep the current AddFile's identity (base row ID / default row commit version already + // reconciled by the row-ID phases) but point it at the merged DV. + val rebasedAdd = currentAdd + .copy(deletionVector = mergedDescriptor, dataChange = true) + .withoutTightBoundStats + // Tombstone the winner's now-live AddFile (carries the winning DV) instead of the stale + // pre-image. + val rebasedRemove = winningAdd.removeWithTimestamp() + replacements(path) = (rebasedAdd, rebasedRemove) + rowLevelResolvedPaths += path + } + // else: overlapping row-level modification -> genuine conflict, leave for standard checks. + } + + if (replacements.nonEmpty) { + val newActions = currentTransactionInfo.actions.map { + case a: AddFile if replacements.contains(a.path) => replacements(a.path)._1 + case r: RemoveFile if replacements.contains(r.path) => replacements(r.path)._2 + case other => other + } + // Resolved files are no longer "read" for the purposes of the delete-read check. + val newReadFiles = currentTransactionInfo.readFiles + .filterNot(f => rowLevelResolvedPaths.contains(f.path)) + currentTransactionInfo = + currentTransactionInfo.copy(actions = newActions, readFiles = newReadFiles) + + recordDeltaEvent( + deltaLog, + opType = "delta.rowLevelConcurrency.deletionVectorsMerged", + data = Map( + "winningCommitVersion" -> winningCommitVersion, + "resolvedPaths" -> rowLevelResolvedPaths.size, + "winningOperation" -> winningOperationName.getOrElse("UNKNOWN"))) + } + } + } + + /** Reads a deletion vector into a [[RoaringBitmapArray]], returning an empty bitmap for none. */ + private def readDeletionVectorOrEmpty( + dvStore: DeletionVectorStore, + dv: DeletionVectorDescriptor, + tablePath: Path): RoaringBitmapArray = { + if (dv == null || dv.isEmpty) new RoaringBitmapArray() else dvStore.read(dv, tablePath) + } + + /** + * Persists a merged bitmap to a new deletion vector file and returns its descriptor. + * + * NOTE: this writes a DV file as a side effect of conflict resolution. If the commit ultimately + * fails or is retried against another winning version, the file is orphaned and later reclaimed + * by VACUUM (same lifecycle as any DV written by DML). This mirrors how the DML write path + * persists DVs (see `DeletionVectorWriter.storeSerializedBitmap`). + */ + private def writeMergedDeletionVector( + dvStore: DeletionVectorStore, + tablePath: Path, + bitmap: RoaringBitmapArray): DeletionVectorDescriptor = { + // An empty DV has no on-disk representation (matches DeletionVectorWriter). + if (bitmap.isEmpty) return DeletionVectorDescriptor.EMPTY + val tablePathWithFs = dvStore.pathWithFileSystem(tablePath) + val fileId = UUID.randomUUID() + val writer = dvStore.createWriter(dvStore.generateFileNameInTable(tablePathWithFs, fileId)) + try { + val serialized = DeletionVectorUtils.serialize( + bitmap, RoaringBitmapArrayFormat.Portable, Some(tablePath)) + val range = writer.write(serialized) + DeletionVectorDescriptor.onDiskWithRelativePath( + id = fileId, + sizeInBytes = serialized.length, + cardinality = bitmap.cardinality, + offset = Some(range.offset)) + } finally { + writer.close() + } + } + /** * Check if the new files added by the already committed transactions should have been read by * the current transaction. @@ -1137,8 +1352,11 @@ private[delta] class ConflictChecker( Seq.empty } + val addedFilesAfterRowLevelResolution = + addedFilesToCheckForConflicts.filterNot(canSkipAddedFileForRowLevelConcurrency) + val fileMatchingPartitionReadPredicates = - getFirstFileMatchingPartitionPredicates(addedFilesToCheckForConflicts) + getFirstFileMatchingPartitionPredicates(addedFilesAfterRowLevelResolution) if (fileMatchingPartitionReadPredicates.nonEmpty) { throw DeltaErrors.concurrentAppendException( @@ -1160,7 +1378,7 @@ private[delta] class ConflictChecker( val readFilePaths = currentTransactionInfo.readFiles.map( f => f.path -> f.partitionValues).toMap val deleteReadOverlap = winningCommitSummary.removedFiles - .find(r => readFilePaths.contains(r.path)) + .find(r => readFilePaths.contains(r.path) && !rowLevelResolvedPaths.contains(r.path)) if (deleteReadOverlap.nonEmpty) { val partitionOpt = getPrettyPartitionMessage(readFilePaths(deleteReadOverlap.get.path)) throw DeltaErrors.concurrentDeleteReadException( @@ -1169,7 +1387,11 @@ private[delta] class ConflictChecker( winningCommitVersion, partitionOpt) } - if (winningCommitSummary.removedFiles.nonEmpty && currentTransactionInfo.readWholeTable) { + // Row-level concurrency: a removed file that was reconciled at the row level must not + // re-trigger the whole-table conflict either. + val unresolvedRemovedFiles = + winningCommitSummary.removedFiles.exists(r => !rowLevelResolvedPaths.contains(r.path)) + if (unresolvedRemovedFiles && currentTransactionInfo.readWholeTable) { throw DeltaErrors.concurrentDeleteReadException( winningCommitSummary.commitInfo, getTableNameOrPath, @@ -1190,7 +1412,7 @@ private[delta] class ConflictChecker( .collect { case r: RemoveFile => r.path -> r.partitionValues } .toMap val deleteOverlap = winningCommitSummary.removedFiles - .find(r => deletedFilePaths.contains(r.path)) + .find(r => deletedFilePaths.contains(r.path) && !rowLevelResolvedPaths.contains(r.path)) if (deleteOverlap.nonEmpty) { val partitionOpt = getPrettyPartitionMessage(deletedFilePaths(deleteOverlap.get.path)) throw DeltaErrors.concurrentDeleteDeleteException( diff --git a/spark/src/main/scala/org/apache/spark/sql/delta/sources/DeltaSQLConf.scala b/spark/src/main/scala/org/apache/spark/sql/delta/sources/DeltaSQLConf.scala index d4d9c813b6c..1277a464b86 100644 --- a/spark/src/main/scala/org/apache/spark/sql/delta/sources/DeltaSQLConf.scala +++ b/spark/src/main/scala/org/apache/spark/sql/delta/sources/DeltaSQLConf.scala @@ -534,6 +534,19 @@ trait DeltaSQLConfBase extends DeltaSQLConfUtils { .booleanConf .createWithDefault(true) + val DELTA_ROW_LEVEL_CONCURRENCY_ENABLED = + buildConf("rowLevelConcurrency.enabled") + .internal() + .doc( + """When enabled, the conflict checker attempts to resolve conflicts between concurrent + |DML operations at the row level using deletion vectors, instead of failing the + |transaction. For every "same physical file" conflict, it decodes both transactions' + |deletion vectors, checks whether the newly-deleted rows are disjoint, and on disjoint + |sets merges the deletion vectors and rebases the losing transaction onto the winner's + |post-image. Only active when deletion vectors are writable.""".stripMargin) + .booleanConf + .createWithDefault(false) + val DELTA_PROTOCOL_DEFAULT_WRITER_VERSION = buildConf("properties.defaults.minWriterVersion") .doc("The default writer protocol version to create new tables with, unless a feature " + diff --git a/spark/src/test/scala/org/apache/spark/sql/delta/RowLevelConcurrencySuite.scala b/spark/src/test/scala/org/apache/spark/sql/delta/RowLevelConcurrencySuite.scala new file mode 100644 index 00000000000..f5377fc38f3 --- /dev/null +++ b/spark/src/test/scala/org/apache/spark/sql/delta/RowLevelConcurrencySuite.scala @@ -0,0 +1,297 @@ +/* + * Copyright (2021) The Delta Lake Project Authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.spark.sql.delta + +import java.io.File + +import scala.concurrent.duration.Duration + +import org.apache.spark.sql.delta.concurrency.PhaseLockingTestMixin +import org.apache.spark.sql.delta.concurrency.TransactionExecutionTestMixin +import org.apache.spark.sql.delta.sources.DeltaSQLConf +import org.apache.spark.sql.delta.test.DeltaSQLCommandTest + +import org.apache.spark.{SparkConf, SparkException} +import org.apache.spark.sql.{QueryTest, Row} +import org.apache.spark.sql.functions.lit +import org.apache.spark.sql.test.SharedSparkSession +import org.apache.spark.util.ThreadUtils + +/** + * End-to-end tests for deletion-vector-based row-level concurrency + * ([[DeltaSQLConf.DELTA_ROW_LEVEL_CONCURRENCY_ENABLED]]). + * + * Two concurrent DML operations that touch the same physical file but modify disjoint rows should + * commit cleanly by merging their deletion vectors, instead of aborting the losing transaction. + */ +class RowLevelConcurrencySuite extends QueryTest + with SharedSparkSession + with DeltaSQLCommandTest + with PhaseLockingTestMixin + with TransactionExecutionTestMixin { + + // Enable deletion vectors on every table created by this suite. + override protected def sparkConf: SparkConf = super.sparkConf + .set(DeltaConfigs.ENABLE_DELETION_VECTORS_CREATION.defaultTablePropertyKey, "true") + + private def tableRef(dir: File): String = s"delta.`${dir.getCanonicalPath}`" + + /** Creates a single-file Delta table with `id` in [0, numRows) and deletion vectors enabled. */ + private def createSingleFileTableWithDVs(dir: File, numRows: Int = 100): DeltaLog = { + spark.range(start = 0, end = numRows, step = 1, numPartitions = 1) + .write.format("delta").mode("append").save(dir.getAbsolutePath) + val log = DeltaLog.forTable(spark, dir.getCanonicalPath) + val snapshot = log.update() + assert( + snapshot.metadata.configuration + .get(DeltaConfigs.ENABLE_DELETION_VECTORS_CREATION.key).contains("true"), + "deletion vectors must be enabled on the test table") + assert(snapshot.allFiles.collect().length === 1, "test table must have a single data file") + log + } + + /** A DELETE/UPDATE transaction that runs under the given row-level-concurrency setting. */ + private def sqlTxn(sqlText: String, rowLevelConcurrency: Boolean): () => Array[Row] = + () => { + withSQLConf( + DeltaSQLConf.DELTA_ROW_LEVEL_CONCURRENCY_ENABLED.key -> rowLevelConcurrency.toString) { + sql(sqlText).collect() + } + Array.empty[Row] + } + + private def deletionVectorCardinalities(log: DeltaLog): Seq[Long] = + log.update().allFiles.collect() + .filter(_.deletionVector != null) + .map(_.deletionVector.cardinality) + .toSeq + + private def assertConcurrentModificationException(e: SparkException): Unit = { + val causeName = e.getCause.getClass.getName + assert( + Seq("ConcurrentAppend", "ConcurrentDeleteRead", "ConcurrentDeleteDelete") + .exists(causeName.contains), + s"Expected a concurrency conflict, got: $causeName") + } + + private def ids(dir: File): Seq[Long] = + spark.read.format("delta").load(dir.getAbsolutePath).select("id") + .collect().map(_.getLong(0)).sorted.toSeq + + // --------------------------------------------------------------------------- + // DELETE vs DELETE (same file) + // --------------------------------------------------------------------------- + + test("disjoint concurrent DELETEs on the same file both commit by merging deletion vectors") { + withTempDir { dir => + val log = createSingleFileTableWithDVs(dir) + val txnA = sqlTxn(s"DELETE FROM ${tableRef(dir)} WHERE id = 10", rowLevelConcurrency = true) + val txnB = sqlTxn(s"DELETE FROM ${tableRef(dir)} WHERE id = 20", rowLevelConcurrency = true) + + val (futureA, futureB) = runTxnsWithOrder__A_Start__B__A_End(txnA, txnB) + ThreadUtils.awaitResult(futureA, Duration.Inf) + ThreadUtils.awaitResult(futureB, Duration.Inf) + + assert(ids(dir) === (0L to 99L).filterNot(id => id == 10 || id == 20)) + // One surviving file carrying the merged deletion vector (cardinality 2). + assert(deletionVectorCardinalities(log) === Seq(2L)) + } + } + + test("overlapping concurrent DELETEs still conflict") { + withTempDir { dir => + val log = createSingleFileTableWithDVs(dir) + val txnA = sqlTxn(s"DELETE FROM ${tableRef(dir)} WHERE id = 10", rowLevelConcurrency = true) + val txnB = sqlTxn(s"DELETE FROM ${tableRef(dir)} WHERE id = 10", rowLevelConcurrency = true) + + val (futureA, futureB) = runTxnsWithOrder__A_Start__B__A_End(txnA, txnB) + ThreadUtils.awaitResult(futureB, Duration.Inf) + val e = intercept[SparkException] { ThreadUtils.awaitResult(futureA, Duration.Inf) } + assertConcurrentModificationException(e) + // Clean abort: only the winner's delete (id=10) is applied; its DV has cardinality 1. + assert(ids(dir) === (0L to 99L).filterNot(_ == 10)) + assert(deletionVectorCardinalities(log) === Seq(1L)) + } + } + + test("feature disabled: disjoint concurrent DELETEs still conflict") { + withTempDir { dir => + val log = createSingleFileTableWithDVs(dir) + val txnA = sqlTxn(s"DELETE FROM ${tableRef(dir)} WHERE id = 10", rowLevelConcurrency = false) + val txnB = sqlTxn(s"DELETE FROM ${tableRef(dir)} WHERE id = 20", rowLevelConcurrency = false) + + val (futureA, futureB) = runTxnsWithOrder__A_Start__B__A_End(txnA, txnB) + ThreadUtils.awaitResult(futureB, Duration.Inf) + val e = intercept[SparkException] { ThreadUtils.awaitResult(futureA, Duration.Inf) } + assertConcurrentModificationException(e) + // Only the winner's delete (id=20) is applied; DVs are still used, so its DV has cardinality 1. + assert(ids(dir) === (0L to 99L).filterNot(_ == 20)) + assert(deletionVectorCardinalities(log) === Seq(1L)) + } + } + + test("deletion vectors disabled: row-level concurrency gate is off, disjoint DELETEs conflict") { + withTempDir { dir => + createSingleFileTableWithDVs(dir) + sql(s"ALTER TABLE ${tableRef(dir)} SET TBLPROPERTIES " + + s"('${DeltaConfigs.ENABLE_DELETION_VECTORS_CREATION.key}' = 'false')") + val log = DeltaLog.forTable(spark, dir.getCanonicalPath) + // Flag ON, but DVs are not writable -> resolveRowLevelConflicts no-ops. + val txnA = sqlTxn(s"DELETE FROM ${tableRef(dir)} WHERE id = 10", rowLevelConcurrency = true) + val txnB = sqlTxn(s"DELETE FROM ${tableRef(dir)} WHERE id = 20", rowLevelConcurrency = true) + + val (futureA, futureB) = runTxnsWithOrder__A_Start__B__A_End(txnA, txnB) + ThreadUtils.awaitResult(futureB, Duration.Inf) + val e = intercept[SparkException] { ThreadUtils.awaitResult(futureA, Duration.Inf) } + assertConcurrentModificationException(e) + // Winner's delete (id=20) applied by rewriting the file (no DVs), so no DV is present. + assert(ids(dir) === (0L to 99L).filterNot(_ == 20)) + assert(deletionVectorCardinalities(log) === Seq.empty[Long]) + } + } + + // --------------------------------------------------------------------------- + // Layer 2: UPDATE (rewrite-only DML adds new data files) + // --------------------------------------------------------------------------- + + test("disjoint DELETE (loser) vs UPDATE (winner) both commit") { + withTempDir { dir => + val log = createSingleFileTableWithDVs(dir) + // A (loser) deletes id=10; B (winner) updates id=20 -> 1020 (masks row 20, appends image). + val txnA = sqlTxn(s"DELETE FROM ${tableRef(dir)} WHERE id = 10", rowLevelConcurrency = true) + val txnB = sqlTxn(s"UPDATE ${tableRef(dir)} SET id = 1020 WHERE id = 20", + rowLevelConcurrency = true) + + val (futureA, futureB) = runTxnsWithOrder__A_Start__B__A_End(txnA, txnB) + ThreadUtils.awaitResult(futureA, Duration.Inf) + ThreadUtils.awaitResult(futureB, Duration.Inf) + + assert(ids(dir) === ((0L to 99L).filterNot(id => id == 10 || id == 20) :+ 1020L).sorted) + // The original file's merged DV masks rows 10 and 20; the updated image lives in a new file. + assert(deletionVectorCardinalities(log) === Seq(2L)) + } + } + + test("disjoint UPDATE vs UPDATE both commit") { + withTempDir { dir => + val log = createSingleFileTableWithDVs(dir) + val txnA = sqlTxn(s"UPDATE ${tableRef(dir)} SET id = 1010 WHERE id = 10", + rowLevelConcurrency = true) + val txnB = sqlTxn(s"UPDATE ${tableRef(dir)} SET id = 1020 WHERE id = 20", + rowLevelConcurrency = true) + + val (futureA, futureB) = runTxnsWithOrder__A_Start__B__A_End(txnA, txnB) + ThreadUtils.awaitResult(futureA, Duration.Inf) + ThreadUtils.awaitResult(futureB, Duration.Inf) + + assert(ids(dir) === + ((0L to 99L).filterNot(id => id == 10 || id == 20) ++ Seq(1010L, 1020L)).sorted) + // Original file's merged DV masks rows 10 and 20; updated images live in two new files. + assert(deletionVectorCardinalities(log) === Seq(2L)) + } + } + + test("overlapping UPDATE vs UPDATE (same row) still conflict") { + withTempDir { dir => + val log = createSingleFileTableWithDVs(dir) + val txnA = sqlTxn(s"UPDATE ${tableRef(dir)} SET id = 1010 WHERE id = 20", + rowLevelConcurrency = true) + val txnB = sqlTxn(s"UPDATE ${tableRef(dir)} SET id = 2020 WHERE id = 20", + rowLevelConcurrency = true) + + val (futureA, futureB) = runTxnsWithOrder__A_Start__B__A_End(txnA, txnB) + ThreadUtils.awaitResult(futureB, Duration.Inf) + val e = intercept[SparkException] { ThreadUtils.awaitResult(futureA, Duration.Inf) } + assertConcurrentModificationException(e) + // Only the winner's update (20 -> 2020) is applied; original file's DV masks row 20. + assert(ids(dir) === ((0L to 99L).filterNot(_ == 20) :+ 2020L).sorted) + assert(deletionVectorCardinalities(log) === Seq(1L)) + } + } + + // --------------------------------------------------------------------------- + // Winner fully removes the file -> not reconcilable -> conflict + // --------------------------------------------------------------------------- + + test("winner that fully removes a file conflicts with a concurrent row-level delete") { + withTempDir { dir => + // Two files: [0,50) and [50,100). + spark.range(start = 0, end = 100, step = 1, numPartitions = 2) + .write.format("delta").mode("append").save(dir.getAbsolutePath) + val log = DeltaLog.forTable(spark, dir.getCanonicalPath) + // A (loser) DV-deletes one row in the first file; B (winner) deletes the whole first file. + val txnA = sqlTxn(s"DELETE FROM ${tableRef(dir)} WHERE id = 10", rowLevelConcurrency = true) + val txnB = sqlTxn(s"DELETE FROM ${tableRef(dir)} WHERE id < 50", rowLevelConcurrency = true) + + val (futureA, futureB) = runTxnsWithOrder__A_Start__B__A_End(txnA, txnB) + ThreadUtils.awaitResult(futureB, Duration.Inf) + val e = intercept[SparkException] { ThreadUtils.awaitResult(futureA, Duration.Inf) } + assertConcurrentModificationException(e) + // Winner fully removed the first file (no DV); loser aborted cleanly. + assert(ids(dir) === (50L to 99L)) + assert(deletionVectorCardinalities(log) === Seq.empty[Long]) + } + } + + // --------------------------------------------------------------------------- + // N-way: three concurrent transactions on the same file + // --------------------------------------------------------------------------- + + test("three concurrent disjoint DELETEs on the same file all commit") { + withTempDir { dir => + val log = createSingleFileTableWithDVs(dir) + val txnA = sqlTxn(s"DELETE FROM ${tableRef(dir)} WHERE id = 10", rowLevelConcurrency = true) + val txnB = sqlTxn(s"DELETE FROM ${tableRef(dir)} WHERE id = 20", rowLevelConcurrency = true) + val txnC = sqlTxn(s"DELETE FROM ${tableRef(dir)} WHERE id = 30", rowLevelConcurrency = true) + + // A starts; B commits; C commits; A commits last (reconciles against both B and C). + val (futureA, futureB, futureC) = + runTxnsWithOrder__A_Start__B__C__A_End(txnA, txnB, txnC) + ThreadUtils.awaitResult(futureA, Duration.Inf) + ThreadUtils.awaitResult(futureB, Duration.Inf) + ThreadUtils.awaitResult(futureC, Duration.Inf) + + assert(ids(dir) === (0L to 99L).filterNot(id => Set(10L, 20L, 30L).contains(id))) + assert(deletionVectorCardinalities(log) === Seq(3L)) + } + } + + // --------------------------------------------------------------------------- + // Partitioned table (DV merge is partition-agnostic) + // --------------------------------------------------------------------------- + + test("disjoint concurrent DELETEs on a partitioned table's file both commit") { + withTempDir { dir => + // Single partition p=0 with a single data file. + spark.range(start = 0, end = 100, step = 1, numPartitions = 1) + .withColumn("p", lit(0)) + .write.partitionBy("p").format("delta").mode("append").save(dir.getAbsolutePath) + val log = DeltaLog.forTable(spark, dir.getCanonicalPath) + assert(log.update().allFiles.collect().length === 1) + + val txnA = sqlTxn(s"DELETE FROM ${tableRef(dir)} WHERE id = 10", rowLevelConcurrency = true) + val txnB = sqlTxn(s"DELETE FROM ${tableRef(dir)} WHERE id = 20", rowLevelConcurrency = true) + + val (futureA, futureB) = runTxnsWithOrder__A_Start__B__A_End(txnA, txnB) + ThreadUtils.awaitResult(futureA, Duration.Inf) + ThreadUtils.awaitResult(futureB, Duration.Inf) + + assert(ids(dir) === (0L to 99L).filterNot(id => id == 10 || id == 20)) + assert(deletionVectorCardinalities(log) === Seq(2L)) + } + } +} From b64e025e48986f05b7ed4b074e5625e70d404026 Mon Sep 17 00:00:00 2001 From: Eunjin Song Date: Thu, 30 Jul 2026 15:25:56 -0700 Subject: [PATCH 02/11] [Spark] Narrow row-level concurrency to sound same-file DV union Remove the unsound "Layer 2" append-suppression for rewrite-only DML winners. Keying the append-check skip on the winner's op type (DELETE/UPDATE) is unsound: an UPDATE can move a row into the loser's predicate (winner `SET x=15`, loser `DELETE WHERE x>10`, row was x=5), a genuine write-skew that suppressing the winner's image file would hide. `canSkipAddedFileForRowLevelConcurrency` now skips only the same-file DV union paths (`rowLevelResolvedPaths`). A rewrite-only DML winner's new image files flow into the standard added-files check (conservative abort, one-way safe); reconciling the provably-disjoint case via conflict-time data skipping is owned by Case 1 (#4/#8). Tests reframed to the Layer-1 envelope (9 tests, all passing): the two "both commit" UPDATE cases move to Case 1; DELETE-vs-UPDATE now asserts the image file conservatively conflicts. Co-Authored-By: Claude Opus 4.8 --- .../spark/sql/delta/ConflictChecker.scala | 38 +++++++-------- .../sql/delta/RowLevelConcurrencySuite.scala | 48 ++++++++----------- 2 files changed, 37 insertions(+), 49 deletions(-) diff --git a/spark/src/main/scala/org/apache/spark/sql/delta/ConflictChecker.scala b/spark/src/main/scala/org/apache/spark/sql/delta/ConflictChecker.scala index b77bca2f9f1..cb54f3449bb 100644 --- a/spark/src/main/scala/org/apache/spark/sql/delta/ConflictChecker.scala +++ b/spark/src/main/scala/org/apache/spark/sql/delta/ConflictChecker.scala @@ -22,7 +22,7 @@ import java.util.concurrent.TimeUnit import scala.collection.mutable -import org.apache.spark.sql.delta.DeltaOperations.{OP_DELETE, OP_SET_TBLPROPERTIES, OP_UPDATE, ROW_TRACKING_BACKFILL_OPERATION_NAME, ROW_TRACKING_UNBACKFILL_OPERATION_NAME} +import org.apache.spark.sql.delta.DeltaOperations.{OP_SET_TBLPROPERTIES, ROW_TRACKING_BACKFILL_OPERATION_NAME, ROW_TRACKING_UNBACKFILL_OPERATION_NAME} import org.apache.spark.sql.delta.RowId.RowTrackingMetadataDomain import org.apache.spark.sql.delta.actions._ import org.apache.spark.sql.delta.catalog.DeltaTableV2 @@ -1144,32 +1144,26 @@ private[delta] class ConflictChecker( private lazy val winningOperationName: Option[String] = winningCommitSummary.commitInfo.map(_.operation) - /** - * Whether the winning commit is a row-level DML that only rewrites or removes existing rows - * (DELETE or UPDATE), i.e. it introduces no net-new logical rows. Files added by such a commit - * are rewrites of rows that the current transaction already observed in its read snapshot (or - * re-adds of files masked by a deletion vector), so they cannot introduce phantom rows for the - * current transaction. MERGE is intentionally excluded because it may insert net-new rows. - */ - private lazy val winningCommitIsRewriteOnlyRowLevelDml: Boolean = { - val usesDeletionVectors = winningCommitSummary.addedFiles.exists(_.deletionVector != null) - usesDeletionVectors && winningOperationName.exists { - case OP_DELETE | OP_UPDATE => true - case _ => false - } - } - /** * Whether a file added by the winning transaction can be skipped in the added-files (append) - * conflict check thanks to row-level concurrency resolution. This is true when: - * 1. the file's "same physical file" conflict was already reconciled by merging deletion - * vectors ([[rowLevelResolvedPaths]]), or - * 2. the winning commit is a rewrite-only row-level DML (DELETE/UPDATE) whose added files - * cannot introduce phantom rows (see [[winningCommitIsRewriteOnlyRowLevelDml]]). + * conflict check thanks to row-level concurrency resolution. + * + * This is true only when the file's "same physical file" conflict was already reconciled by + * merging deletion vectors ([[resolveRowLevelConflicts]] recorded the path in + * [[rowLevelResolvedPaths]]). In that case the winner's re-added `AddFile(P)` carries the winning + * DV that we already folded into the current transaction's merged DV, so re-checking it would be + * a false conflict. + * + * We deliberately do NOT skip a rewrite-only DML winner's *new image* files here (an UPDATE + * writes updated row values to a fresh path). Those are ordinary non-blind changed-data files + * and can legitimately conflict: e.g. an UPDATE can move a row *into* the loser's predicate + * (winner `SET x = 15`, loser `DELETE WHERE x > 10`, row was `x = 5`), a genuine write-skew that + * the DV union cannot detect. They are arbitrated by the standard added-files check (and, when + * enabled, by conflict-time data skipping over their stats). */ private def canSkipAddedFileForRowLevelConcurrency(addFile: AddFile): Boolean = { if (!rowLevelConcurrencyEnabled) return false - rowLevelResolvedPaths.contains(addFile.path) || winningCommitIsRewriteOnlyRowLevelDml + rowLevelResolvedPaths.contains(addFile.path) } /** diff --git a/spark/src/test/scala/org/apache/spark/sql/delta/RowLevelConcurrencySuite.scala b/spark/src/test/scala/org/apache/spark/sql/delta/RowLevelConcurrencySuite.scala index f5377fc38f3..c93b24ea870 100644 --- a/spark/src/test/scala/org/apache/spark/sql/delta/RowLevelConcurrencySuite.scala +++ b/spark/src/test/scala/org/apache/spark/sql/delta/RowLevelConcurrencySuite.scala @@ -37,6 +37,12 @@ import org.apache.spark.util.ThreadUtils * * Two concurrent DML operations that touch the same physical file but modify disjoint rows should * commit cleanly by merging their deletion vectors, instead of aborting the losing transaction. + * + * Scope note: this is the sound *same-file DV union*. A rewrite-only DML that also writes *new + * image* files (an UPDATE emits updated row values to a fresh path) is NOT reconciled here. Those + * image files are ordinary non-blind changed-data files that can carry a genuine conflict (a + * value-flip write-skew), so they fall back to today's abort. Reconciling them on proven + * non-overlap is conflict-time data skipping, owned by Case 1 (fork issue #4). */ class RowLevelConcurrencySuite extends QueryTest with SharedSparkSession @@ -138,7 +144,7 @@ class RowLevelConcurrencySuite extends QueryTest ThreadUtils.awaitResult(futureB, Duration.Inf) val e = intercept[SparkException] { ThreadUtils.awaitResult(futureA, Duration.Inf) } assertConcurrentModificationException(e) - // Only the winner's delete (id=20) is applied; DVs are still used, so its DV has cardinality 1. + // Only the winner's delete (id=20) is applied; DVs still used, so its DV cardinality is 1. assert(ids(dir) === (0L to 99L).filterNot(_ == 20)) assert(deletionVectorCardinalities(log) === Seq(1L)) } @@ -165,10 +171,12 @@ class RowLevelConcurrencySuite extends QueryTest } // --------------------------------------------------------------------------- - // Layer 2: UPDATE (rewrite-only DML adds new data files) + // UPDATE image files: rewrite-only DML writes new data files that Layer-1 (DV union) does not + // reconcile. They conservatively conflict here; conflict-time data skipping (Case 1 / #4) + // reconciles the provably-disjoint case. // --------------------------------------------------------------------------- - test("disjoint DELETE (loser) vs UPDATE (winner) both commit") { + test("DELETE (loser) vs UPDATE (winner): winner image file conservatively conflicts") { withTempDir { dir => val log = createSingleFileTableWithDVs(dir) // A (loser) deletes id=10; B (winner) updates id=20 -> 1020 (masks row 20, appends image). @@ -177,31 +185,17 @@ class RowLevelConcurrencySuite extends QueryTest rowLevelConcurrency = true) val (futureA, futureB) = runTxnsWithOrder__A_Start__B__A_End(txnA, txnB) - ThreadUtils.awaitResult(futureA, Duration.Inf) ThreadUtils.awaitResult(futureB, Duration.Inf) - - assert(ids(dir) === ((0L to 99L).filterNot(id => id == 10 || id == 20) :+ 1020L).sorted) - // The original file's merged DV masks rows 10 and 20; the updated image lives in a new file. - assert(deletionVectorCardinalities(log) === Seq(2L)) - } - } - - test("disjoint UPDATE vs UPDATE both commit") { - withTempDir { dir => - val log = createSingleFileTableWithDVs(dir) - val txnA = sqlTxn(s"UPDATE ${tableRef(dir)} SET id = 1010 WHERE id = 10", - rowLevelConcurrency = true) - val txnB = sqlTxn(s"UPDATE ${tableRef(dir)} SET id = 1020 WHERE id = 20", - rowLevelConcurrency = true) - - val (futureA, futureB) = runTxnsWithOrder__A_Start__B__A_End(txnA, txnB) - ThreadUtils.awaitResult(futureA, Duration.Inf) - ThreadUtils.awaitResult(futureB, Duration.Inf) - - assert(ids(dir) === - ((0L to 99L).filterNot(id => id == 10 || id == 20) ++ Seq(1010L, 1020L)).sorted) - // Original file's merged DV masks rows 10 and 20; updated images live in two new files. - assert(deletionVectorCardinalities(log) === Seq(2L)) + // The shared file's DVs are disjoint and would merge, but the winner's UPDATE also writes an + // *image* file (new path) that the append check cannot prove disjoint from the loser's read + // without conflict-time data skipping (Case 1 / #4). It conservatively conflicts, so the + // loser aborts. This is one-way safe: it never reconciles a genuine value-flip write-skew + // (e.g. winner `SET x = 15`, loser `DELETE WHERE x > 10`), which the DV union cannot detect. + val e = intercept[SparkException] { ThreadUtils.awaitResult(futureA, Duration.Inf) } + assertConcurrentModificationException(e) + // Loser aborted cleanly: only the winner's update is applied (row 20 masked, 1020 appended). + assert(ids(dir) === ((0L to 99L).filterNot(_ == 20) :+ 1020L).sorted) + assert(deletionVectorCardinalities(log) === Seq(1L)) } } From 55f07ab7e8ecfd6ab87422bb5ac4931d534694d3 Mon Sep 17 00:00:00 2001 From: Eunjin Song Date: Mon, 3 Aug 2026 10:59:42 -0700 Subject: [PATCH 03/11] [Spark] Harden row-level DV-merge concurrency tests + scrub internal refs Add coverage for the deletion-vector-merge row-level concurrency path: - non-empty base DV: DELETE id=5 first, then two disjoint concurrent DELETEs, exercising the `(dv_win INTERSECT dv_cur) MINUS base` subtraction that every prior test left empty. - 3-way where the last (overlapping) txn aborts while the two disjoint ones commit. - disjoint DELETEs reconcile under explicit Serializable isolation. - row tracking: surviving rows keep their stable `_metadata.row_id` across the same-file DV merge. - change data feed: each deleted row is emitted once, attributed to the version that deleted it (winner at N, reconciled txn at N+1). Also scrub fork-internal taxonomy from the suite comments (drop "Case 1 / fork issue #4" and "Layer-1" wording) in favor of self-contained descriptions ("conflict-time reader-side data skipping (a separate change)", "same-file DV union"). No behavior change. Co-Authored-By: Claude Opus 4.8 --- .../sql/delta/RowLevelConcurrencySuite.scala | 168 ++++++++++++++++-- 1 file changed, 158 insertions(+), 10 deletions(-) diff --git a/spark/src/test/scala/org/apache/spark/sql/delta/RowLevelConcurrencySuite.scala b/spark/src/test/scala/org/apache/spark/sql/delta/RowLevelConcurrencySuite.scala index c93b24ea870..6ee50ce2519 100644 --- a/spark/src/test/scala/org/apache/spark/sql/delta/RowLevelConcurrencySuite.scala +++ b/spark/src/test/scala/org/apache/spark/sql/delta/RowLevelConcurrencySuite.scala @@ -42,7 +42,7 @@ import org.apache.spark.util.ThreadUtils * image* files (an UPDATE emits updated row values to a fresh path) is NOT reconciled here. Those * image files are ordinary non-blind changed-data files that can carry a genuine conflict (a * value-flip write-skew), so they fall back to today's abort. Reconciling them on proven - * non-overlap is conflict-time data skipping, owned by Case 1 (fork issue #4). + * non-overlap is conflict-time reader-side data skipping (a separate change). */ class RowLevelConcurrencySuite extends QueryTest with SharedSparkSession @@ -56,10 +56,17 @@ class RowLevelConcurrencySuite extends QueryTest private def tableRef(dir: File): String = s"delta.`${dir.getCanonicalPath}`" - /** Creates a single-file Delta table with `id` in [0, numRows) and deletion vectors enabled. */ - private def createSingleFileTableWithDVs(dir: File, numRows: Int = 100): DeltaLog = { + /** + * Creates a single-file Delta table with `id` in [0, numRows) and deletion vectors enabled. + * `extraProperties` are applied as `delta.*` table properties at creation (e.g. row tracking or + * change data feed). + */ + private def createSingleFileTableWithDVs( + dir: File, + numRows: Int = 100, + extraProperties: Map[String, String] = Map.empty): DeltaLog = { spark.range(start = 0, end = numRows, step = 1, numPartitions = 1) - .write.format("delta").mode("append").save(dir.getAbsolutePath) + .write.format("delta").options(extraProperties).mode("append").save(dir.getAbsolutePath) val log = DeltaLog.forTable(spark, dir.getCanonicalPath) val snapshot = log.update() assert( @@ -118,6 +125,48 @@ class RowLevelConcurrencySuite extends QueryTest } } + test("disjoint concurrent DELETEs reconcile on a file that already carries a base DV") { + withTempDir { dir => + val log = createSingleFileTableWithDVs(dir) + // Establish a non-empty base DV: delete id=5 and commit (the file now carries a DV of + // cardinality 1). Both concurrent txns below read this DV as their common base, so the + // overlap test must subtract it (`(dv_win INTERSECT dv_cur) MINUS base`); without that + // subtraction the shared row 5 would look like a false conflict and abort the merge. + sql(s"DELETE FROM ${tableRef(dir)} WHERE id = 5") + assert(deletionVectorCardinalities(log) === Seq(1L)) + + val txnA = sqlTxn(s"DELETE FROM ${tableRef(dir)} WHERE id = 10", rowLevelConcurrency = true) + val txnB = sqlTxn(s"DELETE FROM ${tableRef(dir)} WHERE id = 20", rowLevelConcurrency = true) + + val (futureA, futureB) = runTxnsWithOrder__A_Start__B__A_End(txnA, txnB) + ThreadUtils.awaitResult(futureA, Duration.Inf) + ThreadUtils.awaitResult(futureB, Duration.Inf) + + assert(ids(dir) === (0L to 99L).filterNot(id => Set(5L, 10L, 20L).contains(id))) + // Base row 5 plus the two disjoint new deletes -> merged deletion vector cardinality 3. + assert(deletionVectorCardinalities(log) === Seq(3L)) + } + } + + test("disjoint concurrent DELETEs reconcile under Serializable isolation") { + withTempDir { dir => + val log = createSingleFileTableWithDVs(dir) + sql(s"ALTER TABLE ${tableRef(dir)} SET TBLPROPERTIES " + + s"('${DeltaConfigs.ISOLATION_LEVEL.key}' = 'Serializable')") + // The `current ; winner` schedule of two disjoint-row deletes is a valid serialization even + // under Serializable (neither read a row the other wrote), so the DV union still commits. + val txnA = sqlTxn(s"DELETE FROM ${tableRef(dir)} WHERE id = 10", rowLevelConcurrency = true) + val txnB = sqlTxn(s"DELETE FROM ${tableRef(dir)} WHERE id = 20", rowLevelConcurrency = true) + + val (futureA, futureB) = runTxnsWithOrder__A_Start__B__A_End(txnA, txnB) + ThreadUtils.awaitResult(futureA, Duration.Inf) + ThreadUtils.awaitResult(futureB, Duration.Inf) + + assert(ids(dir) === (0L to 99L).filterNot(id => id == 10 || id == 20)) + assert(deletionVectorCardinalities(log) === Seq(2L)) + } + } + test("overlapping concurrent DELETEs still conflict") { withTempDir { dir => val log = createSingleFileTableWithDVs(dir) @@ -171,9 +220,9 @@ class RowLevelConcurrencySuite extends QueryTest } // --------------------------------------------------------------------------- - // UPDATE image files: rewrite-only DML writes new data files that Layer-1 (DV union) does not - // reconcile. They conservatively conflict here; conflict-time data skipping (Case 1 / #4) - // reconciles the provably-disjoint case. + // UPDATE image files: rewrite-only DML writes new data files that same-file DV union does not + // reconcile. They conservatively conflict here; conflict-time reader-side data skipping (a + // separate change) reconciles the provably-disjoint case. // --------------------------------------------------------------------------- test("DELETE (loser) vs UPDATE (winner): winner image file conservatively conflicts") { @@ -188,9 +237,10 @@ class RowLevelConcurrencySuite extends QueryTest ThreadUtils.awaitResult(futureB, Duration.Inf) // The shared file's DVs are disjoint and would merge, but the winner's UPDATE also writes an // *image* file (new path) that the append check cannot prove disjoint from the loser's read - // without conflict-time data skipping (Case 1 / #4). It conservatively conflicts, so the - // loser aborts. This is one-way safe: it never reconciles a genuine value-flip write-skew - // (e.g. winner `SET x = 15`, loser `DELETE WHERE x > 10`), which the DV union cannot detect. + // without conflict-time reader-side data skipping (a separate change). It conservatively + // conflicts, so the loser aborts. This is one-way safe: it never reconciles a genuine + // value-flip write-skew (e.g. winner `SET x = 15`, loser `DELETE WHERE x > 10`), which the + // DV union cannot detect. val e = intercept[SparkException] { ThreadUtils.awaitResult(futureA, Duration.Inf) } assertConcurrentModificationException(e) // Loser aborted cleanly: only the winner's update is applied (row 20 masked, 1020 appended). @@ -264,6 +314,30 @@ class RowLevelConcurrencySuite extends QueryTest } } + test("three concurrent DELETEs: two disjoint commit, the overlapping last one aborts") { + withTempDir { dir => + val log = createSingleFileTableWithDVs(dir) + // B (id=20) and C (id=10) touch disjoint rows and both commit. A also deletes id=10 and, + // committing last, reconciles cleanly against B (disjoint) but then discovers its overlap + // with C on row 10, so it aborts. + val txnA = sqlTxn(s"DELETE FROM ${tableRef(dir)} WHERE id = 10", rowLevelConcurrency = true) + val txnB = sqlTxn(s"DELETE FROM ${tableRef(dir)} WHERE id = 20", rowLevelConcurrency = true) + val txnC = sqlTxn(s"DELETE FROM ${tableRef(dir)} WHERE id = 10", rowLevelConcurrency = true) + + // A starts; B commits; C commits (reading B's state); A commits last. + val (futureA, futureB, futureC) = + runTxnsWithOrder__A_Start__B__C__A_End(txnA, txnB, txnC) + ThreadUtils.awaitResult(futureB, Duration.Inf) + ThreadUtils.awaitResult(futureC, Duration.Inf) + val e = intercept[SparkException] { ThreadUtils.awaitResult(futureA, Duration.Inf) } + assertConcurrentModificationException(e) + + // Only B (id=20) and C (id=10) applied; A aborted cleanly on the id=10 overlap. + assert(ids(dir) === (0L to 99L).filterNot(id => id == 10 || id == 20)) + assert(deletionVectorCardinalities(log) === Seq(2L)) + } + } + // --------------------------------------------------------------------------- // Partitioned table (DV merge is partition-agnostic) // --------------------------------------------------------------------------- @@ -288,4 +362,78 @@ class RowLevelConcurrencySuite extends QueryTest assert(deletionVectorCardinalities(log) === Seq(2L)) } } + + // --------------------------------------------------------------------------- + // Row tracking: the merge keeps the same physical file, so stable row IDs are preserved + // --------------------------------------------------------------------------- + + test("row tracking: reconciled disjoint DELETEs preserve surviving rows' stable row IDs") { + withTempDir { dir => + val log = createSingleFileTableWithDVs(dir, + extraProperties = Map(DeltaConfigs.ROW_TRACKING_ENABLED.key -> "true")) + // Snapshot each row's stable row id before the concurrent deletes. + val before = spark.read.format("delta").load(dir.getAbsolutePath) + .select("id", "_metadata.row_id") + .collect().map(r => r.getLong(0) -> r.getLong(1)).toMap + + val txnA = sqlTxn(s"DELETE FROM ${tableRef(dir)} WHERE id = 10", rowLevelConcurrency = true) + val txnB = sqlTxn(s"DELETE FROM ${tableRef(dir)} WHERE id = 20", rowLevelConcurrency = true) + + val (futureA, futureB) = runTxnsWithOrder__A_Start__B__A_End(txnA, txnB) + ThreadUtils.awaitResult(futureA, Duration.Inf) + ThreadUtils.awaitResult(futureB, Duration.Inf) + + assert(ids(dir) === (0L to 99L).filterNot(id => id == 10 || id == 20)) + assert(deletionVectorCardinalities(log) === Seq(2L)) + + val after = spark.read.format("delta").load(dir.getAbsolutePath) + .select("id", "_metadata.row_id") + .collect().map(r => r.getLong(0) -> r.getLong(1)).toMap + // Merging DVs on the same physical file leaves base row IDs untouched, so every surviving + // row keeps the exact stable row id it had before the concurrent deletes. + assert(after === (before -- Seq(10L, 20L))) + } + } + + // --------------------------------------------------------------------------- + // Change Data Feed: each deleted row is emitted once, at the version that deleted it + // --------------------------------------------------------------------------- + + test("change data feed: reconciled disjoint DELETEs each emit one delete at their own version") { + withTempDir { dir => + val log = createSingleFileTableWithDVs(dir, + extraProperties = Map(DeltaConfigs.CHANGE_DATA_FEED.key -> "true")) + // Table creation is version 0; the winner commits at version 1 and the reconciled current + // txn at version 2. + val firstDeleteVersion = log.update().version + 1 + + // A starts first but commits last, so B (id=20) is the winner at version 1 and A (id=10) is + // the reconciled current txn at version 2. + val txnA = sqlTxn(s"DELETE FROM ${tableRef(dir)} WHERE id = 10", rowLevelConcurrency = true) + val txnB = sqlTxn(s"DELETE FROM ${tableRef(dir)} WHERE id = 20", rowLevelConcurrency = true) + + val (futureA, futureB) = runTxnsWithOrder__A_Start__B__A_End(txnA, txnB) + ThreadUtils.awaitResult(futureA, Duration.Inf) + ThreadUtils.awaitResult(futureB, Duration.Inf) + + assert(ids(dir) === (0L to 99L).filterNot(id => id == 10 || id == 20)) + assert(deletionVectorCardinalities(log) === Seq(2L)) + + val changes = spark.read.format("delta") + .option("readChangeFeed", "true") + .option("startingVersion", firstDeleteVersion) + .load(dir.getAbsolutePath) + .select("id", "_change_type", "_commit_version") + .where("_change_type = 'delete'") + .collect() + .map(r => (r.getLong(0), r.getString(1), r.getLong(2))) + .sortBy(_._1) + .toSeq + // Each deleted row appears exactly once, attributed to the version that deleted it: the + // winner's id=20 at version 1 and the reconciled txn's id=10 at version 2. + assert(changes === Seq( + (10L, "delete", firstDeleteVersion + 1), + (20L, "delete", firstDeleteVersion))) + } + } } From 538ef84c3fd8409f49a44799a33ab0aab83f221c Mon Sep 17 00:00:00 2001 From: Eunjin Song Date: Tue, 4 Aug 2026 11:25:00 -0700 Subject: [PATCH 04/11] Extract row-level concurrency resolution into a self-typed trait Move the RLC deletion-vector merge (rowLevelConcurrencyEnabled, winningOperationName, canSkipAddedFileForRowLevelConcurrency, resolveRowLevelConflicts, the DV read/write helpers, and rowLevelResolvedPaths) out of the 1862-line ConflictChecker into RowLevelConcurrencyResolution -- a `self: ConflictChecker =>` trait mixed into the checker. ConflictChecker keeps only the call sites; `spark` and `winningCommitSummary` are widened to `protected val`, and the DV helpers stay `protected` so the OPTIMIZE-vs-DML reconciliation (which mixes into the same checker) can reuse them. Also add a concrete worked example to the resolveRowLevelConflicts doc: winner DELETEs {5,10}, current DELETEs {20,21}, disjoint over base {} -> merge to {5,10,20,21}; had current deleted row 5, overlap {5} -> abort. Co-Authored-By: Claude Opus 4.8 --- .../spark/sql/delta/ConflictChecker.scala | 212 +-------------- .../delta/RowLevelConcurrencyResolution.scala | 252 ++++++++++++++++++ 2 files changed, 257 insertions(+), 207 deletions(-) create mode 100644 spark/src/main/scala/org/apache/spark/sql/delta/RowLevelConcurrencyResolution.scala diff --git a/spark/src/main/scala/org/apache/spark/sql/delta/ConflictChecker.scala b/spark/src/main/scala/org/apache/spark/sql/delta/ConflictChecker.scala index cb54f3449bb..cf7a0446bdc 100644 --- a/spark/src/main/scala/org/apache/spark/sql/delta/ConflictChecker.scala +++ b/spark/src/main/scala/org/apache/spark/sql/delta/ConflictChecker.scala @@ -17,7 +17,6 @@ package org.apache.spark.sql.delta // scalastyle:off import.ordering.noEmptyLine -import java.util.UUID import java.util.concurrent.TimeUnit import scala.collection.mutable @@ -26,19 +25,16 @@ import org.apache.spark.sql.delta.DeltaOperations.{OP_SET_TBLPROPERTIES, ROW_TRA import org.apache.spark.sql.delta.RowId.RowTrackingMetadataDomain import org.apache.spark.sql.delta.actions._ import org.apache.spark.sql.delta.catalog.DeltaTableV2 -import org.apache.spark.sql.delta.commands.DeletionVectorUtils -import org.apache.spark.sql.delta.deletionvectors.{RoaringBitmapArray, RoaringBitmapArrayFormat} import org.apache.spark.sql.delta.logging.DeltaLogKeys import org.apache.spark.sql.delta.metering.DeltaLogging import org.apache.spark.sql.delta.sources.DeltaSourceUtils import org.apache.spark.sql.delta.sources.DeltaSQLConf -import org.apache.spark.sql.delta.storage.dv.DeletionVectorStore import org.apache.spark.sql.delta.util.DeltaSparkPlanUtils.CheckDeterministicOptions import org.apache.spark.sql.delta.util.FileNames import io.delta.storage.commit.UpdatedActions import io.delta.storage.commit.uccommitcoordinator.UCCommitCoordinatorClient import io.delta.storage.commit.uniform.UniformMetadata -import org.apache.hadoop.fs.{FileStatus, Path} +import org.apache.hadoop.fs.FileStatus import org.apache.spark.internal.{MDC, MessageWithContext} import org.apache.spark.sql.{DataFrame, SparkSession} @@ -220,11 +216,12 @@ object WinningCommitSummary { } private[delta] class ConflictChecker( - spark: SparkSession, + protected val spark: SparkSession, initialCurrentTransactionInfo: CurrentTransactionInfo, - winningCommitSummary: WinningCommitSummary, + protected val winningCommitSummary: WinningCommitSummary, isolationLevel: IsolationLevel) - extends DeltaLogging with ConflictCheckerPredicateElimination { + extends DeltaLogging with ConflictCheckerPredicateElimination + with RowLevelConcurrencyResolution { protected val winningCommitVersion = winningCommitSummary.commitVersion protected val startTimeMs = System.currentTimeMillis() @@ -233,13 +230,6 @@ private[delta] class ConflictChecker( protected var currentTransactionInfo: CurrentTransactionInfo = initialCurrentTransactionInfo - /** - * Paths of files whose "same physical file" conflict with the winning transaction was resolved at - * the row level by [[resolveRowLevelConflicts]] (deletion vectors merged). The file-level delete - * and append checks skip these paths, since they have already been reconciled. - */ - private val rowLevelResolvedPaths = mutable.Set.empty[String] - protected def recordSkippedPhase(phase: String): Unit = timingStats += phase -> 0 /** @@ -1134,198 +1124,6 @@ private[delta] class ConflictChecker( false } - /** Whether row-level concurrency resolution is enabled and applicable to this table. */ - private lazy val rowLevelConcurrencyEnabled: Boolean = - spark.conf.get(DeltaSQLConf.DELTA_ROW_LEVEL_CONCURRENCY_ENABLED) && - DeletionVectorUtils.deletionVectorsWritable( - currentTransactionInfo.protocol, currentTransactionInfo.metadata) - - /** The operation name of the winning commit, if available. */ - private lazy val winningOperationName: Option[String] = - winningCommitSummary.commitInfo.map(_.operation) - - /** - * Whether a file added by the winning transaction can be skipped in the added-files (append) - * conflict check thanks to row-level concurrency resolution. - * - * This is true only when the file's "same physical file" conflict was already reconciled by - * merging deletion vectors ([[resolveRowLevelConflicts]] recorded the path in - * [[rowLevelResolvedPaths]]). In that case the winner's re-added `AddFile(P)` carries the winning - * DV that we already folded into the current transaction's merged DV, so re-checking it would be - * a false conflict. - * - * We deliberately do NOT skip a rewrite-only DML winner's *new image* files here (an UPDATE - * writes updated row values to a fresh path). Those are ordinary non-blind changed-data files - * and can legitimately conflict: e.g. an UPDATE can move a row *into* the loser's predicate - * (winner `SET x = 15`, loser `DELETE WHERE x > 10`, row was `x = 5`), a genuine write-skew that - * the DV union cannot detect. They are arbitrated by the standard added-files check (and, when - * enabled, by conflict-time data skipping over their stats). - */ - private def canSkipAddedFileForRowLevelConcurrency(addFile: AddFile): Boolean = { - if (!rowLevelConcurrencyEnabled) return false - rowLevelResolvedPaths.contains(addFile.path) - } - - /** - * Resolves "same physical file" conflicts with the winning transaction at the row level. - * - * A DV-based DELETE/UPDATE emits, for each touched file `P`, a `RemoveFile(P)` (tombstone of the - * pre-image) and an `AddFile(P)` carrying a larger deletion vector. When both the winning and the - * current transaction touch the same file `P` this way, the two operations are logically - * independent as long as they mark *different* rows deleted. For every such shared file we: - * 1. decode the winning DV, the current DV and their common base DV (from the pre-image - * `RemoveFile`) as [[RoaringBitmapArray]]s; - * 2. check whether the newly-deleted rows are disjoint, i.e. `(dv_win INTERSECT dv_cur) MINUS - * dv_base` is empty. If they overlap, this is a genuine row-level conflict and we leave the - * file for the standard checks to abort; - * 3. on disjoint sets, merge the DVs (`dv_win UNION dv_cur`), persist a new DV file, - * and rebase the current transaction onto the winner's post-image: the current `AddFile(P)` - * now carries the merged DV and the current `RemoveFile(P)` now tombstones the winner's - * `AddFile(P)`. - * - * Resolved paths are recorded in [[rowLevelResolvedPaths]] and skipped by the file-level delete - * and append checks. Row identity is preserved for free: the merged file is the same physical - * file, so its base row ID is unchanged and [[reassignOverlappingRowIds]] (already run) leaves it - * alone. Deletion vectors index physical row positions within one immutable Parquet file, so the - * merge needs no row tracking. - */ - private def resolveRowLevelConflicts(): Unit = { - if (!rowLevelConcurrencyEnabled) return - - // Winning transaction's DV updates: path present in both an AddFile (with a DV) and a - // RemoveFile of the winning commit. - val winningRemovedPaths = winningCommitSummary.removedFiles.map(_.path).toSet - val winningDvUpdates: Map[String, AddFile] = winningCommitSummary.addedFiles.iterator - .filter(a => a.deletionVector != null && winningRemovedPaths.contains(a.path)) - .map(a => a.path -> a) - .toMap - if (winningDvUpdates.isEmpty) return - - // Current transaction's DV updates, indexed by path. - val currentAddByPath = currentTransactionInfo.actions.collect { - case a: AddFile if a.deletionVector != null => a.path -> a - }.toMap - val currentRemoveByPath = currentTransactionInfo.actions.collect { - case r: RemoveFile => r.path -> r - }.toMap - - val sharedPaths = winningDvUpdates.keySet - .intersect(currentAddByPath.keySet) - .intersect(currentRemoveByPath.keySet) - if (sharedPaths.isEmpty) return - - recordTime("resolved-row-level-conflicts") { - val dvStore = DeletionVectorStore.createInstance(deltaLog.newDeltaHadoopConf()) - val tablePath = deltaLog.dataPath - - // path -> (rebased AddFile, rebased RemoveFile) - val replacements = mutable.Map.empty[String, (AddFile, RemoveFile)] - for (path <- sharedPaths) { - val winningAdd = winningDvUpdates(path) - val currentAdd = currentAddByPath(path) - val currentRemove = currentRemoveByPath(path) - - def bitmapOf(dv: DeletionVectorDescriptor): RoaringBitmapArray = - readDeletionVectorOrEmpty(dvStore, dv, tablePath) - val baseBitmap = bitmapOf(currentRemove.deletionVector) - val winningBitmap = bitmapOf(winningAdd.deletionVector) - val currentBitmap = bitmapOf(currentAdd.deletionVector) - - // `baseBitmap` is the DV of `P` at the current txn's read time (carried on its RemoveFile); - // for a 2-way conflict it equals the winner's pre-image too. Both `dv_win` and `dv_cur` are - // supersets of it (a DV only grows), so the newly-deleted rows are `dv \ base` on each side - // and their overlap is `(dv_win INTERSECT dv_cur) MINUS base`. If empty, the two txns - // touched disjoint rows and the schedule `current ; winner` is a valid serialization under - // both WriteSerializable and Serializable (the winner's rewrites/deletes are of rows the - // current txn did not touch), so merging is safe. If non-empty, the same row was touched by - // both -> genuine conflict, left for the standard checks. (For 3+ way chains `base` becomes - // previous winner's DV rather than the original pre-image; the merge stays correct and the - // overlap test stays conservative.) - val newlyDeletedOverlap = winningBitmap.copy() - newlyDeletedOverlap.and(currentBitmap) - newlyDeletedOverlap.andNot(baseBitmap) - - if (newlyDeletedOverlap.isEmpty) { - // Disjoint: merge the deletion vectors and rebase onto the winner's post-image. - val mergedBitmap = winningBitmap.copy() - mergedBitmap.merge(currentBitmap) - val mergedDescriptor = writeMergedDeletionVector(dvStore, tablePath, mergedBitmap) - // Keep the current AddFile's identity (base row ID / default row commit version already - // reconciled by the row-ID phases) but point it at the merged DV. - val rebasedAdd = currentAdd - .copy(deletionVector = mergedDescriptor, dataChange = true) - .withoutTightBoundStats - // Tombstone the winner's now-live AddFile (carries the winning DV) instead of the stale - // pre-image. - val rebasedRemove = winningAdd.removeWithTimestamp() - replacements(path) = (rebasedAdd, rebasedRemove) - rowLevelResolvedPaths += path - } - // else: overlapping row-level modification -> genuine conflict, leave for standard checks. - } - - if (replacements.nonEmpty) { - val newActions = currentTransactionInfo.actions.map { - case a: AddFile if replacements.contains(a.path) => replacements(a.path)._1 - case r: RemoveFile if replacements.contains(r.path) => replacements(r.path)._2 - case other => other - } - // Resolved files are no longer "read" for the purposes of the delete-read check. - val newReadFiles = currentTransactionInfo.readFiles - .filterNot(f => rowLevelResolvedPaths.contains(f.path)) - currentTransactionInfo = - currentTransactionInfo.copy(actions = newActions, readFiles = newReadFiles) - - recordDeltaEvent( - deltaLog, - opType = "delta.rowLevelConcurrency.deletionVectorsMerged", - data = Map( - "winningCommitVersion" -> winningCommitVersion, - "resolvedPaths" -> rowLevelResolvedPaths.size, - "winningOperation" -> winningOperationName.getOrElse("UNKNOWN"))) - } - } - } - - /** Reads a deletion vector into a [[RoaringBitmapArray]], returning an empty bitmap for none. */ - private def readDeletionVectorOrEmpty( - dvStore: DeletionVectorStore, - dv: DeletionVectorDescriptor, - tablePath: Path): RoaringBitmapArray = { - if (dv == null || dv.isEmpty) new RoaringBitmapArray() else dvStore.read(dv, tablePath) - } - - /** - * Persists a merged bitmap to a new deletion vector file and returns its descriptor. - * - * NOTE: this writes a DV file as a side effect of conflict resolution. If the commit ultimately - * fails or is retried against another winning version, the file is orphaned and later reclaimed - * by VACUUM (same lifecycle as any DV written by DML). This mirrors how the DML write path - * persists DVs (see `DeletionVectorWriter.storeSerializedBitmap`). - */ - private def writeMergedDeletionVector( - dvStore: DeletionVectorStore, - tablePath: Path, - bitmap: RoaringBitmapArray): DeletionVectorDescriptor = { - // An empty DV has no on-disk representation (matches DeletionVectorWriter). - if (bitmap.isEmpty) return DeletionVectorDescriptor.EMPTY - val tablePathWithFs = dvStore.pathWithFileSystem(tablePath) - val fileId = UUID.randomUUID() - val writer = dvStore.createWriter(dvStore.generateFileNameInTable(tablePathWithFs, fileId)) - try { - val serialized = DeletionVectorUtils.serialize( - bitmap, RoaringBitmapArrayFormat.Portable, Some(tablePath)) - val range = writer.write(serialized) - DeletionVectorDescriptor.onDiskWithRelativePath( - id = fileId, - sizeInBytes = serialized.length, - cardinality = bitmap.cardinality, - offset = Some(range.offset)) - } finally { - writer.close() - } - } - /** * Check if the new files added by the already committed transactions should have been read by * the current transaction. diff --git a/spark/src/main/scala/org/apache/spark/sql/delta/RowLevelConcurrencyResolution.scala b/spark/src/main/scala/org/apache/spark/sql/delta/RowLevelConcurrencyResolution.scala new file mode 100644 index 00000000000..f343e78bc6e --- /dev/null +++ b/spark/src/main/scala/org/apache/spark/sql/delta/RowLevelConcurrencyResolution.scala @@ -0,0 +1,252 @@ +/* + * Copyright (2021) The Delta Lake Project Authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.spark.sql.delta + +import java.util.UUID + +import scala.collection.mutable + +import org.apache.spark.sql.delta.actions.{AddFile, DeletionVectorDescriptor, RemoveFile} +import org.apache.spark.sql.delta.commands.DeletionVectorUtils +import org.apache.spark.sql.delta.deletionvectors.{RoaringBitmapArray, RoaringBitmapArrayFormat} +import org.apache.spark.sql.delta.metering.DeltaLogging +import org.apache.spark.sql.delta.sources.DeltaSQLConf +import org.apache.spark.sql.delta.storage.dv.DeletionVectorStore +import org.apache.hadoop.fs.Path + +/** + * Row-level concurrency resolution for the [[ConflictChecker]]: instead of aborting a concurrent + * DV-based DELETE/UPDATE that touches the same physical files as the winning transaction, MERGE the + * two transactions' deletion vectors when they deleted disjoint rows. + * + * Mixed into [[ConflictChecker]] as a self-typed trait: it reads and rewrites the checker's + * transaction state (`currentTransactionInfo`, `winningCommitSummary`, `deltaLog`, `spark`) + * directly, and lives in its own file so ConflictChecker stays focused on file-level conflict + * detection. The two deletion-vector helpers are `protected` so the OPTIMIZE-vs-DML reconciliation + * (which mixes into the same checker) can reuse them. + */ +trait RowLevelConcurrencyResolution extends DeltaLogging { self: ConflictChecker => + + /** + * Paths of files whose "same physical file" conflict with the winning transaction was resolved at + * the row level by [[resolveRowLevelConflicts]] (deletion vectors merged). The file-level delete + * and append checks skip these paths, since they have already been reconciled. + */ + protected val rowLevelResolvedPaths = mutable.Set.empty[String] + + /** Whether row-level concurrency resolution is enabled and applicable to this table. */ + protected lazy val rowLevelConcurrencyEnabled: Boolean = + spark.conf.get(DeltaSQLConf.DELTA_ROW_LEVEL_CONCURRENCY_ENABLED) && + DeletionVectorUtils.deletionVectorsWritable( + currentTransactionInfo.protocol, currentTransactionInfo.metadata) + + /** The operation name of the winning commit, if available. */ + protected lazy val winningOperationName: Option[String] = + winningCommitSummary.commitInfo.map(_.operation) + + /** + * Whether a file added by the winning transaction can be skipped in the added-files (append) + * conflict check thanks to row-level concurrency resolution. + * + * This is true only when the file's "same physical file" conflict was already reconciled by + * merging deletion vectors ([[resolveRowLevelConflicts]] recorded the path in + * [[rowLevelResolvedPaths]]). In that case the winner's re-added `AddFile(P)` carries the winning + * DV that we already folded into the current transaction's merged DV, so re-checking it would be + * a false conflict. + * + * We deliberately do NOT skip a rewrite-only DML winner's *new image* files here (an UPDATE + * writes updated row values to a fresh path). Those are ordinary non-blind changed-data files + * and can legitimately conflict: e.g. an UPDATE can move a row *into* the loser's predicate + * (winner `SET x = 15`, loser `DELETE WHERE x > 10`, row was `x = 5`), a genuine write-skew that + * the DV union cannot detect. They are arbitrated by the standard added-files check (and, when + * enabled, by conflict-time data skipping over their stats). + */ + protected def canSkipAddedFileForRowLevelConcurrency(addFile: AddFile): Boolean = { + if (!rowLevelConcurrencyEnabled) return false + rowLevelResolvedPaths.contains(addFile.path) + } + + /** + * Resolves "same physical file" conflicts with the winning transaction at the row level. + * + * A DV-based DELETE/UPDATE emits, for each touched file `P`, a `RemoveFile(P)` (tombstone of the + * pre-image) and an `AddFile(P)` carrying a larger deletion vector. When both the winning and the + * current transaction touch the same file `P` this way, the two operations are logically + * independent as long as they mark *different* rows deleted. For every such shared file we: + * 1. decode the winning DV, the current DV and their common base DV (from the pre-image + * `RemoveFile`) as [[RoaringBitmapArray]]s; + * 2. check whether the newly-deleted rows are disjoint, i.e. `(dv_win INTERSECT dv_cur) MINUS + * dv_base` is empty. If they overlap, this is a genuine row-level conflict and we leave the + * file for the standard checks to abort; + * 3. on disjoint sets, merge the DVs (`dv_win UNION dv_cur`), persist a new DV file, + * and rebase the current transaction onto the winner's post-image: the current `AddFile(P)` + * now carries the merged DV and the current `RemoveFile(P)` now tombstones the winner's + * `AddFile(P)`. + * + * Worked example: file `P` holds rows 0..99, undeleted at the current txn's read time + * (`dv_base = {}`). The winner commits `DELETE WHERE id IN (5, 10)` so `dv_win = {5, 10}`; the + * current txn holds `DELETE WHERE id IN (20, 21)` so `dv_cur = {20, 21}`. The overlap is + * `({5,10} INTERSECT {20,21}) MINUS {} = {}` -> disjoint, so we merge to `dv = {5, 10, 20, 21}`, + * point the current `AddFile(P)` at it, and tombstone the winner's `AddFile(P)`; both deletes + * survive. Had the current txn instead deleted row 5 (`dv_cur = {5, 21}`), the overlap would be + * `{5}` -> the same row was deleted twice concurrently, a genuine conflict left for the standard + * checks to abort. (For a 3+ way chain `dv_base` is the previous winner's DV rather than the + * original pre-image, which keeps the overlap test conservative and the merge correct.) + * + * Resolved paths are recorded in [[rowLevelResolvedPaths]] and skipped by the file-level delete + * and append checks. Row identity is preserved for free: the merged file is the same physical + * file, so its base row ID is unchanged and [[reassignOverlappingRowIds]] (already run) leaves it + * alone. Deletion vectors index physical row positions within one immutable Parquet file, so the + * merge needs no row tracking. + */ + protected def resolveRowLevelConflicts(): Unit = { + if (!rowLevelConcurrencyEnabled) return + + // Winning transaction's DV updates: path present in both an AddFile (with a DV) and a + // RemoveFile of the winning commit. + val winningRemovedPaths = winningCommitSummary.removedFiles.map(_.path).toSet + val winningDvUpdates: Map[String, AddFile] = winningCommitSummary.addedFiles.iterator + .filter(a => a.deletionVector != null && winningRemovedPaths.contains(a.path)) + .map(a => a.path -> a) + .toMap + if (winningDvUpdates.isEmpty) return + + // Current transaction's DV updates, indexed by path. + val currentAddByPath = currentTransactionInfo.actions.collect { + case a: AddFile if a.deletionVector != null => a.path -> a + }.toMap + val currentRemoveByPath = currentTransactionInfo.actions.collect { + case r: RemoveFile => r.path -> r + }.toMap + + val sharedPaths = winningDvUpdates.keySet + .intersect(currentAddByPath.keySet) + .intersect(currentRemoveByPath.keySet) + if (sharedPaths.isEmpty) return + + recordTime("resolved-row-level-conflicts") { + val dvStore = DeletionVectorStore.createInstance(deltaLog.newDeltaHadoopConf()) + val tablePath = deltaLog.dataPath + + // path -> (rebased AddFile, rebased RemoveFile) + val replacements = mutable.Map.empty[String, (AddFile, RemoveFile)] + for (path <- sharedPaths) { + val winningAdd = winningDvUpdates(path) + val currentAdd = currentAddByPath(path) + val currentRemove = currentRemoveByPath(path) + + def bitmapOf(dv: DeletionVectorDescriptor): RoaringBitmapArray = + readDeletionVectorOrEmpty(dvStore, dv, tablePath) + val baseBitmap = bitmapOf(currentRemove.deletionVector) + val winningBitmap = bitmapOf(winningAdd.deletionVector) + val currentBitmap = bitmapOf(currentAdd.deletionVector) + + // `baseBitmap` is the DV of `P` at the current txn's read time (carried on its RemoveFile); + // for a 2-way conflict it equals the winner's pre-image too. Both `dv_win` and `dv_cur` are + // supersets of it (a DV only grows), so the newly-deleted rows are `dv \ base` on each side + // and their overlap is `(dv_win INTERSECT dv_cur) MINUS base`. If empty, the two txns + // touched disjoint rows and the schedule `current ; winner` is a valid serialization under + // both WriteSerializable and Serializable (the winner's rewrites/deletes are of rows the + // current txn did not touch), so merging is safe. If non-empty, the same row was touched by + // both -> genuine conflict, left for the standard checks. (For 3+ way chains `base` becomes + // previous winner's DV rather than the original pre-image; the merge stays correct and the + // overlap test stays conservative.) + val newlyDeletedOverlap = winningBitmap.copy() + newlyDeletedOverlap.and(currentBitmap) + newlyDeletedOverlap.andNot(baseBitmap) + + if (newlyDeletedOverlap.isEmpty) { + // Disjoint: merge the deletion vectors and rebase onto the winner's post-image. + val mergedBitmap = winningBitmap.copy() + mergedBitmap.merge(currentBitmap) + val mergedDescriptor = writeMergedDeletionVector(dvStore, tablePath, mergedBitmap) + // Keep the current AddFile's identity (base row ID / default row commit version already + // reconciled by the row-ID phases) but point it at the merged DV. + val rebasedAdd = currentAdd + .copy(deletionVector = mergedDescriptor, dataChange = true) + .withoutTightBoundStats + // Tombstone the winner's now-live AddFile (carries the winning DV) instead of the stale + // pre-image. + val rebasedRemove = winningAdd.removeWithTimestamp() + replacements(path) = (rebasedAdd, rebasedRemove) + rowLevelResolvedPaths += path + } + // else: overlapping row-level modification -> genuine conflict, leave for standard checks. + } + + if (replacements.nonEmpty) { + val newActions = currentTransactionInfo.actions.map { + case a: AddFile if replacements.contains(a.path) => replacements(a.path)._1 + case r: RemoveFile if replacements.contains(r.path) => replacements(r.path)._2 + case other => other + } + // Resolved files are no longer "read" for the purposes of the delete-read check. + val newReadFiles = currentTransactionInfo.readFiles + .filterNot(f => rowLevelResolvedPaths.contains(f.path)) + currentTransactionInfo = + currentTransactionInfo.copy(actions = newActions, readFiles = newReadFiles) + + recordDeltaEvent( + deltaLog, + opType = "delta.rowLevelConcurrency.deletionVectorsMerged", + data = Map( + "winningCommitVersion" -> winningCommitVersion, + "resolvedPaths" -> rowLevelResolvedPaths.size, + "winningOperation" -> winningOperationName.getOrElse("UNKNOWN"))) + } + } + } + + /** Reads a deletion vector into a [[RoaringBitmapArray]], returning an empty bitmap for none. */ + protected def readDeletionVectorOrEmpty( + dvStore: DeletionVectorStore, + dv: DeletionVectorDescriptor, + tablePath: Path): RoaringBitmapArray = { + if (dv == null || dv.isEmpty) new RoaringBitmapArray() else dvStore.read(dv, tablePath) + } + + /** + * Persists a merged bitmap to a new deletion vector file and returns its descriptor. + * + * NOTE: this writes a DV file as a side effect of conflict resolution. If the commit ultimately + * fails or is retried against another winning version, the file is orphaned and later reclaimed + * by VACUUM (same lifecycle as any DV written by DML). This mirrors how the DML write path + * persists DVs (see `DeletionVectorWriter.storeSerializedBitmap`). + */ + protected def writeMergedDeletionVector( + dvStore: DeletionVectorStore, + tablePath: Path, + bitmap: RoaringBitmapArray): DeletionVectorDescriptor = { + // An empty DV has no on-disk representation (matches DeletionVectorWriter). + if (bitmap.isEmpty) return DeletionVectorDescriptor.EMPTY + val tablePathWithFs = dvStore.pathWithFileSystem(tablePath) + val fileId = UUID.randomUUID() + val writer = dvStore.createWriter(dvStore.generateFileNameInTable(tablePathWithFs, fileId)) + try { + val serialized = DeletionVectorUtils.serialize( + bitmap, RoaringBitmapArrayFormat.Portable, Some(tablePath)) + val range = writer.write(serialized) + DeletionVectorDescriptor.onDiskWithRelativePath( + id = fileId, + sizeInBytes = serialized.length, + cardinality = bitmap.cardinality, + offset = Some(range.offset)) + } finally { + writer.close() + } + } +} From 1898a55dc524c2a402f85c6cd61db3a4d4c41ed2 Mon Sep 17 00:00:00 2001 From: Eunjin Song Date: Tue, 4 Aug 2026 12:09:27 -0700 Subject: [PATCH 05/11] [Spark] Skip added-file re-scan when no row-level conflict was reconciled canSkipAddedFileForRowLevelConcurrency can only return true for a path in rowLevelResolvedPaths, so when that set is empty the filterNot is a no-op. Guard on rowLevelResolvedPaths.isEmpty to skip the traversal and its allocation on the common no-conflict path. Co-Authored-By: Claude Opus 4.8 --- .../scala/org/apache/spark/sql/delta/ConflictChecker.scala | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/spark/src/main/scala/org/apache/spark/sql/delta/ConflictChecker.scala b/spark/src/main/scala/org/apache/spark/sql/delta/ConflictChecker.scala index cf7a0446bdc..9933b68715e 100644 --- a/spark/src/main/scala/org/apache/spark/sql/delta/ConflictChecker.scala +++ b/spark/src/main/scala/org/apache/spark/sql/delta/ConflictChecker.scala @@ -1144,8 +1144,12 @@ private[delta] class ConflictChecker( Seq.empty } + // Only re-scan the added files when row-level resolution actually reconciled something: a + // file is skippable only if its path is in `rowLevelResolvedPaths`, so an empty set means the + // filter is a no-op. Skips the traversal (and its allocation) on the common no-conflict path. val addedFilesAfterRowLevelResolution = - addedFilesToCheckForConflicts.filterNot(canSkipAddedFileForRowLevelConcurrency) + if (rowLevelResolvedPaths.isEmpty) addedFilesToCheckForConflicts + else addedFilesToCheckForConflicts.filterNot(canSkipAddedFileForRowLevelConcurrency) val fileMatchingPartitionReadPredicates = getFirstFileMatchingPartitionPredicates(addedFilesAfterRowLevelResolution) From 6a03f588300f53e3a3ea21c2d15e43ee83a23756 Mon Sep 17 00:00:00 2001 From: Eunjin Song Date: Tue, 4 Aug 2026 23:17:42 -0700 Subject: [PATCH 06/11] [Spark] Add MERGE matched-clause tests to RowLevelConcurrencySuite Extend the same-file deletion-vector union coverage to MERGE: - disjoint concurrent MERGE matched-deletes reconcile (DV union, like DELETE) - a MERGE matched-delete reconciles against a concurrent plain DELETE (the union is operation-agnostic) - overlapping MERGE matched-deletes still conflict - a MERGE matched-update (winner) writes an image file, so a concurrent loser conservatively conflicts, mirroring the standalone-UPDATE case MERGE inserts / WHEN NOT MATCHED BY SOURCE stay out of scope (they need per-file row-tracking classification) and are not exercised. Co-Authored-By: Claude Opus 4.8 --- .../sql/delta/RowLevelConcurrencySuite.scala | 84 +++++++++++++++++++ 1 file changed, 84 insertions(+) diff --git a/spark/src/test/scala/org/apache/spark/sql/delta/RowLevelConcurrencySuite.scala b/spark/src/test/scala/org/apache/spark/sql/delta/RowLevelConcurrencySuite.scala index 6ee50ce2519..47a795da6fb 100644 --- a/spark/src/test/scala/org/apache/spark/sql/delta/RowLevelConcurrencySuite.scala +++ b/spark/src/test/scala/org/apache/spark/sql/delta/RowLevelConcurrencySuite.scala @@ -267,6 +267,90 @@ class RowLevelConcurrencySuite extends QueryTest } } + // --------------------------------------------------------------------------- + // MERGE (matched clauses only): a matched-DELETE writes an in-place deletion vector, exactly like + // a standalone DELETE, so disjoint matched-deletes reconcile through the same DV union. A + // matched-UPDATE additionally writes an *image* file (like a standalone UPDATE), so it + // conservatively conflicts. MERGE inserts / WHEN NOT MATCHED BY SOURCE are out of scope here + // (they need per-file row-tracking classification) and are not exercised. + // --------------------------------------------------------------------------- + + /** `MERGE INTO t USING (single-row source) ... ON t.id = s.id` for the given matched action. */ + private def mergeMatched(dir: File, matchId: Long, action: String): String = + s"""MERGE INTO ${tableRef(dir)} t USING (SELECT id FROM range($matchId, ${matchId + 1})) s + |ON t.id = s.id WHEN MATCHED THEN $action""".stripMargin + + test("disjoint concurrent MERGE matched-deletes on the same file both commit by merging DVs") { + withTempDir { dir => + val log = createSingleFileTableWithDVs(dir) + val txnA = sqlTxn(mergeMatched(dir, 10, "DELETE"), rowLevelConcurrency = true) + val txnB = sqlTxn(mergeMatched(dir, 20, "DELETE"), rowLevelConcurrency = true) + + val (futureA, futureB) = runTxnsWithOrder__A_Start__B__A_End(txnA, txnB) + ThreadUtils.awaitResult(futureA, Duration.Inf) + ThreadUtils.awaitResult(futureB, Duration.Inf) + + assert(ids(dir) === (0L to 99L).filterNot(id => id == 10 || id == 20)) + // A matched-DELETE MERGE writes only a DV (no image file), so the two disjoint deletes merge + // into a single surviving file's deletion vector of cardinality 2 -- identical to DELETE. + assert(deletionVectorCardinalities(log) === Seq(2L)) + } + } + + test("MERGE matched-delete reconciles against a concurrent DELETE on the same file") { + withTempDir { dir => + val log = createSingleFileTableWithDVs(dir) + // The DV union is op-agnostic: a MERGE matched-delete and a plain DELETE on disjoint rows + // reconcile just like two DELETEs. + val txnA = sqlTxn(mergeMatched(dir, 10, "DELETE"), rowLevelConcurrency = true) + val txnB = sqlTxn(s"DELETE FROM ${tableRef(dir)} WHERE id = 20", rowLevelConcurrency = true) + + val (futureA, futureB) = runTxnsWithOrder__A_Start__B__A_End(txnA, txnB) + ThreadUtils.awaitResult(futureA, Duration.Inf) + ThreadUtils.awaitResult(futureB, Duration.Inf) + + assert(ids(dir) === (0L to 99L).filterNot(id => id == 10 || id == 20)) + assert(deletionVectorCardinalities(log) === Seq(2L)) + } + } + + test("overlapping concurrent MERGE matched-deletes still conflict") { + withTempDir { dir => + val log = createSingleFileTableWithDVs(dir) + val txnA = sqlTxn(mergeMatched(dir, 10, "DELETE"), rowLevelConcurrency = true) + val txnB = sqlTxn(mergeMatched(dir, 10, "DELETE"), rowLevelConcurrency = true) + + val (futureA, futureB) = runTxnsWithOrder__A_Start__B__A_End(txnA, txnB) + ThreadUtils.awaitResult(futureB, Duration.Inf) + val e = intercept[SparkException] { ThreadUtils.awaitResult(futureA, Duration.Inf) } + assertConcurrentModificationException(e) + // Clean abort: only the winner's matched-delete (id=10) is applied; its DV has cardinality 1. + assert(ids(dir) === (0L to 99L).filterNot(_ == 10)) + assert(deletionVectorCardinalities(log) === Seq(1L)) + } + } + + test("DELETE (loser) vs MERGE matched-update (winner): winner image file conservatively " + + "conflicts") { + withTempDir { dir => + val log = createSingleFileTableWithDVs(dir) + // A (loser) deletes id=10; B (winner) matched-updates id=20 -> 1020, which masks row 20 with a + // DV and appends an image file for 1020. Like the standalone-UPDATE case above, that image + // file is a non-blind changed-data add the append check cannot prove disjoint, so the loser + // conservatively aborts rather than reconciling. + val txnA = sqlTxn(s"DELETE FROM ${tableRef(dir)} WHERE id = 10", rowLevelConcurrency = true) + val txnB = sqlTxn(mergeMatched(dir, 20, "UPDATE SET id = 1020"), rowLevelConcurrency = true) + + val (futureA, futureB) = runTxnsWithOrder__A_Start__B__A_End(txnA, txnB) + ThreadUtils.awaitResult(futureB, Duration.Inf) + val e = intercept[SparkException] { ThreadUtils.awaitResult(futureA, Duration.Inf) } + assertConcurrentModificationException(e) + // Loser aborted cleanly: only the winner's update is applied (row 20 masked, 1020 appended). + assert(ids(dir) === ((0L to 99L).filterNot(_ == 20) :+ 1020L).sorted) + assert(deletionVectorCardinalities(log) === Seq(1L)) + } + } + // --------------------------------------------------------------------------- // Winner fully removes the file -> not reconcilable -> conflict // --------------------------------------------------------------------------- From ccc0f82d7d48220d0bb7b09dbae8b7757464c4e8 Mon Sep 17 00:00:00 2001 From: Eunjin Song Date: Fri, 7 Aug 2026 21:14:47 -0700 Subject: [PATCH 07/11] Fail safe on DV resolution errors; single-pass action indexing Apply the same review feedback from Case 1 (#7358) proactively to Case 2: - Fail-safe fallback: extract the per-file DV read/overlap/merge/write into reconcileFileDeletionVectors and wrap the per-path call in try/catch (NonFatal). If resolving one shared file throws (unreadable/corrupt DV, transient I/O), skip row-level resolution for it and let the standard file-level checks abort cleanly with a retryable Concurrent* exception, instead of surfacing an unexpected error out of conflict detection. Other shared files are still resolved independently. - Index the current transaction's AddFile/RemoveFile maps in a single pass over actions instead of two `.collect{}.toMap` traversals. No behavior change on the happy/genuine-conflict paths: RowLevelConcurrencySuite 18/18 still pass. Co-Authored-By: Claude Opus 4.8 --- .../delta/RowLevelConcurrencyResolution.scala | 130 +++++++++++------- 1 file changed, 83 insertions(+), 47 deletions(-) diff --git a/spark/src/main/scala/org/apache/spark/sql/delta/RowLevelConcurrencyResolution.scala b/spark/src/main/scala/org/apache/spark/sql/delta/RowLevelConcurrencyResolution.scala index f343e78bc6e..6c64c6387ea 100644 --- a/spark/src/main/scala/org/apache/spark/sql/delta/RowLevelConcurrencyResolution.scala +++ b/spark/src/main/scala/org/apache/spark/sql/delta/RowLevelConcurrencyResolution.scala @@ -19,15 +19,19 @@ package org.apache.spark.sql.delta import java.util.UUID import scala.collection.mutable +import scala.util.control.NonFatal import org.apache.spark.sql.delta.actions.{AddFile, DeletionVectorDescriptor, RemoveFile} import org.apache.spark.sql.delta.commands.DeletionVectorUtils import org.apache.spark.sql.delta.deletionvectors.{RoaringBitmapArray, RoaringBitmapArrayFormat} +import org.apache.spark.sql.delta.logging.DeltaLogKeys import org.apache.spark.sql.delta.metering.DeltaLogging import org.apache.spark.sql.delta.sources.DeltaSQLConf import org.apache.spark.sql.delta.storage.dv.DeletionVectorStore import org.apache.hadoop.fs.Path +import org.apache.spark.internal.MDC + /** * Row-level concurrency resolution for the [[ConflictChecker]]: instead of aborting a concurrent * DV-based DELETE/UPDATE that touches the same physical files as the winning transaction, MERGE the @@ -125,13 +129,15 @@ trait RowLevelConcurrencyResolution extends DeltaLogging { self: ConflictChecker .toMap if (winningDvUpdates.isEmpty) return - // Current transaction's DV updates, indexed by path. - val currentAddByPath = currentTransactionInfo.actions.collect { - case a: AddFile if a.deletionVector != null => a.path -> a - }.toMap - val currentRemoveByPath = currentTransactionInfo.actions.collect { - case r: RemoveFile => r.path -> r - }.toMap + // Current transaction's DV updates and file removes, indexed by path. Built in a single pass + // over the actions (last-writer-wins per path, matching the prior `.collect{}.toMap`). + val currentAddByPath = mutable.Map.empty[String, AddFile] + val currentRemoveByPath = mutable.Map.empty[String, RemoveFile] + currentTransactionInfo.actions.foreach { + case a: AddFile if a.deletionVector != null => currentAddByPath(a.path) = a + case r: RemoveFile => currentRemoveByPath(r.path) = r + case _ => + } val sharedPaths = winningDvUpdates.keySet .intersect(currentAddByPath.keySet) @@ -145,47 +151,24 @@ trait RowLevelConcurrencyResolution extends DeltaLogging { self: ConflictChecker // path -> (rebased AddFile, rebased RemoveFile) val replacements = mutable.Map.empty[String, (AddFile, RemoveFile)] for (path <- sharedPaths) { - val winningAdd = winningDvUpdates(path) - val currentAdd = currentAddByPath(path) - val currentRemove = currentRemoveByPath(path) - - def bitmapOf(dv: DeletionVectorDescriptor): RoaringBitmapArray = - readDeletionVectorOrEmpty(dvStore, dv, tablePath) - val baseBitmap = bitmapOf(currentRemove.deletionVector) - val winningBitmap = bitmapOf(winningAdd.deletionVector) - val currentBitmap = bitmapOf(currentAdd.deletionVector) - - // `baseBitmap` is the DV of `P` at the current txn's read time (carried on its RemoveFile); - // for a 2-way conflict it equals the winner's pre-image too. Both `dv_win` and `dv_cur` are - // supersets of it (a DV only grows), so the newly-deleted rows are `dv \ base` on each side - // and their overlap is `(dv_win INTERSECT dv_cur) MINUS base`. If empty, the two txns - // touched disjoint rows and the schedule `current ; winner` is a valid serialization under - // both WriteSerializable and Serializable (the winner's rewrites/deletes are of rows the - // current txn did not touch), so merging is safe. If non-empty, the same row was touched by - // both -> genuine conflict, left for the standard checks. (For 3+ way chains `base` becomes - // previous winner's DV rather than the original pre-image; the merge stays correct and the - // overlap test stays conservative.) - val newlyDeletedOverlap = winningBitmap.copy() - newlyDeletedOverlap.and(currentBitmap) - newlyDeletedOverlap.andNot(baseBitmap) - - if (newlyDeletedOverlap.isEmpty) { - // Disjoint: merge the deletion vectors and rebase onto the winner's post-image. - val mergedBitmap = winningBitmap.copy() - mergedBitmap.merge(currentBitmap) - val mergedDescriptor = writeMergedDeletionVector(dvStore, tablePath, mergedBitmap) - // Keep the current AddFile's identity (base row ID / default row commit version already - // reconciled by the row-ID phases) but point it at the merged DV. - val rebasedAdd = currentAdd - .copy(deletionVector = mergedDescriptor, dataChange = true) - .withoutTightBoundStats - // Tombstone the winner's now-live AddFile (carries the winning DV) instead of the stale - // pre-image. - val rebasedRemove = winningAdd.removeWithTimestamp() - replacements(path) = (rebasedAdd, rebasedRemove) - rowLevelResolvedPaths += path + try { + reconcileFileDeletionVectors( + dvStore, tablePath, + winningDvUpdates(path), currentAddByPath(path), currentRemoveByPath(path)) + .foreach { rebased => + replacements(path) = rebased + rowLevelResolvedPaths += path + } + } catch { + case NonFatal(e) => + // Fail safe: DV decode/merge/write is a pure optimization over the conservative + // default. If it fails for this file (unreadable/corrupt DV, transient I/O), skip + // row-level resolution for it so the standard file-level checks abort cleanly with a + // retryable Concurrent* exception instead of surfacing an unexpected error out of + // conflict detection. Other shared files are still resolved independently. + logWarning(log"Row-level concurrency resolution failed for file " + + log"${MDC(DeltaLogKeys.PATH, path)}; leaving it for the standard conflict checks", e) } - // else: overlapping row-level modification -> genuine conflict, leave for standard checks. } if (replacements.nonEmpty) { @@ -211,6 +194,59 @@ trait RowLevelConcurrencyResolution extends DeltaLogging { self: ConflictChecker } } + /** + * Reconciles the winning and current transactions' deletion vectors for a single shared file `P`. + * Returns the rebased `(AddFile, RemoveFile)` to substitute into the current transaction when the + * two transactions deleted disjoint rows, or `None` when they overlap (a genuine row-level + * conflict, left for the standard checks). All deletion-vector I/O for `P` happens here, so its + * caller can wrap it in a fail-safe boundary. + */ + private def reconcileFileDeletionVectors( + dvStore: DeletionVectorStore, + tablePath: Path, + winningAdd: AddFile, + currentAdd: AddFile, + currentRemove: RemoveFile): Option[(AddFile, RemoveFile)] = { + def bitmapOf(dv: DeletionVectorDescriptor): RoaringBitmapArray = + readDeletionVectorOrEmpty(dvStore, dv, tablePath) + val baseBitmap = bitmapOf(currentRemove.deletionVector) + val winningBitmap = bitmapOf(winningAdd.deletionVector) + val currentBitmap = bitmapOf(currentAdd.deletionVector) + + // `baseBitmap` is the DV of `P` at the current txn's read time (carried on its RemoveFile); + // for a 2-way conflict it equals the winner's pre-image too. Both `dv_win` and `dv_cur` are + // supersets of it (a DV only grows), so the newly-deleted rows are `dv \ base` on each side + // and their overlap is `(dv_win INTERSECT dv_cur) MINUS base`. If empty, the two txns + // touched disjoint rows and the schedule `current ; winner` is a valid serialization under + // both WriteSerializable and Serializable (the winner's rewrites/deletes are of rows the + // current txn did not touch), so merging is safe. If non-empty, the same row was touched by + // both -> genuine conflict, left for the standard checks. (For 3+ way chains `base` becomes + // previous winner's DV rather than the original pre-image; the merge stays correct and the + // overlap test stays conservative.) + val newlyDeletedOverlap = winningBitmap.copy() + newlyDeletedOverlap.and(currentBitmap) + newlyDeletedOverlap.andNot(baseBitmap) + + if (!newlyDeletedOverlap.isEmpty) { + // Overlapping row-level modification -> genuine conflict, leave for the standard checks. + None + } else { + // Disjoint: merge the deletion vectors and rebase onto the winner's post-image. + val mergedBitmap = winningBitmap.copy() + mergedBitmap.merge(currentBitmap) + val mergedDescriptor = writeMergedDeletionVector(dvStore, tablePath, mergedBitmap) + // Keep the current AddFile's identity (base row ID / default row commit version already + // reconciled by the row-ID phases) but point it at the merged DV. + val rebasedAdd = currentAdd + .copy(deletionVector = mergedDescriptor, dataChange = true) + .withoutTightBoundStats + // Tombstone the winner's now-live AddFile (carries the winning DV) instead of the stale + // pre-image. + val rebasedRemove = winningAdd.removeWithTimestamp() + Some((rebasedAdd, rebasedRemove)) + } + } + /** Reads a deletion vector into a [[RoaringBitmapArray]], returning an empty bitmap for none. */ protected def readDeletionVectorOrEmpty( dvStore: DeletionVectorStore, From 64caa4c6b8a8b1a385fd4c3519e97a3cdb6eb05a Mon Sep 17 00:00:00 2001 From: Eunjin Song Date: Fri, 7 Aug 2026 22:38:34 -0700 Subject: [PATCH 08/11] [Spark] Parallelize row-level DV reconciliation across shared files resolveRowLevelConflicts reconciled each shared file's deletion vectors sequentially on the caller thread. That per-file work is driver-side object-store DV I/O (read + merge + write), so when a winning transaction conflicts on many files the reconciliation window grows with the file count. Reconcile each shared file independently on a bounded pool (cf. DeltaFileOperations footer reads, which use 8); a single conflicting file stays on the caller thread, so the common case is unchanged. Per-file results are collected and then applied on the caller thread, so there is no shared-state race across the pool. The existing per-file NonFatal fail-safe is preserved (moved into reconcileOnePath): a DV decode/merge/write failure for one file skips row-level resolution for it (the standard checks then abort cleanly) without affecting the others. No behavior change on the happy/genuine-conflict paths. Co-Authored-By: Claude Opus 4.8 --- .../delta/RowLevelConcurrencyResolution.scala | 28 ++++++++++++++----- 1 file changed, 21 insertions(+), 7 deletions(-) diff --git a/spark/src/main/scala/org/apache/spark/sql/delta/RowLevelConcurrencyResolution.scala b/spark/src/main/scala/org/apache/spark/sql/delta/RowLevelConcurrencyResolution.scala index 6c64c6387ea..1c6580ebe1d 100644 --- a/spark/src/main/scala/org/apache/spark/sql/delta/RowLevelConcurrencyResolution.scala +++ b/spark/src/main/scala/org/apache/spark/sql/delta/RowLevelConcurrencyResolution.scala @@ -31,6 +31,7 @@ import org.apache.spark.sql.delta.storage.dv.DeletionVectorStore import org.apache.hadoop.fs.Path import org.apache.spark.internal.MDC +import org.apache.spark.util.ThreadUtils /** * Row-level concurrency resolution for the [[ConflictChecker]]: instead of aborting a concurrent @@ -148,17 +149,15 @@ trait RowLevelConcurrencyResolution extends DeltaLogging { self: ConflictChecker val dvStore = DeletionVectorStore.createInstance(deltaLog.newDeltaHadoopConf()) val tablePath = deltaLog.dataPath - // path -> (rebased AddFile, rebased RemoveFile) - val replacements = mutable.Map.empty[String, (AddFile, RemoveFile)] - for (path <- sharedPaths) { + // Reconcile each shared file's DVs independently; the work is driver-side object-store DV + // I/O (read + merge + write). When many files conflict, parallelize across a bounded pool + // to shorten the conflict window; a single file stays on the caller thread. + def reconcileOnePath(path: String): Option[(String, (AddFile, RemoveFile))] = try { reconcileFileDeletionVectors( dvStore, tablePath, winningDvUpdates(path), currentAddByPath(path), currentRemoveByPath(path)) - .foreach { rebased => - replacements(path) = rebased - rowLevelResolvedPaths += path - } + .map(path -> _) } catch { case NonFatal(e) => // Fail safe: DV decode/merge/write is a pure optimization over the conservative @@ -168,7 +167,22 @@ trait RowLevelConcurrencyResolution extends DeltaLogging { self: ConflictChecker // conflict detection. Other shared files are still resolved independently. logWarning(log"Row-level concurrency resolution failed for file " + log"${MDC(DeltaLogKeys.PATH, path)}; leaving it for the standard conflict checks", e) + None } + + // Bounded driver parallelism (cf. DeltaFileOperations footer reads, which use 8). + val pathList = sharedPaths.toSeq + val parallelism = math.min(pathList.size, 8) + val reconciled = + if (parallelism <= 1) pathList.flatMap(reconcileOnePath) + else ThreadUtils.parmap(pathList, "rowLevelConflictResolution", parallelism)( + reconcileOnePath).flatten + + // Apply per-file results on the caller thread (no shared-state races across the pool). + val replacements = mutable.Map.empty[String, (AddFile, RemoveFile)] + for ((path, rebased) <- reconciled) { + replacements(path) = rebased + rowLevelResolvedPaths += path } if (replacements.nonEmpty) { From 8af17fd5756b55644b77653239e6d88294e9be6c Mon Sep 17 00:00:00 2001 From: Eunjin Song Date: Sat, 8 Aug 2026 07:25:55 -0700 Subject: [PATCH 09/11] [Spark] Test: multi-file row-level DV reconcile exercises the parallel path Every existing RowLevelConcurrencySuite conflict is single-file (sharedPaths.size == 1), which takes the caller-thread branch of resolveRowLevelConflicts. Add a two-file test: a DELETE whose predicate spans two data files DV-updates both in one commit, so the loser reconciles two shared paths at once (parallelism == 2), exercising the ThreadUtils.parmap branch. parmap preserves input order, so the result matches the sequential branch. Co-Authored-By: Claude Opus 4.8 --- .../sql/delta/RowLevelConcurrencySuite.scala | 36 +++++++++++++++++++ 1 file changed, 36 insertions(+) diff --git a/spark/src/test/scala/org/apache/spark/sql/delta/RowLevelConcurrencySuite.scala b/spark/src/test/scala/org/apache/spark/sql/delta/RowLevelConcurrencySuite.scala index 47a795da6fb..9d7d7c55ba1 100644 --- a/spark/src/test/scala/org/apache/spark/sql/delta/RowLevelConcurrencySuite.scala +++ b/spark/src/test/scala/org/apache/spark/sql/delta/RowLevelConcurrencySuite.scala @@ -422,6 +422,42 @@ class RowLevelConcurrencySuite extends QueryTest } } + // --------------------------------------------------------------------------- + // Two shared files in one conflict -> bounded-parallel reconcile (ThreadUtils.parmap). + // Every test above conflicts on one file (sharedPaths.size == 1 -> caller-thread branch). + // A DELETE whose predicate spans two files DV-updates both in ONE commit, so the loser + // reconciles two shared paths at once, crossing the `parallelism <= 1` boundary into parmap. + // parmap preserves input order, so the reconciled result matches the sequential branch. + // --------------------------------------------------------------------------- + + test("disjoint DELETEs spanning two files reconcile via the bounded-parallel path") { + withTempDir { dir => + // Two data files: ids [0,100) and [100,200), one file each (separate appends). + spark.range(start = 0, end = 100, step = 1, numPartitions = 1) + .write.format("delta").mode("append").save(dir.getAbsolutePath) + spark.range(start = 100, end = 200, step = 1, numPartitions = 1) + .write.format("delta").mode("append").save(dir.getAbsolutePath) + val log = DeltaLog.forTable(spark, dir.getCanonicalPath) + assert(log.update().allFiles.collect().length === 2, "test table must have two data files") + + // Each DELETE's IN list spans both files (one id < 100, one id >= 100), so each commit + // removes and re-adds BOTH files with a per-file DV. The loser therefore has two shared + // paths in one resolveRowLevelConflicts call -> parallelism == 2 (the parmap branch). + val txnA = + sqlTxn(s"DELETE FROM ${tableRef(dir)} WHERE id IN (20, 120)", rowLevelConcurrency = true) + val txnB = + sqlTxn(s"DELETE FROM ${tableRef(dir)} WHERE id IN (10, 110)", rowLevelConcurrency = true) + + val (futureA, futureB) = runTxnsWithOrder__A_Start__B__A_End(txnA, txnB) + ThreadUtils.awaitResult(futureA, Duration.Inf) + ThreadUtils.awaitResult(futureB, Duration.Inf) + + assert(ids(dir) === (0L to 199L).filterNot(id => Set(10L, 20L, 110L, 120L).contains(id))) + // Both files survive, each carrying its own merged DV (winner + loser delete): cardinality 2. + assert(deletionVectorCardinalities(log).sorted === Seq(2L, 2L)) + } + } + // --------------------------------------------------------------------------- // Partitioned table (DV merge is partition-agnostic) // --------------------------------------------------------------------------- From e9370cffffeb7ea5f3c5fbffa6b41aef26439b5c Mon Sep 17 00:00:00 2001 From: Eunjin Song Date: Wed, 12 Aug 2026 23:23:38 -0700 Subject: [PATCH 10/11] [Spark] Document N-way DV-merge correctness; add base-DV chain and Serializable read-op tests Expand reconcileFileDeletionVectors' comment into an explicit induction on the winner chain: after merging winner k the current txn's base advances to winner k's DV, so each step's overlap test reduces to the two ops' own new deletes -- precise for any number of winners and independent of winner order. Add two RowLevelConcurrencySuite tests: - three disjoint DELETEs reconciling over a non-empty base DV (the deepest chain the ordering helpers reach; combines the base-DV and N-way subtleties). - a Serializable MERGE matched-delete loser (a read-then-modify op) vs a concurrent disjoint DELETE, confirming reconciled paths drop from readFiles under the stricter isolation. Co-Authored-By: Claude Opus 4.8 --- .../delta/RowLevelConcurrencyResolution.scala | 29 +++++++---- .../sql/delta/RowLevelConcurrencySuite.scala | 52 +++++++++++++++++++ 2 files changed, 71 insertions(+), 10 deletions(-) diff --git a/spark/src/main/scala/org/apache/spark/sql/delta/RowLevelConcurrencyResolution.scala b/spark/src/main/scala/org/apache/spark/sql/delta/RowLevelConcurrencyResolution.scala index 1c6580ebe1d..a9e5a82e8c2 100644 --- a/spark/src/main/scala/org/apache/spark/sql/delta/RowLevelConcurrencyResolution.scala +++ b/spark/src/main/scala/org/apache/spark/sql/delta/RowLevelConcurrencyResolution.scala @@ -227,16 +227,25 @@ trait RowLevelConcurrencyResolution extends DeltaLogging { self: ConflictChecker val winningBitmap = bitmapOf(winningAdd.deletionVector) val currentBitmap = bitmapOf(currentAdd.deletionVector) - // `baseBitmap` is the DV of `P` at the current txn's read time (carried on its RemoveFile); - // for a 2-way conflict it equals the winner's pre-image too. Both `dv_win` and `dv_cur` are - // supersets of it (a DV only grows), so the newly-deleted rows are `dv \ base` on each side - // and their overlap is `(dv_win INTERSECT dv_cur) MINUS base`. If empty, the two txns - // touched disjoint rows and the schedule `current ; winner` is a valid serialization under - // both WriteSerializable and Serializable (the winner's rewrites/deletes are of rows the - // current txn did not touch), so merging is safe. If non-empty, the same row was touched by - // both -> genuine conflict, left for the standard checks. (For 3+ way chains `base` becomes - // previous winner's DV rather than the original pre-image; the merge stays correct and the - // overlap test stays conservative.) + // `baseBitmap` is the DV of `P` at the current txn's read time (carried on its RemoveFile). + // Both `dv_win` and `dv_cur` are supersets of it (a DV only grows), so each side's *newly* + // deleted rows are `dv \ base`, and their overlap is `(dv_win INTERSECT dv_cur) MINUS base`. + // Empty overlap => the two txns deleted disjoint rows, and `current ; winner` is a valid + // serialization under both WriteSerializable and Serializable (each deleted only rows the + // other did not touch), so the DV union is safe. Non-empty => the same row was deleted + // concurrently: a genuine conflict, left for the standard checks to abort. + // + // N-way: the current txn reconciles against a chain of winners, once per winning commit. + // Induction on the number of prior winners `k` already merged into the current txn: + // k = 0 (first winner): `base` is P's read-time DV, so `dv_cur \ base` and `dv_win \ base` + // are exactly the two txns' own new deletes -- the test is precise. + // k -> k+1: merging winner k rebased the current RemoveFile onto winner k's AddFile (so now + // `base = dv_win_k`) and the current AddFile onto the merged DV. Winner k+1 committed on + // top of winner k, so `dv_win_k` is a subset of `dv_win_{k+1}`. The test + // `(dv_win_{k+1} INTERSECT dv_cur) MINUS dv_win_k` then reduces to winner k+1's *own* new + // deletes intersected with the current txn's own new deletes -- prior winners cancel out. + // Every step compares only the two operations' genuinely new rows, and the merged DV is a + // union of disjoint contributions, so the result is independent of winner order. val newlyDeletedOverlap = winningBitmap.copy() newlyDeletedOverlap.and(currentBitmap) newlyDeletedOverlap.andNot(baseBitmap) diff --git a/spark/src/test/scala/org/apache/spark/sql/delta/RowLevelConcurrencySuite.scala b/spark/src/test/scala/org/apache/spark/sql/delta/RowLevelConcurrencySuite.scala index 9d7d7c55ba1..edc43929b81 100644 --- a/spark/src/test/scala/org/apache/spark/sql/delta/RowLevelConcurrencySuite.scala +++ b/spark/src/test/scala/org/apache/spark/sql/delta/RowLevelConcurrencySuite.scala @@ -167,6 +167,29 @@ class RowLevelConcurrencySuite extends QueryTest } } + test("Serializable: a MERGE matched-delete (read-then-modify) reconciles vs a concurrent DELETE") { + withTempDir { dir => + val log = createSingleFileTableWithDVs(dir) + sql(s"ALTER TABLE ${tableRef(dir)} SET TBLPROPERTIES " + + s"('${DeltaConfigs.ISOLATION_LEVEL.key}' = 'Serializable')") + // The disjoint-DELETE Serializable test above already covers a delete-only loser. Here the + // loser is a MERGE, which *scans* the file to evaluate its ON condition, so the shared path + // enters the current txn's readFiles as a genuine read (not merely as a delete target). The + // winner's disjoint DELETE removes that same file, which under Serializable would trip the + // delete-read check -- unless row-level resolution drops the reconciled path from readFiles. + // This confirms that removal holds for a read-then-modify op under the stricter isolation. + val txnA = sqlTxn(mergeMatched(dir, 10, "DELETE"), rowLevelConcurrency = true) + val txnB = sqlTxn(s"DELETE FROM ${tableRef(dir)} WHERE id = 20", rowLevelConcurrency = true) + + val (futureA, futureB) = runTxnsWithOrder__A_Start__B__A_End(txnA, txnB) + ThreadUtils.awaitResult(futureA, Duration.Inf) + ThreadUtils.awaitResult(futureB, Duration.Inf) + + assert(ids(dir) === (0L to 99L).filterNot(id => id == 10 || id == 20)) + assert(deletionVectorCardinalities(log) === Seq(2L)) + } + } + test("overlapping concurrent DELETEs still conflict") { withTempDir { dir => val log = createSingleFileTableWithDVs(dir) @@ -422,6 +445,35 @@ class RowLevelConcurrencySuite extends QueryTest } } + test("three concurrent disjoint DELETEs reconcile on a file that already carries a base DV") { + withTempDir { dir => + val log = createSingleFileTableWithDVs(dir) + // Establish a non-empty base DV before any concurrency: delete id=5 (file DV cardinality 1). + // All three concurrent txns read this DV as their common base, and the last one (A) then + // reconciles against a two-deep winner chain (B, then C) stacked on that base. This is the + // deepest chain the ordering helpers reach, and it combines the base-DV subtlety with the + // N-way induction: each step must subtract the *prior winner's* DV, not the read-time base, + // or the accumulated deletes would look like a false overlap. + sql(s"DELETE FROM ${tableRef(dir)} WHERE id = 5") + assert(deletionVectorCardinalities(log) === Seq(1L)) + + val txnA = sqlTxn(s"DELETE FROM ${tableRef(dir)} WHERE id = 10", rowLevelConcurrency = true) + val txnB = sqlTxn(s"DELETE FROM ${tableRef(dir)} WHERE id = 20", rowLevelConcurrency = true) + val txnC = sqlTxn(s"DELETE FROM ${tableRef(dir)} WHERE id = 30", rowLevelConcurrency = true) + + // A starts; B commits; C commits (reading B's state); A commits last (reconciles vs B, C). + val (futureA, futureB, futureC) = + runTxnsWithOrder__A_Start__B__C__A_End(txnA, txnB, txnC) + ThreadUtils.awaitResult(futureA, Duration.Inf) + ThreadUtils.awaitResult(futureB, Duration.Inf) + ThreadUtils.awaitResult(futureC, Duration.Inf) + + assert(ids(dir) === (0L to 99L).filterNot(id => Set(5L, 10L, 20L, 30L).contains(id))) + // Base row 5 plus three disjoint concurrent deletes -> merged deletion vector cardinality 4. + assert(deletionVectorCardinalities(log) === Seq(4L)) + } + } + // --------------------------------------------------------------------------- // Two shared files in one conflict -> bounded-parallel reconcile (ThreadUtils.parmap). // Every test above conflicts on one file (sharedPaths.size == 1 -> caller-thread branch). From 0b8f4923428612bbd5b7467333fbaea67265dd4f Mon Sep 17 00:00:00 2001 From: Eunjin Song Date: Wed, 12 Aug 2026 23:38:09 -0700 Subject: [PATCH 11/11] [Spark] Simplify RLC integration: prune the winning summary instead of guarding each check Row-level resolution previously threaded a `!rowLevelResolvedPaths.contains(...)` guard into four separate file-level checks (append, delete-read, whole-table, delete-delete) in ConflictChecker. Instead, rewrite both sides of the conflict once in resolveRowLevelConflicts: rebase the current transaction (as before) and prune the reconciled AddFile(P)/RemoveFile(P) pair from a single effective `winningCommitSummary` (now a `var`, mirroring `currentTransactionInfo`). The four checks revert to byte-identical-to-upstream, since they now see a winner that never touched the reconciled files. Removes the trait-level mutable `rowLevelResolvedPaths` set and `canSkipAddedFileForRowLevelConcurrency`; the "don't skip UPDATE image files" reasoning moves to pruneReconciledFiles (only the same-path pair is pruned). ConflictChecker footprint drops to a mixin + one `var` + one call site; all subtle DV logic stays confined to reconcileFileDeletionVectors. RowLevelConcurrencySuite 21/21, FeatureEnablementConcurrencySuite 40/40. Co-Authored-By: Claude Opus 4.8 --- .../spark/sql/delta/ConflictChecker.scala | 25 +++--- .../delta/RowLevelConcurrencyResolution.scala | 77 +++++++++---------- 2 files changed, 47 insertions(+), 55 deletions(-) diff --git a/spark/src/main/scala/org/apache/spark/sql/delta/ConflictChecker.scala b/spark/src/main/scala/org/apache/spark/sql/delta/ConflictChecker.scala index 9933b68715e..623b14b4866 100644 --- a/spark/src/main/scala/org/apache/spark/sql/delta/ConflictChecker.scala +++ b/spark/src/main/scala/org/apache/spark/sql/delta/ConflictChecker.scala @@ -218,7 +218,11 @@ object WinningCommitSummary { private[delta] class ConflictChecker( protected val spark: SparkSession, initialCurrentTransactionInfo: CurrentTransactionInfo, - protected val winningCommitSummary: WinningCommitSummary, + // A `var` (reassigned once by [[resolveRowLevelConflicts]]) to drop files that were reconciled + // at the row level, mirroring how `currentTransactionInfo` is rebased. After that the + // file-level checks run against a summary that never mentions the reconciled files, so they + // need no changes. + protected var winningCommitSummary: WinningCommitSummary, isolationLevel: IsolationLevel) extends DeltaLogging with ConflictCheckerPredicateElimination with RowLevelConcurrencyResolution { @@ -1144,15 +1148,8 @@ private[delta] class ConflictChecker( Seq.empty } - // Only re-scan the added files when row-level resolution actually reconciled something: a - // file is skippable only if its path is in `rowLevelResolvedPaths`, so an empty set means the - // filter is a no-op. Skips the traversal (and its allocation) on the common no-conflict path. - val addedFilesAfterRowLevelResolution = - if (rowLevelResolvedPaths.isEmpty) addedFilesToCheckForConflicts - else addedFilesToCheckForConflicts.filterNot(canSkipAddedFileForRowLevelConcurrency) - val fileMatchingPartitionReadPredicates = - getFirstFileMatchingPartitionPredicates(addedFilesAfterRowLevelResolution) + getFirstFileMatchingPartitionPredicates(addedFilesToCheckForConflicts) if (fileMatchingPartitionReadPredicates.nonEmpty) { throw DeltaErrors.concurrentAppendException( @@ -1174,7 +1171,7 @@ private[delta] class ConflictChecker( val readFilePaths = currentTransactionInfo.readFiles.map( f => f.path -> f.partitionValues).toMap val deleteReadOverlap = winningCommitSummary.removedFiles - .find(r => readFilePaths.contains(r.path) && !rowLevelResolvedPaths.contains(r.path)) + .find(r => readFilePaths.contains(r.path)) if (deleteReadOverlap.nonEmpty) { val partitionOpt = getPrettyPartitionMessage(readFilePaths(deleteReadOverlap.get.path)) throw DeltaErrors.concurrentDeleteReadException( @@ -1183,11 +1180,7 @@ private[delta] class ConflictChecker( winningCommitVersion, partitionOpt) } - // Row-level concurrency: a removed file that was reconciled at the row level must not - // re-trigger the whole-table conflict either. - val unresolvedRemovedFiles = - winningCommitSummary.removedFiles.exists(r => !rowLevelResolvedPaths.contains(r.path)) - if (unresolvedRemovedFiles && currentTransactionInfo.readWholeTable) { + if (winningCommitSummary.removedFiles.nonEmpty && currentTransactionInfo.readWholeTable) { throw DeltaErrors.concurrentDeleteReadException( winningCommitSummary.commitInfo, getTableNameOrPath, @@ -1208,7 +1201,7 @@ private[delta] class ConflictChecker( .collect { case r: RemoveFile => r.path -> r.partitionValues } .toMap val deleteOverlap = winningCommitSummary.removedFiles - .find(r => deletedFilePaths.contains(r.path) && !rowLevelResolvedPaths.contains(r.path)) + .find(r => deletedFilePaths.contains(r.path)) if (deleteOverlap.nonEmpty) { val partitionOpt = getPrettyPartitionMessage(deletedFilePaths(deleteOverlap.get.path)) throw DeltaErrors.concurrentDeleteDeleteException( diff --git a/spark/src/main/scala/org/apache/spark/sql/delta/RowLevelConcurrencyResolution.scala b/spark/src/main/scala/org/apache/spark/sql/delta/RowLevelConcurrencyResolution.scala index a9e5a82e8c2..84f40e32911 100644 --- a/spark/src/main/scala/org/apache/spark/sql/delta/RowLevelConcurrencyResolution.scala +++ b/spark/src/main/scala/org/apache/spark/sql/delta/RowLevelConcurrencyResolution.scala @@ -46,13 +46,6 @@ import org.apache.spark.util.ThreadUtils */ trait RowLevelConcurrencyResolution extends DeltaLogging { self: ConflictChecker => - /** - * Paths of files whose "same physical file" conflict with the winning transaction was resolved at - * the row level by [[resolveRowLevelConflicts]] (deletion vectors merged). The file-level delete - * and append checks skip these paths, since they have already been reconciled. - */ - protected val rowLevelResolvedPaths = mutable.Set.empty[String] - /** Whether row-level concurrency resolution is enabled and applicable to this table. */ protected lazy val rowLevelConcurrencyEnabled: Boolean = spark.conf.get(DeltaSQLConf.DELTA_ROW_LEVEL_CONCURRENCY_ENABLED) && @@ -63,28 +56,6 @@ trait RowLevelConcurrencyResolution extends DeltaLogging { self: ConflictChecker protected lazy val winningOperationName: Option[String] = winningCommitSummary.commitInfo.map(_.operation) - /** - * Whether a file added by the winning transaction can be skipped in the added-files (append) - * conflict check thanks to row-level concurrency resolution. - * - * This is true only when the file's "same physical file" conflict was already reconciled by - * merging deletion vectors ([[resolveRowLevelConflicts]] recorded the path in - * [[rowLevelResolvedPaths]]). In that case the winner's re-added `AddFile(P)` carries the winning - * DV that we already folded into the current transaction's merged DV, so re-checking it would be - * a false conflict. - * - * We deliberately do NOT skip a rewrite-only DML winner's *new image* files here (an UPDATE - * writes updated row values to a fresh path). Those are ordinary non-blind changed-data files - * and can legitimately conflict: e.g. an UPDATE can move a row *into* the loser's predicate - * (winner `SET x = 15`, loser `DELETE WHERE x > 10`, row was `x = 5`), a genuine write-skew that - * the DV union cannot detect. They are arbitrated by the standard added-files check (and, when - * enabled, by conflict-time data skipping over their stats). - */ - protected def canSkipAddedFileForRowLevelConcurrency(addFile: AddFile): Boolean = { - if (!rowLevelConcurrencyEnabled) return false - rowLevelResolvedPaths.contains(addFile.path) - } - /** * Resolves "same physical file" conflicts with the winning transaction at the row level. * @@ -179,35 +150,63 @@ trait RowLevelConcurrencyResolution extends DeltaLogging { self: ConflictChecker reconcileOnePath).flatten // Apply per-file results on the caller thread (no shared-state races across the pool). - val replacements = mutable.Map.empty[String, (AddFile, RemoveFile)] - for ((path, rebased) <- reconciled) { - replacements(path) = rebased - rowLevelResolvedPaths += path - } - + val replacements = reconciled.toMap if (replacements.nonEmpty) { + val resolvedPaths = replacements.keySet + + // Rewrite BOTH sides of the conflict so the residual is a plain no-conflict state that the + // file-level checks (unchanged from upstream) handle. Current txn: rebase each AddFile(P) + // onto the merged DV and each RemoveFile(P) onto the winner's post-image, and drop P from + // readFiles (it is no longer "read" for the delete-read check). val newActions = currentTransactionInfo.actions.map { case a: AddFile if replacements.contains(a.path) => replacements(a.path)._1 case r: RemoveFile if replacements.contains(r.path) => replacements(r.path)._2 case other => other } - // Resolved files are no longer "read" for the purposes of the delete-read check. - val newReadFiles = currentTransactionInfo.readFiles - .filterNot(f => rowLevelResolvedPaths.contains(f.path)) + val newReadFiles = + currentTransactionInfo.readFiles.filterNot(f => resolvedPaths.contains(f.path)) currentTransactionInfo = currentTransactionInfo.copy(actions = newActions, readFiles = newReadFiles) + // Winning side: drop the reconciled AddFile(P)/RemoveFile(P) pair from the summary, so no + // check sees P as a winner-side add or remove. + winningCommitSummary = pruneReconciledFiles(winningCommitSummary, resolvedPaths) recordDeltaEvent( deltaLog, opType = "delta.rowLevelConcurrency.deletionVectorsMerged", data = Map( "winningCommitVersion" -> winningCommitVersion, - "resolvedPaths" -> rowLevelResolvedPaths.size, + "resolvedPaths" -> resolvedPaths.size, "winningOperation" -> winningOperationName.getOrElse("UNKNOWN"))) } } } + /** + * Returns a copy of `summary` with the reconciled files removed: for each resolved path we drop + * the winner's `AddFile(P)` (its deletion vector was folded into the current transaction's merged + * DV) and its `RemoveFile(P)` (tombstone of the shared pre-image). Rebuilding from the filtered + * action list recomputes the derived views (`addedFiles`, `removedFiles`, + * `changedDataAddedFiles`, ...), so the file-level conflict checks see a winner that never + * touched these files. + * + * Only the same-path reconciled pair is pruned. A rewrite-only DML winner's *new image* files (an + * UPDATE writes updated row values to a fresh path) are left in the summary and arbitrated by the + * standard added-files check: an UPDATE can move a row *into* the loser's predicate (winner + * `SET x = 15`, loser `DELETE WHERE x > 10`, row was `x = 5`), a genuine write-skew the DV union + * cannot detect. All non-file actions (protocol, metadata, domain metadata) are preserved. + */ + private def pruneReconciledFiles( + summary: WinningCommitSummary, + resolvedPaths: Set[String]): WinningCommitSummary = { + val prunedActions = summary.actions.filterNot { + case a: AddFile => resolvedPaths.contains(a.path) + case r: RemoveFile => resolvedPaths.contains(r.path) + case _ => false + } + new WinningCommitSummary(prunedActions, summary.fileStatus, summary.readTimeMs) + } + /** * Reconciles the winning and current transactions' deletion vectors for a single shared file `P`. * Returns the rebased `(AddFile, RemoveFile)` to substitute into the current transaction when the