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..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 @@ -216,11 +216,16 @@ object WinningCommitSummary { } private[delta] class ConflictChecker( - spark: SparkSession, + protected val spark: SparkSession, initialCurrentTransactionInfo: CurrentTransactionInfo, - 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 { + extends DeltaLogging with ConflictCheckerPredicateElimination + with RowLevelConcurrencyResolution { protected val winningCommitVersion = winningCommitSummary.commitVersion protected val startTimeMs = System.currentTimeMillis() @@ -300,6 +305,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() 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..84f40e32911 --- /dev/null +++ b/spark/src/main/scala/org/apache/spark/sql/delta/RowLevelConcurrencyResolution.scala @@ -0,0 +1,310 @@ +/* + * 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 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 +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. 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 => + + /** 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. + * + * 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 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 + + recordTime("resolved-row-level-conflicts") { + val dvStore = DeletionVectorStore.createInstance(deltaLog.newDeltaHadoopConf()) + val tablePath = deltaLog.dataPath + + // 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)) + .map(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) + 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 = 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 + } + 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" -> 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 + * 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). + // 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) + + 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, + 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() + } + } +} 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..edc43929b81 --- /dev/null +++ b/spark/src/test/scala/org/apache/spark/sql/delta/RowLevelConcurrencySuite.scala @@ -0,0 +1,611 @@ +/* + * 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. + * + * 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") + } + + 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')") + // 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("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) + 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("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. + // --------------------------------------------------------------------------- + + 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) + // --------------------------------------------------------------------------- + + 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))) + } + } +}