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 72f973b3810..72e3a8b790a 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 @@ -270,11 +270,16 @@ object WinningCommitSummary { } private[delta] class ConflictChecker( - spark: SparkSession, + protected val spark: SparkSession, initialCurrentTransactionInfo: CurrentTransactionInfo, - winningCommitSummary: WinningCommitSummary, + // A `var` (reassigned once by [[prepareRowLevelConflictResolution]]) to drop files that are + // resolved at the row level, mirroring how `currentTransactionInfo` is rebased. After that the + // file-level checks run against a summary that never mentions those files, so they need no + // changes. + protected var winningCommitSummary: WinningCommitSummary, isolationLevel: IsolationLevel) - extends DeltaLogging with ConflictCheckerPredicateElimination { + extends DeltaLogging with ConflictCheckerPredicateElimination + with RowLevelConcurrencyResolution { protected val dataPath = initialCurrentTransactionInfo.readSnapshot.dataPath protected val winningCommitVersion = winningCommitSummary.commitVersion @@ -352,12 +357,24 @@ private[delta] class ConflictChecker( // Update the table version in newly added type widening metadata. updateTypeWideningMetadata() + // Row-level concurrency: set aside the "same physical file" conflicts that merging deletion + // vectors may resolve, so that concurrent DML touching disjoint rows of the same file no + // longer aborts. The file-level checks below then see only the other conflicts. Runs after + // row-ID reassignment so merged files keep stable base row IDs. + prepareRowLevelConflictResolution() + // Data file checks. checkForAddedFilesThatShouldHaveBeenReadByCurrentTxn() checkForDeletedFilesAgainstCurrentTxnReadFiles() checkForDeletedFilesAgainstCurrentTxnDeletedFiles() resolveTimestampOrderingConflicts() + // Resolve the set-aside files last, so a transaction that aborts on another conflict decodes + // no deletion vector, scans no read and writes no merged deletion vector. + resolveRowLevelConflicts() + + recordRowLevelConflictResolution() + // Fold the winning commit into our transaction info before the next attempt: this advances // `preCommitLatestAMTCheckpointOpt` to the winner's inline AMT checkpoint (when the winner // wrote one) and clears `currentCommitAttemptAMTCheckpointOpt`. 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..52a57d7973b --- /dev/null +++ b/spark/src/main/scala/org/apache/spark/sql/delta/RowLevelConcurrencyResolution.scala @@ -0,0 +1,548 @@ +/* + * 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 scala.collection.mutable +import scala.util.control.NonFatal + +import org.apache.spark.sql.delta.ClassicColumnConversions._ +import org.apache.spark.sql.delta.actions.{AddFile, DeletionVectorDescriptor, RemoveFile} +import org.apache.spark.sql.delta.commands.{DeletionVectorUtils, DeletionVectorWriter} +import org.apache.spark.sql.delta.commands.DeletionVectorWriter.DeletionVectorMapperContext +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 +import org.apache.spark.sql.{Column, DataFrame} +import org.apache.spark.sql.catalyst.expressions.{AttributeReference, Expression} +import org.apache.spark.sql.functions.{coalesce, col, lit, sum} +import org.apache.spark.util.ThreadUtils + +/** + * 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. [[readDeletionVectorOrEmpty]] is `protected` and [[PackedDeletionVectorWriter]] is + * shared so the OPTIMIZE-vs-DML reconciliation (which mixes into the same checker) can reuse them. + */ +trait RowLevelConcurrencyResolution extends DeltaLogging { self: ConflictChecker => + + /** 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) + + /** + * Resolves "same physical file" conflicts with the winning transaction at the row level, for the + * files set aside by [[prepareRowLevelConflictResolution]]. Runs once every other check against + * the winning commit passed, and throws the conflict when any of the files cannot be reconciled. + * + * 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 the commit + * aborts; + * 3. on disjoint sets, merge the DVs (`dv_win UNION dv_cur`), persist them packed into new DV + * files, 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 that aborts the + * commit. (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.) + * + * Reconciling `P` hides the winner's deletes in `P` from the delete-read check, so a file is only + * reconciled when the current transaction's reads are provably unaffected by them: none of the + * rows the winner newly deleted may match any of the current transaction's read predicates (see + * [[currentReadsUnaffectedByWinnerDeletes]]). The current transaction then produces the same + * actions whether it runs before or after the winner, so `winner ; current` -- the commit order + * -- is a valid serialization. For a DELETE/UPDATE the only read is its own condition, whose + * matching rows in `P` are exactly the rows it deleted, so the disjointness test above already + * implies this. A MERGE, or any transaction that also read other rows, can see a winner's delete: + * `MERGE ... WHEN MATCHED AND cond THEN DELETE WHEN NOT MATCHED THEN INSERT` that matched a row + * without deleting it must INSERT the source row once the winner deletes that target row. + * + * 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 (rowLevelCandidates.isEmpty) return + recordTime("resolved-row-level-conflicts") { + val dvStore = DeletionVectorStore.createInstance(deltaLog.newDeltaHadoopConf()) + val tablePath = deltaLog.dataPath + val replacements = failSafe { + val disjointFiles = checkDisjointDeletionVectors(dvStore, tablePath) + val disjointPaths = disjointFiles.map(_.file.path).toSet + // The two transactions deleted the same row of a file, or one brought deleted rows back. + rowLevelCandidates.find(f => !disjointPaths.contains(f.path)).foreach { f => + throw conflictOnFile(f, deleteRead = false) + } + if (!currentReadsUnaffectedByWinnerDeletes(rowLevelCandidates)) { + throw conflictOnFile(rowLevelCandidates.head, deleteRead = true) + } + writeMergedDeletionVectors(dvStore, tablePath, disjointFiles) + } + + // Rebase the current transaction onto the winner's post-image: each AddFile(P) now carries + // the merged DV and each RemoveFile(P) tombstones the winner's AddFile(P). The winning + // summary and the current read files no longer mention P since the preparation. + 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 + } + currentTransactionInfo = currentTransactionInfo.copy(actions = newActions) + numRowLevelResolvedPaths = replacements.size + } + } + + /** + * Sets aside the files [[resolveRowLevelConflicts]] may reconcile: those whose deletion vector + * both the winning and the current transaction updated, i.e. a `RemoveFile(P)` plus an + * `AddFile(P)` with a DV on both sides. Only compares actions. + * + * The files are dropped from the winning summary and from the current transaction's read files, + * so the file-level checks that run next see only the other conflicts, and abort before any + * deletion vector is decoded, any read is scanned or any merged deletion vector is written. + */ + protected def prepareRowLevelConflictResolution(): Unit = { + if (!rowLevelConcurrencyEnabled) return + // A whole-table read observes every row, so any row the winner deleted affects it. + if (currentTransactionInfo.readWholeTable) return + // RESTORE is not reconciled: it can bring deleted rows back, which the DV merge does not model. + if (currentTransactionInfo.op.name == DeltaOperations.OP_RESTORE || + winningOperationName.contains(DeltaOperations.OP_RESTORE)) { + 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 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) + .intersect(currentRemoveByPath.keySet) + if (sharedPaths.isEmpty) return + + rowLevelCandidates = sharedPaths.toSeq.map { path => + SharedFile(winningDvUpdates(path), currentAddByPath(path), currentRemoveByPath(path)) + } + currentTransactionInfo = currentTransactionInfo.copy( + readFiles = currentTransactionInfo.readFiles.filterNot(f => sharedPaths.contains(f.path))) + winningCommitSummary = pruneReconciledFiles(winningCommitSummary, sharedPaths) + } + + /** The files set aside by [[prepareRowLevelConflictResolution]]. */ + private var rowLevelCandidates: Seq[SharedFile] = Seq.empty + + /** A file whose deletion vector both the winning and the current transaction updated. */ + private case class SharedFile( + winningAdd: AddFile, + currentAdd: AddFile, + currentRemove: RemoveFile) { + def path: String = currentAdd.path + } + + /** + * Fail safe: an unreadable or corrupt DV, or a transient I/O error, aborts the commit with a + * retryable conflict instead of surfacing an unexpected error out of conflict detection. + */ + private def failSafe[A](f: => A): A = + try f catch { + case e: DeltaConcurrentModificationException => throw e + case NonFatal(e) => + val file = rowLevelCandidates.head + logWarning(log"Row-level concurrency resolution failed for file " + + log"${MDC(DeltaLogKeys.PATH, file.path)}; aborting with a conflict", e) + throw conflictOnFile(file, deleteRead = false) + } + + /** The conflict for a set-aside file that cannot be reconciled. */ + private def conflictOnFile(file: SharedFile, deleteRead: Boolean): Throwable = { + val partitionOpt = getPrettyPartitionMessage(file.currentRemove.partitionValues) + if (deleteRead) { + DeltaErrors.concurrentDeleteReadException( + winningCommitSummary.commitInfo, getTableNameOrPath, winningCommitVersion, partitionOpt) + } else { + DeltaErrors.concurrentDeleteDeleteException( + winningCommitSummary.commitInfo, getTableNameOrPath, winningCommitVersion, partitionOpt) + } + } + + /** + * Applies `f` to each item, across a bounded pool when there are several: the work is + * driver-side object-store DV I/O, so parallelism shortens the conflict window. A single item + * stays on the caller thread. + */ + private def forEachFile[A, B](items: Seq[A], name: String)(f: A => Option[B]): Seq[B] = { + // Bounded driver parallelism (cf. DeltaFileOperations footer reads, which use 8). + val parallelism = math.min(items.size, 8) + if (parallelism <= 1) items.flatMap(f) + else ThreadUtils.parmap(items, name, parallelism)(f).flatten + } + + /** Number of files reconciled by [[resolveRowLevelConflicts]] against this winning commit. */ + private var numRowLevelResolvedPaths = 0 + + /** + * Records the row-level resolution, if any. Called once every check against the winning commit + * has passed, so a reconciliation undone by a later conflict is not reported as a merge. + */ + protected def recordRowLevelConflictResolution(): Unit = { + if (numRowLevelResolvedPaths == 0) return + recordDeltaEvent( + deltaLog, + opType = "delta.rowLevelConcurrency.deletionVectorsMerged", + data = Map( + "winningCommitVersion" -> winningCommitVersion, + "resolvedPaths" -> numRowLevelResolvedPaths, + "winningOperation" -> winningOperationName.getOrElse("UNKNOWN"))) + } + + /** + * Whether none of the rows the winner newly deleted from `files` match any of the current + * transaction's read predicates, i.e. its reads return the same rows before and after the + * winner's deletes. + * + * The winner's newly-deleted rows of a file are its live rows under the current transaction's + * base DV minus its live rows under the winner's DV (a subset, as the winner's DV contains the + * base DV, see [[disjointFileDeletionVectors]]), so for the read + * condition `cond` (OR across reads, AND within a read): + * + * matches(newly deleted) = matches(base-DV live rows) - matches(winner-DV live rows) + * + * Both images of all files are scanned with signed weights (+1 base, -1 winner) and summed in a + * single Spark job. Conservative: a read predicate is evaluated on its deterministic, + * subquery-free conjuncts (dropping a conjunct only matches more rows), a read with none of + * those matches every row, and any failure counts as affected. + * + * Two cases are provably unaffected without a scan: a DELETE/UPDATE whose only read is its own + * condition, and most MERGEs whose only read is the target scan (which usually matches every + * row), see [[mergeUnaffectedByDeletesOfUntouchedRows]]. + */ + private def currentReadsUnaffectedByWinnerDeletes(files: Seq[SharedFile]): Boolean = { + val readPredicates = currentTransactionInfo.readPredicates + currentTransactionInfo.op match { + // A DELETE/UPDATE whose only read is its own condition marked every row of the shared files + // that matches it, so the disjointness test already shows none of the winner's new deletes + // match. No scan needed. + case _: DeltaOperations.Delete | _: DeltaOperations.Update if readPredicates.size == 1 => + return true + // The shortcut reasons about the target scan only, so it applies only when that scan is the + // transaction's only read: a MERGE whose source also reads the table records more reads. + case merge: DeltaOperations.Merge + if readPredicates.size == 1 && mergeUnaffectedByDeletesOfUntouchedRows(merge) => + return true + case _ => + } + if (readPredicates.isEmpty) return false + val conjunctsPerRead = readPredicates.map { rp => + (rp.partitionPredicates ++ rp.dataPredicates).filter { e => + e.deterministic && !DeltaTableUtils.containsSubquery(e) + } + } + if (conjunctsPerRead.exists(_.isEmpty)) return false + try { + val snapshot = currentTransactionInfo.readSnapshot + def matchingRowWeights(images: Seq[AddFile], weight: Long): DataFrame = { + val df = deltaLog.createDataFrame( + snapshot, images, actionTypeOpt = Some("rowLevelConcurrencyReadCheck")) + val condition = conjunctsPerRead + .map(_.map(e => rebindToDataFrame(e, df)).reduce(_ && _)) + .reduce(_ || _) + df.where(condition).select(lit(weight).as("weight")) + } + val baseImages = + files.map(f => f.currentAdd.copy(deletionVector = f.currentRemove.deletionVector)) + val winnerImages = files.map(_.winningAdd) + val newlyDeletedMatches = matchingRowWeights(baseImages, 1L) + .union(matchingRowWeights(winnerImages, -1L)) + .select(coalesce(sum(col("weight")), lit(0L))) + .head() + .getLong(0) + newlyDeletedMatches == 0 + } catch { + case NonFatal(e) => + logWarning(log"Row-level concurrency read check failed; leaving the shared files for " + + log"the standard conflict checks", e) + false + } + } + + /** + * Whether a MERGE's outcome is unchanged when rows it did not modify are deleted first. Each + * target row's fate depends only on that row and the source: a row that matched a MATCHED (or + * NOT MATCHED BY SOURCE) clause was modified, so it is in the MERGE's own new deletes and + * disjointness already excludes it from the winner's. The exception is NOT MATCHED: a source + * row is inserted only if NO target row matched it, so deleting a matched-but-unmodified target + * row changes the inserts. That can only happen when no MATCHED clause is unconditional. + */ + private def mergeUnaffectedByDeletesOfUntouchedRows(merge: DeltaOperations.Merge): Boolean = + merge.notMatchedPredicates.isEmpty || merge.matchedPredicates.exists(_.predicate.isEmpty) + + /** Rebinds `e`'s attributes to `df`'s columns by name. */ + private def rebindToDataFrame(e: Expression, df: DataFrame): Column = + Column(e.transformUp { + case a: AttributeReference => df.col("`" + a.name.replace("`", "``") + "`").expr + }) + + /** + * 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) + } + + /** + * A set-aside file whose newly-deleted rows are disjoint between the winning and current + * transactions, with its decoded `(winning, current)` bitmaps when they are kept for the merge. + */ + private case class DisjointFile( + file: SharedFile, + bitmaps: Option[(RoaringBitmapArray, RoaringBitmapArray)]) + + /** + * Decodes the deletion vectors of every set-aside file and returns those whose newly-deleted + * rows are disjoint. The decoded bitmaps are kept for the merge only while their total size stays + * within [[DeltaSQLConf.DELTA_ROW_LEVEL_CONCURRENCY_MAX_RETAINED_DV_BYTES]], so the driver does + * not hold every file's bitmaps at once; the other files' are read again when merged. + */ + private def checkDisjointDeletionVectors( + dvStore: DeletionVectorStore, + tablePath: Path): Seq[DisjointFile] = { + val maxRetainedBytes = + spark.conf.get(DeltaSQLConf.DELTA_ROW_LEVEL_CONCURRENCY_MAX_RETAINED_DV_BYTES) + var retainedBytes = 0L + val filesToCheck = rowLevelCandidates.map { f => + retainedBytes += + f.winningAdd.deletionVector.sizeInBytes + f.currentAdd.deletionVector.sizeInBytes + f -> (retainedBytes <= maxRetainedBytes) + } + forEachFile(filesToCheck, "rowLevelConflictCheck") { case (f, retain) => + disjointFileDeletionVectors(dvStore, tablePath, f, retain) + } + } + + /** + * Decodes the winning, current and base deletion vectors of a single set-aside file `P` and + * returns it when the two transactions deleted disjoint rows, or `None` when they cannot be + * merged (a genuine row-level conflict). + */ + private def disjointFileDeletionVectors( + dvStore: DeletionVectorStore, + tablePath: Path, + file: SharedFile, + retainBitmaps: Boolean): Option[DisjointFile] = { + def bitmapOf(dv: DeletionVectorDescriptor): RoaringBitmapArray = + readDeletionVectorOrEmpty(dvStore, dv, tablePath) + val baseBitmap = bitmapOf(file.currentRemove.deletionVector) + val winningBitmap = bitmapOf(file.winningAdd.deletionVector) + val currentBitmap = bitmapOf(file.currentAdd.deletionVector) + // `baseBitmap` is the DV of `P` at the current txn's read time (carried on its RemoveFile). + // DELETE/UPDATE/MERGE only add rows to a DV, so `dv_win` and `dv_cur` contain `base` and each + // side's *newly* deleted rows are `dv \ base`. A commit that brings deleted rows back breaks + // this: the union would delete those rows again, and the read check would not see them. Such + // a file is not reconciled. + if (!containsAll(winningBitmap, baseBitmap) || !containsAll(currentBitmap, baseBitmap)) { + return None + } + // The newly-deleted rows overlap in `(dv_win INTERSECT dv_cur) MINUS base`. + // Empty overlap => the two txns deleted disjoint rows, so the DV union is safe as far as the + // writes go (the reads are checked separately, see currentReadsUnaffectedByWinnerDeletes). + // Non-empty => the same row was deleted concurrently: a genuine conflict, and the commit + // aborts. + // + // 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) + + if (newlyDeletedOverlap.isEmpty) { + Some(DisjointFile(file, if (retainBitmaps) Some((winningBitmap, currentBitmap)) else None)) + } else { + None + } + } + + /** Whether `bitmap` contains every row of `subset`. */ + private def containsAll(bitmap: RoaringBitmapArray, subset: RoaringBitmapArray): Boolean = { + val missing = subset.copy() + missing.andNot(bitmap) + missing.isEmpty + } + + /** + * Merges each disjoint file's deletion vectors and writes them packed into as few new DV files as + * [[DeltaSQLConf.DELETION_VECTOR_PACKING_TARGET_SIZE]] allows, like the DML write path does. + * Bitmaps not kept from the disjointness check are read again here, a bounded batch at a time. + * Returns the current transaction's rebased `(AddFile, RemoveFile)` per path. + * + * NOTE: this writes DV files as a side effect of conflict resolution. If the commit ultimately + * fails or is retried against another winning version, they are orphaned and later reclaimed by + * VACUUM (same lifecycle as any DV written by DML). + */ + private def writeMergedDeletionVectors( + dvStore: DeletionVectorStore, + tablePath: Path, + files: Seq[DisjointFile]): Map[String, (AddFile, RemoveFile)] = { + val writer = new PackedDeletionVectorWriter( + dvStore, tablePath, spark.conf.get(DeltaSQLConf.DELETION_VECTOR_PACKING_TARGET_SIZE)) + val replacements = Map.newBuilder[String, (AddFile, RemoveFile)] + try { + // Merge a batch in parallel, then append it to the packed file on this thread. + files.grouped(64).foreach { batch => + val merged = forEachFile(batch, "rowLevelConflictMerge") { f => + val (winningBitmap, currentBitmap) = f.bitmaps.getOrElse(( + readDeletionVectorOrEmpty(dvStore, f.file.winningAdd.deletionVector, tablePath), + readDeletionVectorOrEmpty(dvStore, f.file.currentAdd.deletionVector, tablePath))) + val mergedBitmap = winningBitmap.copy() + mergedBitmap.merge(currentBitmap) + Some(f.file -> mergedBitmap) + } + merged.foreach { case (file, mergedBitmap) => + replacements += file.path -> rebased(file, writer.write(mergedBitmap)) + } + } + } finally { + writer.close() + } + replacements.result() + } + + /** The current transaction's `(AddFile, RemoveFile)` for `file`, rebased onto the winner's. */ + private def rebased( + file: SharedFile, + mergedDeletionVector: DeletionVectorDescriptor): (AddFile, RemoveFile) = { + // 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 = file.currentAdd + .copy(deletionVector = mergedDeletionVector, dataChange = true) + .withoutTightBoundStats + // Tombstone the winner's now-live AddFile (carries the winning DV) instead of the stale + // pre-image. + (rebasedAdd, file.winningAdd.removeWithTimestamp()) + } + + /** 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) + } +} + +/** + * Writes deletion vectors packed into as few new files as `targetSize` allows, as the DML write + * path does (see [[DeletionVectorWriter.createMapperToStoreDeletionVectors]]). + */ +private[delta] class PackedDeletionVectorWriter( + dvStore: DeletionVectorStore, + tablePath: Path, + targetSize: Long) extends AutoCloseable { + private val tablePathWithFs = dvStore.pathWithFileSystem(tablePath) + private var context: Option[DeletionVectorMapperContext] = None + private var bytesInFile = 0L + + /** Writes `bitmap` and returns its descriptor. An empty bitmap needs no file. */ + def write(bitmap: RoaringBitmapArray): DeletionVectorDescriptor = { + if (bitmap.isEmpty) return DeletionVectorDescriptor.EMPTY + val serialized = + DeletionVectorUtils.serialize(bitmap, RoaringBitmapArrayFormat.Portable, Some(tablePath)) + if (context.isEmpty || bytesInFile + serialized.length > targetSize) { + close() + val (writer, fileId) = DeletionVectorWriter.createWriter(dvStore, tablePathWithFs) + context = Some(DeletionVectorMapperContext(dvStore, writer, tablePath, fileId, prefix = "")) + bytesInFile = 0L + } + bytesInFile += serialized.length + DeletionVectorWriter.storeSerializedBitmap(context.get, serialized, bitmap.cardinality) + } + + override def close(): Unit = { + context.foreach(_.writer.close()) + context = None + } +} 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 bb44fde6ad0..e90bf58ad93 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 @@ -607,6 +607,31 @@ 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_ROW_LEVEL_CONCURRENCY_MAX_RETAINED_DV_BYTES = + buildConf("rowLevelConcurrency.maxRetainedDeletionVectorBytes") + .internal() + .doc( + """Maximum total size of the deletion vectors that row-level concurrency keeps in driver + |memory between checking and merging the files it reconciles. The deletion vectors of + |the remaining files are read again, in bounded batches, when they are merged.""" + .stripMargin) + .bytesConf(ByteUnit.BYTE) + .checkValue(_ >= 0, "rowLevelConcurrency.maxRetainedDeletionVectorBytes must be non-negative") + .createWithDefault(64L * 1024L * 1024L) + 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..db97b0a5a9d --- /dev/null +++ b/spark/src/test/scala/org/apache/spark/sql/delta/RowLevelConcurrencySuite.scala @@ -0,0 +1,955 @@ +/* + * 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 java.util.concurrent.atomic.AtomicInteger + +import scala.concurrent.duration.Duration + +import org.apache.spark.sql.delta.actions.{Action, AddFile, RemoveFile} +import org.apache.spark.sql.delta.concurrency.PhaseLockingTestMixin +import org.apache.spark.sql.delta.concurrency.TransactionExecutionTestMixin +import org.apache.spark.sql.delta.files.TahoeBatchFileIndex +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.catalyst.expressions.{AttributeReference, EqualTo, Literal} +import org.apache.spark.sql.execution.QueryExecution +import org.apache.spark.sql.execution.datasources.{HadoopFsRelation, LogicalRelation} +import org.apache.spark.sql.functions.lit +import org.apache.spark.sql.test.SharedSparkSession +import org.apache.spark.sql.types.LongType +import org.apache.spark.sql.util.QueryExecutionListener +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. + * + * 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 reader-side data skipping (a separate change). + */ +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. + * `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").options(extraProperties).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") + } + + /** Runs `f` and returns how many row-level read-check scans it issued. */ + private def countReadCheckScans(f: => Unit): Int = { + val count = new AtomicInteger(0) + def isReadCheck(qe: QueryExecution): Boolean = qe.analyzed.exists { + case l: LogicalRelation => l.relation match { + case h: HadoopFsRelation => h.location match { + case index: TahoeBatchFileIndex => index.actionType == "rowLevelConcurrencyReadCheck" + case _ => false + } + case _ => false + } + case _ => false + } + val listener = new QueryExecutionListener { + override def onSuccess(funcName: String, qe: QueryExecution, durationNs: Long): Unit = + if (isReadCheck(qe)) count.incrementAndGet() + override def onFailure(funcName: String, qe: QueryExecution, exception: Exception): Unit = + if (isReadCheck(qe)) count.incrementAndGet() + } + spark.listenerManager.register(listener) + try { + f + spark.sparkContext.listenerBus.waitUntilEmpty() + } finally { + spark.listenerManager.unregister(listener) + } + count.get() + } + + 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("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')") + // Each DELETE reads only the rows matching its own condition, which are the rows it deletes, + // so neither read observes the other's disjoint delete. Committing in `winner ; current` + // order is a valid serialization even under Serializable, 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("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. + // A MERGE without a NOT MATCHED clause cannot observe the winner deleting a row it did not + // modify, so this reconciles 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) + 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 still used, so its DV cardinality is 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]) + } + } + + // --------------------------------------------------------------------------- + // 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") { + 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(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 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). + assert(ids(dir) === ((0L to 99L).filterNot(_ == 20) :+ 1020L).sorted) + assert(deletionVectorCardinalities(log) === Seq(1L)) + } + } + + 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)) + } + } + + // --------------------------------------------------------------------------- + // 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("crossed MERGEs whose sources read the target table are not reconciled") { + withTempDir { dir => + createSingleFileTableWithDVs(dir) + // A reads row 20 as its source and deletes row 5; B reads row 5 and deletes row 20. The + // deletes are disjoint, but either serial order deletes only one row: the second MERGE's + // source is empty. A's source read of row 20 must abort it. + def selfMerge(sourceId: Long, targetId: Long): String = + s"""MERGE INTO ${tableRef(dir)} t + |USING (SELECT id + ${targetId - sourceId} AS id FROM ${tableRef(dir)} + | WHERE id = $sourceId) s + |ON t.id = s.id WHEN MATCHED THEN DELETE""".stripMargin + val txnA = sqlTxn(selfMerge(sourceId = 20, targetId = 5), rowLevelConcurrency = true) + val txnB = sqlTxn(selfMerge(sourceId = 5, targetId = 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) + assert(ids(dir) === (0L to 99L).filterNot(_ == 20L)) + } + } + + 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 + // --------------------------------------------------------------------------- + + 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)) + } + } + + 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)) + } + } + + 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). + // 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. + // --------------------------------------------------------------------------- + + for ((name, confs, expectedDvFiles) <- Seq( + ("", Seq.empty[(String, String)], 1), + (", re-reading bitmaps that were not kept", + Seq(DeltaSQLConf.DELTA_ROW_LEVEL_CONCURRENCY_MAX_RETAINED_DV_BYTES.key -> "0"), 1), + (", one DV file per packing target size", + Seq(DeltaSQLConf.DELETION_VECTOR_PACKING_TARGET_SIZE.key -> "1"), 2))) + test(s"disjoint DELETEs spanning two files reconcile via the bounded-parallel path$name") { + withTempDir { dir => withSQLConf(confs: _*) { + // 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)) + // The merged DVs are packed into one DV file unless the packing target size forbids it. + val dvFiles = log.update().allFiles.collect().map(_.deletionVector.pathOrInlineDv).distinct + assert(dvFiles.length === expectedDvFiles) + }} + } + + // --------------------------------------------------------------------------- + // 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)) + } + } + + // --------------------------------------------------------------------------- + // 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))) + } + } + + // --------------------------------------------------------------------------- + // Reads of the current transaction that observe the winner's deletes + // --------------------------------------------------------------------------- + + /** + * The file actions of `DELETE FROM t WHERE id = ` against the table at `dir`, produced on a + * copy so they can be committed by a hand-driven transaction on `dir`. The copy's DV files are + * copied back so the actions resolve. + */ + private def deleteActionsFromCopy(dir: File, id: Long): Seq[Action] = + fileActionsFromCopy(dir)(copy => s"DELETE FROM ${tableRef(copy)} WHERE id = $id") + + /** The file actions of the `sqlOnCopy(copy)` command, produced on a copy of the table. */ + private def fileActionsFromCopy(dir: File)(sqlOnCopy: File => String): Seq[Action] = { + val copy = new File(dir.getParentFile, s"${dir.getName}-copy-${java.util.UUID.randomUUID()}") + org.apache.commons.io.FileUtils.copyDirectory(dir, copy) + sql(sqlOnCopy(copy)) + val copyLog = DeltaLog.forTable(spark, copy.getCanonicalPath) + copy.listFiles().filter(_.getName.contains("deletion_vector")) + .foreach(f => org.apache.commons.io.FileUtils.copyFileToDirectory(f, dir)) + copyLog.getChanges(copyLog.update().version).next()._2 + .filter(a => a.isInstanceOf[AddFile] || a.isInstanceOf[RemoveFile]) + } + + private val idAttr = AttributeReference("id", LongType)() + + /** + * A hand-driven transaction that runs `read`, then commits `DELETE ... WHERE id = 5`'s actions + * after a concurrent `DELETE ... WHERE id = 20` committed. The two deletes are row-disjoint. + */ + private def deleteFiveAfterRead(dir: File, log: DeltaLog)(read: OptimisticTransaction => Unit) + : scala.util.Try[Long] = { + val loserActions = deleteActionsFromCopy(dir, 5) + withSQLConf(DeltaSQLConf.DELTA_ROW_LEVEL_CONCURRENCY_ENABLED.key -> "true") { + val loser = log.startTransaction() + read(loser) + sql(s"DELETE FROM ${tableRef(dir)} WHERE id = 20") + scala.util.Try(loser.commit(loserActions, DeltaOperations.ManualUpdate)) + } + } + + private def assertConflict(result: scala.util.Try[Long]): Unit = assert( + result.failed.toOption.exists(e => + Seq("ConcurrentAppend", "ConcurrentDeleteRead", "ConcurrentDeleteDelete") + .exists(e.getClass.getName.contains)), + s"Expected a concurrency conflict, got: $result") + + test("a read of a row the winner deleted is not reconciled") { + withTempDir { dir => + val log = createSingleFileTableWithDVs(dir) + sql(s"ALTER TABLE ${tableRef(dir)} SET TBLPROPERTIES " + + s"('${DeltaConfigs.ISOLATION_LEVEL.key}' = 'Serializable')") + // The loser read row 20 ("delete 5 if 20 exists"); the winner deleted row 20. Merging the + // DVs would commit a read the winner invalidated. + assertConflict( + deleteFiveAfterRead(dir, log)(_.filterFiles(Seq(EqualTo(idAttr, Literal(20L)))))) + assert(ids(dir) === (0L to 99L).filterNot(_ == 20L)) + } + } + + for ((name, loserSql) <- Seq( + "DELETE" -> ((dir: File) => s"DELETE FROM ${tableRef(dir)} WHERE id = 50"), + "UPDATE" -> ((dir: File) => s"UPDATE ${tableRef(dir)} SET id = 1050 WHERE id = 50"))) { + test(s"a $name whose only read is its condition reconciles without a read-check scan") { + withTempDir { dir => + createSingleFileTableWithDVs(dir) + val txnA = sqlTxn(loserSql(dir), rowLevelConcurrency = true) + val txnB = + sqlTxn(s"DELETE FROM ${tableRef(dir)} WHERE id = 20", rowLevelConcurrency = true) + val scans = countReadCheckScans { + val (futureA, futureB) = runTxnsWithOrder__A_Start__B__A_End(txnA, txnB) + ThreadUtils.awaitResult(futureA, Duration.Inf) + ThreadUtils.awaitResult(futureB, Duration.Inf) + } + assert(scans === 0) + val expected = (0L to 99L).filterNot(id => id == 20L || id == 50L) + assert(ids(dir) === (if (name == "UPDATE") (expected :+ 1050L).sorted else expected)) + } + } + } + + test("a loser that aborts on another conflict scans no read and writes no merged DV") { + withTempDir { dir => + val log = createSingleFileTableWithDVs(dir) + val loserActions = deleteActionsFromCopy(dir, 5) + def dvFilesOnDisk(): Set[String] = + dir.listFiles().map(_.getName).filter(_.contains("deletion_vector_")).toSet + var result: scala.util.Try[Long] = null + var dvFilesWritten = Set.empty[String] + val scans = countReadCheckScans { + withSQLConf(DeltaSQLConf.DELTA_ROW_LEVEL_CONCURRENCY_ENABLED.key -> "true") { + val loser = log.startTransaction() + loser.filterFiles(Seq(EqualTo(idAttr, Literal(30L)))) + // The winner's UPDATE masks row 20 in the shared file, which alone would reconcile, and + // appends an image file with id 1020, which the loser's read cannot rule out. + sql(s"UPDATE ${tableRef(dir)} SET id = 1020 WHERE id = 20") + val before = dvFilesOnDisk() + result = scala.util.Try(loser.commit(loserActions, DeltaOperations.ManualUpdate)) + dvFilesWritten = dvFilesOnDisk() -- before + } + } + assert(result.failed.toOption.exists(_.isInstanceOf[ConcurrentAppendException]), result) + assert(scans === 0) + assert(dvFilesWritten.isEmpty, dvFilesWritten) + } + } + + test("a whole-table read is not reconciled") { + withTempDir { dir => + val log = createSingleFileTableWithDVs(dir) + assertConflict(deleteFiveAfterRead(dir, log)(_.readWholeTable())) + assert(ids(dir) === (0L to 99L).filterNot(_ == 20L)) + } + } + + test("a read the winner's deletes do not match is reconciled") { + withTempDir { dir => + val log = createSingleFileTableWithDVs(dir) + sql(s"ALTER TABLE ${tableRef(dir)} SET TBLPROPERTIES " + + s"('${DeltaConfigs.ISOLATION_LEVEL.key}' = 'Serializable')") + var result: scala.util.Try[Long] = null + val scans = countReadCheckScans { + result = deleteFiveAfterRead(dir, log)(_.filterFiles(Seq(EqualTo(idAttr, Literal(30L))))) + } + assert(result.isSuccess, s"Expected the DVs to merge, got: $result") + // A read that is not the DML's own condition is checked by scanning the shared file. + assert(scans === 1) + assert(ids(dir) === (0L to 99L).filterNot(id => id == 5L || id == 20L)) + assert(deletionVectorCardinalities(log) === Seq(2L)) + } + } + + // --------------------------------------------------------------------------- + // RESTORE and other commits that bring deleted rows back + // --------------------------------------------------------------------------- + + private def restoreTo(dir: File, version: Long): String = + s"RESTORE TABLE ${tableRef(dir)} TO VERSION AS OF $version" + + test("RESTORE winner that brings back a deleted row is not reconciled") { + withTempDir { dir => + createSingleFileTableWithDVs(dir) // v0 + sql(s"DELETE FROM ${tableRef(dir)} WHERE id = 1") // v1: dv = {1} + sql(s"DELETE FROM ${tableRef(dir)} WHERE id = 2") // v2: dv = {1, 2} + // A prepares dv = {1, 2, 50}; B restores dv = {1}. Merging them would delete row 2 again. + val txnA = sqlTxn(s"DELETE FROM ${tableRef(dir)} WHERE id = 50", rowLevelConcurrency = true) + val txnB = sqlTxn(restoreTo(dir, 1), 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) + assert(ids(dir) === (0L to 99L).filterNot(_ == 1L)) + } + } + + test("RESTORE winner that only deletes rows is not reconciled") { + withTempDir { dir => + createSingleFileTableWithDVs(dir) // v0 + sql(s"DELETE FROM ${tableRef(dir)} WHERE id = 1") // v1: dv = {1} + sql(restoreTo(dir, 0)) // v2: no dv + // B restores dv = {1}, a superset of A's base. RESTORE is still left to the file checks. + val txnA = sqlTxn(s"DELETE FROM ${tableRef(dir)} WHERE id = 50", rowLevelConcurrency = true) + val txnB = sqlTxn(restoreTo(dir, 1), 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) + assert(ids(dir) === (0L to 99L).filterNot(_ == 1L)) + } + } + + test("RESTORE loser aborts on a concurrent DELETE") { + withTempDir { dir => + createSingleFileTableWithDVs(dir) // v0 + sql(s"DELETE FROM ${tableRef(dir)} WHERE id = 1") // v1: dv = {1} + sql(s"DELETE FROM ${tableRef(dir)} WHERE id = 2") // v2: dv = {1, 2} + // A restores dv = {1}; B deletes row 50 from the same file. RESTORE commits through + // `commitLarge`, which aborts on any concurrent commit without running conflict resolution. + val txnA = sqlTxn(restoreTo(dir, 1), 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) } + assert(e.getCause.isInstanceOf[io.delta.exceptions.ConcurrentWriteException], e.getCause) + assert(ids(dir) === (0L to 99L).filterNot(id => Set(1L, 2L, 50L).contains(id))) + } + } + + test("RESTORE loser committed through conflict resolution is not reconciled") { + withTempDir { dir => + val log = createSingleFileTableWithDVs(dir) // v0 + sql(s"DELETE FROM ${tableRef(dir)} WHERE id = 1") // v1: dv = {1} + sql(restoreTo(dir, 0)) // v2: no dv + // The loser restores dv = {1} (rows only added) after reading row 1; the winner deletes row + // 20. Both DV checks pass, so only the RESTORE exclusion keeps this from reconciling. + val loserActions = fileActionsFromCopy(dir)(restoreTo(_, 1)) + val result = withSQLConf(DeltaSQLConf.DELTA_ROW_LEVEL_CONCURRENCY_ENABLED.key -> "true") { + val loser = log.startTransaction() + loser.filterFiles(Seq(EqualTo(idAttr, Literal(1L)))) + sql(s"DELETE FROM ${tableRef(dir)} WHERE id = 20") + scala.util.Try(loser.commit(loserActions, DeltaOperations.Restore(Some(1L), None))) + } + assertConflict(result) + assert(ids(dir) === (0L to 99L).filterNot(_ == 20L)) + } + } + + test("a non-RESTORE commit that brings back a deleted row is not reconciled") { + withTempDir { dir => + val log = createSingleFileTableWithDVs(dir) // v0 + sql(s"DELETE FROM ${tableRef(dir)} WHERE id = 1") // v1: dv = {1} + sql(s"DELETE FROM ${tableRef(dir)} WHERE id = 2") // v2: dv = {1, 2} + // The winner un-deletes row 2 like RESTORE does, but under another operation name (e.g. + // written by another engine). The DVs themselves show the winner's DV lost a base row. + val winnerActions = fileActionsFromCopy(dir)(restoreTo(_, 1)) + val loserActions = deleteActionsFromCopy(dir, 50) + val result = withSQLConf(DeltaSQLConf.DELTA_ROW_LEVEL_CONCURRENCY_ENABLED.key -> "true") { + val loser = log.startTransaction() + loser.filterFiles(Seq(EqualTo(idAttr, Literal(50L)))) + log.startTransaction().commit(winnerActions, DeltaOperations.ManualUpdate) + scala.util.Try(loser.commit(loserActions, DeltaOperations.ManualUpdate)) + } + assertConflict(result) + assert(ids(dir) === (0L to 99L).filterNot(_ == 1L)) + } + } + + test("MERGE with a conditional matched clause and an insert is not reconciled") { + withTempDir { dir => + createSingleFileTableWithDVs(dir) + // Source {5, 20}: 5 matches and is deleted; 20 matches the ON condition but not the clause, + // so it is left alone and NOT inserted. After the winner deletes 20, the same MERGE would + // insert 20 -- so committing the MERGE after the winner without rerunning it is wrong. + val txnA = sqlTxn( + s"""MERGE INTO ${tableRef(dir)} t USING (SELECT explode(array(5L, 20L)) AS id) s + |ON t.id = s.id + |WHEN MATCHED AND s.id = 5 THEN DELETE + |WHEN NOT MATCHED THEN INSERT *""".stripMargin, + 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) + assert(ids(dir) === (0L to 99L).filterNot(_ == 20L)) + } + } + + test("MERGE with an unconditional matched clause and an insert reconciles") { + withTempDir { dir => + val log = createSingleFileTableWithDVs(dir) + // Every matched row is modified, so a winner can only delete rows the MERGE did not match, + // which cannot change what it inserts. + val txnA = sqlTxn( + s"""MERGE INTO ${tableRef(dir)} t USING (SELECT explode(array(5L, 1000L)) AS id) s + |ON t.id = s.id + |WHEN MATCHED THEN DELETE + |WHEN NOT MATCHED THEN INSERT *""".stripMargin, + 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 == 5L || id == 20L) :+ 1000L)) + assert(deletionVectorCardinalities(log).contains(2L)) + } + } +}