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 0201dfe8403..1b80ff0759b 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 @@ -1148,6 +1148,17 @@ private[delta] class ConflictChecker( false } + /** + * Whether the row-level concurrency delete/read refinement (Case 1b) is active. It reads data + * during conflict detection, so it rides on the value-exact added-files skipping flags rather + * than a separate config: when on, a merge-on-read file the winner removed and re-added at the + * same path is excluded from the added-files check (it carries no new rows), and the delete/read + * check aborts only when a row the winner actually removed matches what the transaction read. + */ + private def deleteReadRowLevelRefinementEnabled: Boolean = + spark.conf.get(DeltaSQLConf.DELTA_CONFLICT_DETECTION_DATA_SKIPPING_ENABLED) && + spark.conf.get(DeltaSQLConf.DELTA_CONFLICT_DETECTION_DATA_SKIPPING_VALUE_EXACT_ENABLED) + /** * Check if the new files added by the already committed transactions should have been read by * the current transaction. @@ -1168,8 +1179,24 @@ private[delta] class ConflictChecker( Seq.empty } + // Row-level concurrency (Case 1b companion): a file the winner both removed and re-added at + // the same path is a merge-on-read modification (e.g. a DELETE that only widened its deletion + // vector), not new data -- its rows already existed in the current transaction's read + // snapshot, so it can never be a file the transaction "should have read but could not". The + // only conflict it can raise is delete/read, resolved row-level in + // checkForDeletedFilesAgainstCurrentTxnReadFiles; leaving it here would abort a reader + // disjoint from the removed rows with a spurious ConcurrentAppendException. Genuinely new + // rows from an UPDATE/MERGE land at a fresh path (not in removedFiles) and stay checked. + val addedFilesToCheck = + if (deleteReadRowLevelRefinementEnabled) { + val reAddedPaths = winningCommitSummary.removedFiles.map(_.path).toSet + addedFilesToCheckForConflicts.filterNot(a => reAddedPaths.contains(a.path)) + } else { + addedFilesToCheckForConflicts + } + val fileMatchingPartitionReadPredicates = - getFirstFileMatchingPartitionPredicates(addedFilesToCheckForConflicts) + getFirstFileMatchingPartitionPredicates(addedFilesToCheck) if (fileMatchingPartitionReadPredicates.nonEmpty) { throw DeltaErrors.concurrentAppendException( @@ -1190,7 +1217,30 @@ private[delta] class ConflictChecker( // Fail if files have been deleted that the txn read. val readFilePaths = currentTransactionInfo.readFiles.map( f => f.path -> f.partitionValues).toMap - val deleteReadOverlap = winningCommitSummary.removedFiles + // Row-level refinement (Case 1b), the delete/read analogue of the added-files skipping in + // getFirstFileMatchingPartitionPredicates: of the removed files the txn read, drop those the + // winner removed no read-matching row from. Same guards as that skipping; when off / a + // whole-table read / no read predicates, this keeps today's path-keyed abort over all + // removed files. + val candidateRemovedFiles = + if (deleteReadRowLevelRefinementEnabled && + !currentTransactionInfo.readWholeTable && + currentTransactionInfo.readPredicates.nonEmpty) { + val overlap = + winningCommitSummary.removedFiles.filter(r => readFilePaths.contains(r.path)) + val remaining = removedFilesWithReadMatchingRemovedRows(overlap) + if (remaining.size < overlap.size) { + recordDeltaEvent(deltaLog, + opType = "delta.conflictDetection.deleteReadDataSkipping.filesSkipped", + data = Map( + "candidateFiles" -> overlap.size, + "skippedFiles" -> (overlap.size - remaining.size))) + } + remaining + } else { + winningCommitSummary.removedFiles + } + val deleteReadOverlap = candidateRemovedFiles .find(r => readFilePaths.contains(r.path)) if (deleteReadOverlap.nonEmpty) { val partitionOpt = getPrettyPartitionMessage(readFilePaths(deleteReadOverlap.get.path)) @@ -1210,6 +1260,56 @@ private[delta] class ConflictChecker( } } + /** + * Row-level refinement of the delete/read check (row-level concurrency Case 1b): given the + * concurrently-removed files the current transaction read (`overlappingRemovedFiles`), returns + * those still in conflict -- i.e. the winner removed from them at least one row matching the + * transaction's read predicates. It is an all-or-nothing existence check, so the result is either + * every file (conflict) or none (the removed rows are proven not to match). The delete/read + * analogue of the value-exact added-files skipping in + * [[getFirstFileMatchingPartitionPredicates]]. + * + * The removed rows are obtained without an inverse deletion-vector read: a `RemoveFile` carries + * the pre-image DV and the winner's paired re-added [[AddFile]] (if any) the post-image DV, and + * since a DML only adds deletions the post-image live rows are a subset of the pre-image ones, so + * matches(removed) = matches(pre-image live) - matches(post-image live). One-way safe and + * fail-safe: any missing information or error keeps every file as a conflict (the count check + * is `ConflictDataSkippingReader.anyRemovedRowMatchesReadPredicate`). + */ + private def removedFilesWithReadMatchingRemovedRows( + overlappingRemovedFiles: Seq[RemoveFile]): Seq[RemoveFile] = { + if (overlappingRemovedFiles.isEmpty) return overlappingRemovedFiles + val readSnapshot = currentTransactionInfo.readSnapshot + // A partitioned removed file that did not record its partition values cannot be read back + // safely, so keep every file as a conflict. + if (readSnapshot.metadata.partitionColumns.nonEmpty && + overlappingRemovedFiles.exists(_.partitionValues == null)) { + return overlappingRemovedFiles + } + // Pre-image view: each removed file under the DV it carried before removal. Post-image view: + // the winner's re-added file at the same path (absent for a full-file removal). + val preImageFiles = overlappingRemovedFiles.map(r => + AddFile( + path = r.path, + partitionValues = Option(r.partitionValues).getOrElse(Map.empty), + size = r.size.getOrElse(0L), + modificationTime = 0L, + dataChange = false, + stats = r.stats, + tags = r.tags, + deletionVector = r.deletionVector)) + val postImageFiles = overlappingRemovedFiles + .flatMap(r => winningCommitSummary.addedFilePathToActionMap.get(r.path)) + if (readSnapshot.anyRemovedRowMatchesReadPredicate( + preImageFiles, + postImageFiles, + currentTransactionInfo.readPredicates.map(_.dataPredicates).toSeq)) { + overlappingRemovedFiles + } else { + Seq.empty + } + } + /** * Check if [[RemoveFile]] actions added by already committed transactions conflicts with * [[RemoveFile]] actions this transaction is trying to add. 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 aa94d616968..2dceec0bb24 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 @@ -555,7 +555,18 @@ trait DeltaSQLConfBase extends DeltaSQLConfUtils { |transaction's read predicates. This resolves predicates min/max stats cannot skip (e.g. |modulo or other non-range expressions), at the cost of reading the (already stats- |narrowed) added files during commit. One-way safe: a file is excluded only when an actual - |scan proves no row matches.""".stripMargin) + |scan proves no row matches. + | + |This flag additionally refines the delete/read check with the same read-the-data + |approach: a merge-on-read file the winner removed and re-added at the same path carries + |no new rows, so it is excluded from the added-files check, and the delete/read check + |aborts only when a row the winner actually removed matches the current transaction's read + |predicates. The removed rows are obtained without an inverse deletion-vector read: + |because the winner's new deletion vector is a superset of the pre-image one, the count of + |predicate-matching removed rows equals matches(pre-image live view) - matches(post-image + |live view), so two ordinary reads of the (few) overlapping files suffice. One-way safe: + |the loser aborts unless the removed rows are proven not to match; any error or missing + |information falls back to the path-keyed abort.""".stripMargin) .booleanConf .createWithDefault(false) diff --git a/spark/src/main/scala/org/apache/spark/sql/delta/stats/ConflictDataSkippingReader.scala b/spark/src/main/scala/org/apache/spark/sql/delta/stats/ConflictDataSkippingReader.scala index b95b1dcb6ae..d573975b286 100644 --- a/spark/src/main/scala/org/apache/spark/sql/delta/stats/ConflictDataSkippingReader.scala +++ b/spark/src/main/scala/org/apache/spark/sql/delta/stats/ConflictDataSkippingReader.scala @@ -29,7 +29,7 @@ import org.apache.spark.sql.delta.sources.DeltaSQLConf import org.apache.spark.SparkContext import org.apache.spark.sql.{Column, DataFrame} import org.apache.spark.sql.catalyst.expressions._ -import org.apache.spark.sql.functions.{col, from_json} +import org.apache.spark.sql.functions.{coalesce, col, from_json, lit, sum} /** * Reader-side data skipping used during conflict detection (row-level concurrency Case 1: writers @@ -275,6 +275,84 @@ trait ConflictDataSkippingReader extends DeltaLogging { self: DataSkippingReader } } + /** + * Delete/read conflict refinement (row-level concurrency Case 1b): decides whether any row the + * winning transaction REMOVED from the given files matches the current transaction's read + * predicates. The caller ([[org.apache.spark.sql.delta.ConflictChecker]]) uses this to replace + * the path-keyed delete/read abort with a row-level one -- a concurrently-removed file the + * transaction read is a genuine conflict only when a removed row actually matches what it read. + * + * Rather than reading the removed rows directly (which would need an inverse deletion-vector + * scan), we exploit that a DML only ever ADDS deletions: the winner's post-image deletion vector + * is a superset of the pre-image one, so its post-image live rows are a SUBSET of the pre-image + * live rows, and the removed rows are exactly their difference. Hence, for the read condition + * `cond` (OR across reads, AND within a read), summed over the overlapping files: + * + * matches(removed) = matches(pre-image live rows) - matches(post-image live rows) + * + * `preImageFiles` are the removed files viewed under their pre-image DV (one [[AddFile]] per + * `RemoveFile`); `postImageFiles` are the winner's paired re-added [[AddFile]]s under their + * post-image DV (absent for a full-file removal, whose removed set is the whole pre-image). Every + * per-file term is non-negative (subset), so the total difference is >= 1 iff some removed row + * matches. The two images are unioned with signed weights (+1 pre, -1 post) and summed in a + * SINGLE Spark job over the (few) overlapping files, with no per-file attribution and no giant + * row-index predicate. + * + * One-way safe: returns true (conflict) unless the scan PROVES no removed row matches. A read + * with no eligible (deterministic, subquery-free, non-metadata) filter matches everything, so it + * cannot prove non-match and returns true. Any failure falls back to true (path-keyed abort). + */ + private[delta] def anyRemovedRowMatchesReadPredicate( + preImageFiles: Seq[AddFile], + postImageFiles: Seq[AddFile], + dataFiltersPerRead: Seq[Seq[Expression]]): Boolean = { + if (preImageFiles.isEmpty || dataFiltersPerRead.isEmpty || schema.isEmpty) return true + try { + // AND within a read over its eligible filters; a read with none matches everything, so we + // cannot prove any removed row fails it -> conservative conflict. + val perReadEligible = dataFiltersPerRead.map(eligibleSkippingFilters) + if (perReadEligible.exists(_.isEmpty)) return true + val snapshot = snapshotToScan + val sc = spark.sparkContext + val prevJobDesc = sc.getLocalProperty(SparkContext.SPARK_JOB_DESCRIPTION) + val removedMatchCount = + try { + sc.setJobDescription("Delta conflict detection: delete/read row-level skipping") + recordFrameProfile("Delta", "DataSkippingReader.anyRemovedRowMatchesReadPredicate") { + // Rows of `files` matching the read condition, tagged with `weight` so a single + // aggregation over the union yields the signed pre-minus-post difference. The condition + // is rebound to each freshly built DataFrame's columns (OR across reads, AND within). + def matchingRowWeights(files: Seq[AddFile], weight: Long): DataFrame = { + val df = snapshot.deltaLog.createDataFrame( + snapshot, files, actionTypeOpt = Some("conflictDetectionDeleteRead")) + val condition = perReadEligible + .map(filters => filters.map(f => rebindToDataFrame(f, df)).reduce(_ && _)) + .reduce(_ || _) + df.where(condition).select(lit(weight).as("weight")) + } + // matches(removed) = matches(pre-image live) - matches(post-image live), evaluated in a + // SINGLE Spark job: union the per-image weighted matches (+1 pre, -1 post) and sum. A + // full-file removal has no post-image, so the pre-image count alone is the removed set. + val weights = + if (postImageFiles.isEmpty) matchingRowWeights(preImageFiles, 1L) + else matchingRowWeights(preImageFiles, 1L) + .union(matchingRowWeights(postImageFiles, -1L)) + weights.select(coalesce(sum(col("weight")), lit(0L))).head().getLong(0) + } + } finally { + sc.setJobDescription(prevJobDesc) + } + removedMatchCount > 0 + } catch { + case NonFatal(e) => + // Optimization only: never let a scan failure abort or silently pass a commit. Fall back to + // the default (feature-off) behavior of treating the removed files as a conflict. + logWarning(log"Conflict-time delete/read row-level skipping failed to evaluate; falling " + + log"back to treating the removed files as a conflict", e) + true + } + } + /** * Rebinds `e`'s attribute references to `df`'s output columns by name, so a read predicate * resolved against the transaction's plan can be evaluated on a freshly built DataFrame. Nested diff --git a/spark/src/test/scala/org/apache/spark/sql/delta/DeleteReadConflictDataSkippingSuite.scala b/spark/src/test/scala/org/apache/spark/sql/delta/DeleteReadConflictDataSkippingSuite.scala new file mode 100644 index 00000000000..e52b28eb1ea --- /dev/null +++ b/spark/src/test/scala/org/apache/spark/sql/delta/DeleteReadConflictDataSkippingSuite.scala @@ -0,0 +1,305 @@ +/* + * 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 org.apache.spark.sql.delta.actions.AddFile +import org.apache.spark.sql.delta.sources.DeltaSQLConf +import org.apache.spark.sql.delta.test.DeltaSQLCommandTest + +import org.apache.spark.sql.{QueryTest, SaveMode} +import org.apache.spark.sql.catalyst.expressions.{AttributeReference, EqualTo, Expression, GreaterThanOrEqual, LessThan, Literal, Remainder} +import org.apache.spark.sql.test.SharedSparkSession +import org.apache.spark.sql.types.LongType + +/** + * Tests for the delete/read row-level refinement of conflict detection (row-level concurrency + * Case 1b), which rides on the value-exact conflict-skipping flags + * ([[DeltaSQLConf.DELTA_CONFLICT_DETECTION_DATA_SKIPPING_ENABLED]] + + * [[DeltaSQLConf.DELTA_CONFLICT_DETECTION_DATA_SKIPPING_VALUE_EXACT_ENABLED]]). + * + * The path-keyed delete/read check aborts the current transaction whenever a file the winner + * removed is in the current transaction's read set, regardless of whether the removed rows actually + * match what it read. This refinement reads the rows the winner ACTUALLY removed and conflicts only + * when one matches the read predicate. It is the delete/read analogue of the value-exact + * added-files skipping in [[ConflictDataSkippingSuite]]; it must be one-way safe (abort unless the + * removed rows are proven not to match) and fail-safe (any error keeps today's abort). A companion + * fix excludes a merge-on-read re-add (a path in both the winner's added and removed sets) from the + * added-files check, since it carries no new rows -- otherwise a reader disjoint from the removed + * rows would still abort on the added-files arm. + */ +class DeleteReadConflictDataSkippingSuite extends QueryTest + with SharedSparkSession + with DeltaSQLCommandTest { + + private def tableRef(dir: File): String = s"delta.`${dir.getCanonicalPath}`" + + private val id: AttributeReference = AttributeReference("id", LongType)() + private def lt(v: Long): Expression = LessThan(id, Literal(v)) + private def ge(v: Long): Expression = GreaterThanOrEqual(id, Literal(v)) + private def even: Expression = EqualTo(Remainder(id, Literal(2L)), Literal(0L)) + private def odd: Expression = EqualTo(Remainder(id, Literal(2L)), Literal(1L)) + + private def manufacturedAdd(name: String): AddFile = + AddFile(name, Map.empty[String, String], size = 1L, modificationTime = 1L, dataChange = true) + + // --------------------------------------------------------------------------------------------- + // Direct reader tests: the count-difference core, exercised against real deletion vectors. + // --------------------------------------------------------------------------------------------- + + test("anyRemovedRowMatchesReadPredicate: true only when a removed row matches (real DVs)") { + withTempDir { dir => + val path = dir.getCanonicalPath + // DV-enabled single file, ids [0, 100). + spark.range(0, 100).repartition(1) + .write.format("delta").mode("append").save(path) + sql(s"ALTER TABLE ${tableRef(dir)} SET TBLPROPERTIES " + + s"('delta.enableDeletionVectors' = 'true')") + val log = DeltaLog.forTable(spark, path) + + // Pre-image: the file with no deletions (all of [0, 100) live). + val preImage = log.update().allFiles.collect().toSeq + assert(preImage.size == 1, s"expected a single file, got ${preImage.size}") + + // Winner deletes ids [0, 10) via a merge-on-read deletion vector (same file re-added). + sql(s"DELETE FROM ${tableRef(dir)} WHERE id < 10") + val postSnapshot = log.update() + val postImage = postSnapshot.allFiles.collect().toSeq + assert(postImage.size == 1, "expected the same file re-added with a DV, not a COW rewrite") + assert(postImage.head.path == preImage.head.path, "a DV delete must re-add at the same path") + assert(postImage.head.deletionVector != null, "expected a deletion vector (merge-on-read)") + + def matches(pred: Expression): Boolean = + postSnapshot.anyRemovedRowMatchesReadPredicate(preImage, postImage, Seq(Seq(pred))) + + // Removed rows are exactly [0, 10). + assert(matches(lt(5)), "id < 5 intersects the removed rows [0, 10) -> conflict") + assert(matches(lt(10)), "id < 10 equals the removed rows -> conflict") + assert(!matches(ge(50)), "id >= 50 is disjoint from the removed rows -> no conflict") + assert(!matches(ge(10)), "id >= 10 excludes the removed rows -> no conflict") + } + } + + test("anyRemovedRowMatchesReadPredicate: full-file removal counts every pre-image row removed") { + withTempDir { dir => + val path = dir.getCanonicalPath + spark.range(0, 100).repartition(1) + .write.format("delta").mode("append").save(path) + val log = DeltaLog.forTable(spark, path) + val preImage = log.update().allFiles.collect().toSeq + assert(preImage.size == 1) + val snapshot = log.update() + + // No post-image file (a full-file removal): matches(removed) == matches(pre-image live rows). + def matches(pred: Expression): Boolean = + snapshot.anyRemovedRowMatchesReadPredicate(preImage, Seq.empty, Seq(Seq(pred))) + + assert(matches(lt(5)), "some removed row satisfies id < 5 -> conflict") + assert(!matches(ge(200)), "no removed row satisfies id >= 200 -> no conflict") + } + } + + test("anyRemovedRowMatchesReadPredicate: OR across reads, conflict if any read matches") { + withTempDir { dir => + val path = dir.getCanonicalPath + spark.range(0, 100).repartition(1) + .write.format("delta").mode("append").save(path) + val log = DeltaLog.forTable(spark, path) + val preImage = log.update().allFiles.collect().toSeq + val snapshot = log.update() + + // Removed = whole pre-image. One read disjoint (id >= 200), one matching (id < 5) => conflict + assert( + snapshot.anyRemovedRowMatchesReadPredicate( + preImage, Seq.empty, Seq(Seq(ge(200)), Seq(lt(5)))), + "a removed row matches the second read -> conflict") + // Both reads disjoint from [0, 100) -> no conflict. + assert( + !snapshot.anyRemovedRowMatchesReadPredicate( + preImage, Seq.empty, Seq(Seq(ge(200)), Seq(lt(-5)))), + "no removed row matches either read -> no conflict") + } + } + + test("anyRemovedRowMatchesReadPredicate: a read with no eligible filter conflicts") { + withTempDir { dir => + val path = dir.getCanonicalPath + spark.range(0, 100).repartition(1) + .write.format("delta").mode("append").save(path) + val log = DeltaLog.forTable(spark, path) + val preImage = log.update().allFiles.collect().toSeq + val snapshot = log.update() + + // An empty read matches everything, so we cannot prove non-match -> conservative conflict. + assert(snapshot.anyRemovedRowMatchesReadPredicate(preImage, Seq.empty, Seq(Seq.empty)), + "a read with no eligible filter must conflict") + } + } + + test("anyRemovedRowMatchesReadPredicate: unresolvable predicate falls back to conflict") { + withTempDir { dir => + val path = dir.getCanonicalPath + spark.range(0, 100).repartition(1) + .write.format("delta").mode("append").save(path) + val log = DeltaLog.forTable(spark, path) + val preImage = log.update().allFiles.collect().toSeq + val snapshot = log.update() + + // A predicate on a column that does not exist: rebinding throws and the fail-safe conflicts + // rather than letting the scan failure silently pass (or abort) the commit. + val ghost = EqualTo(AttributeReference("does_not_exist", LongType)(), Literal(1L)) + assert(snapshot.anyRemovedRowMatchesReadPredicate(preImage, Seq.empty, Seq(Seq(ghost))), + "an unresolvable predicate must fall back to conflict") + } + } + + // --------------------------------------------------------------------------------------------- + // End-to-end tests: a reader loser racing a concurrent DELETE winner. The winner removes an + // entire file (all rows match), so it re-adds nothing -- only the delete/read arm is exercised, + // never the added-files arm. + // --------------------------------------------------------------------------------------------- + + /** + * Runs a reader loser that scans the single all-even file under `readPredicate` against a winner + * that deletes the whole file, with the refinement toggled by `enabled`. Returns whether the + * loser committed (true) or aborted with a delete/read conflict (false). + */ + private def runDeleteReadRace(readPredicate: Expression, enabled: Boolean): Boolean = { + var committed = false + withTempDir { dir => + val path = dir.getCanonicalPath + // Single file of all-EVEN ids in [0, 100). DVs are OFF (the default), so the winner's + // whole-file delete removes it outright with no re-added survivor file. + spark.range(0, 100, step = 2).repartition(1) + .write.format("delta").mode("append").save(path) + val log = DeltaLog.forTable(spark, path) + withSQLConf( + DeltaSQLConf.DELTA_CONFLICT_DETECTION_DATA_SKIPPING_ENABLED.key -> enabled.toString, + DeltaSQLConf.DELTA_CONFLICT_DETECTION_DATA_SKIPPING_VALUE_EXACT_ENABLED.key -> + enabled.toString) { + val loser = log.startTransaction() + val readFiles = loser.filterFiles(Seq(readPredicate)) + assert(readFiles.size == 1, s"loser must read the file, got ${readFiles.size}") + + // Winner removes the whole file (every row matches) -> RemoveFile only, no AddFile. + sql(s"DELETE FROM ${tableRef(dir)} WHERE id >= 0") + + committed = + try { + loser.commit( + Seq(manufacturedAdd("loser.parquet")), DeltaOperations.Write(SaveMode.Append)) + true + } catch { + case _: io.delta.exceptions.ConcurrentDeleteReadException => false + } + } + } + committed + } + + test("e2e: removed rows disjoint from the read predicate -> loser commits when enabled") { + // The winner removed all-even rows; the loser only read odd rows, so no removed row matches. + assert(runDeleteReadRace(odd, enabled = true), + "removed rows are all even, read predicate is odd -> no conflict") + } + + test("e2e: removed rows match the read predicate -> loser still conflicts when enabled") { + // The winner removed all-even rows; the loser read even rows, so removed rows match. + assert(!runDeleteReadRace(even, enabled = true), + "removed rows are even, read predicate is even -> conflict") + } + + test("e2e: feature disabled -> disjoint removed rows still conflict (path-keyed)") { + assert(!runDeleteReadRace(odd, enabled = false), + "with the refinement off, any removed-and-read file conflicts") + } + + test("e2e: whole-table read is never refined -> conflicts with a concurrent delete") { + withTempDir { dir => + val path = dir.getCanonicalPath + spark.range(0, 100, step = 2).repartition(1) + .write.format("delta").mode("append").save(path) + val log = DeltaLog.forTable(spark, path) + withSQLConf( + DeltaSQLConf.DELTA_CONFLICT_DETECTION_DATA_SKIPPING_ENABLED.key -> "true", + DeltaSQLConf.DELTA_CONFLICT_DETECTION_DATA_SKIPPING_VALUE_EXACT_ENABLED.key -> "true") { + val loser = log.startTransaction() + loser.readWholeTable() + sql(s"DELETE FROM ${tableRef(dir)} WHERE id >= 0") + intercept[io.delta.exceptions.ConcurrentDeleteReadException] { + loser.commit( + Seq(manufacturedAdd("loser.parquet")), DeltaOperations.Write(SaveMode.Append)) + } + } + } + } + + // --------------------------------------------------------------------------------------------- + // Combined merge-on-read tests: a real DELETE winner that re-adds the file with a wider deletion + // vector, exercising both arms together. The re-added file's live rows still match a disjoint + // reader's predicate, so without the added-files re-add exclusion the reader would abort on the + // ADD arm (ConcurrentAppendException) and never reach the row-level delete/read refinement. + // --------------------------------------------------------------------------------------------- + + /** + * Reader loser scans a DV-enabled single file under `readPredicate` while a winner deletes + * `[0, 10)` via a merge-on-read deletion vector (RemoveFile(P, oldDV) + AddFile(P, newDV)). + * Returns "committed", "append-conflict" or "deleteread-conflict". + */ + private def runMoRDeleteRace(readPredicate: Expression): String = { + var outcome = "committed" + withTempDir { dir => + val path = dir.getCanonicalPath + spark.range(0, 100).repartition(1) + .write.format("delta").mode("append").save(path) + sql(s"ALTER TABLE ${tableRef(dir)} SET TBLPROPERTIES " + + s"('delta.enableDeletionVectors' = 'true')") + val log = DeltaLog.forTable(spark, path) + withSQLConf( + DeltaSQLConf.DELTA_CONFLICT_DETECTION_DATA_SKIPPING_ENABLED.key -> "true", + DeltaSQLConf.DELTA_CONFLICT_DETECTION_DATA_SKIPPING_VALUE_EXACT_ENABLED.key -> "true") { + val loser = log.startTransaction() + loser.filterFiles(Seq(readPredicate)) + sql(s"DELETE FROM ${tableRef(dir)} WHERE id < 10") + outcome = + try { + loser.commit( + Seq(manufacturedAdd("loser.parquet")), DeltaOperations.Write(SaveMode.Append)) + "committed" + } catch { + case _: io.delta.exceptions.ConcurrentAppendException => "append-conflict" + case _: io.delta.exceptions.ConcurrentDeleteReadException => "deleteread-conflict" + } + } + } + outcome + } + + test("e2e MoR: reader disjoint from removed rows commits (add-arm re-add is not new data)") { + // Winner removed [0, 10); the re-added file's live rows [10, 100) include the read range, but + // they are not new data, so the reader must commit rather than hit ConcurrentAppendException. + assert(runMoRDeleteRace(ge(50)) == "committed", + "removed rows [0, 10) are disjoint from the read id >= 50 -> loser commits") + } + + test("e2e MoR: reader overlapping the removed rows aborts with delete/read") { + // Winner removed [0, 10); the reader read id < 5, which a removed row satisfies. + assert(runMoRDeleteRace(lt(5)) == "deleteread-conflict", + "removed rows [0, 10) intersect the read id < 5 -> delete/read conflict") + } +}