From 4fcc895726bf468dfa0f2081558a249a6a9fbc7f Mon Sep 17 00:00:00 2001 From: Eunjin Song Date: Tue, 28 Jul 2026 23:44:31 -0700 Subject: [PATCH 1/3] [Spark] Exclude dataChange=false added files from the append conflict check OPTIMIZE (compaction / Z-ORDER) commits its compacted outputs as dataChange=false AddFiles. Because OPTIMIZE is not a blind append (isBlindAppend=false), those files land in changedDataAddedFiles, so a concurrent non-blind writer (UPDATE/DELETE/MERGE with a read predicate) hits a spurious ConcurrentAppendException -- even though OPTIMIZE changed no logical data. A dataChange=false file only rearranges rows that already existed, so it can never be a row the losing txn "should have read." Add conflictDetection.excludeNoDataChangeAddedFiles.enabled (internal, default off) that filters dataChange=false files out of the added-files check in checkForAddedFilesThatShouldHaveBeenReadByCurrentTxn. All FileActions in a commit share one dataChange value (see trackConsistentDataChange), so this is effectively a commit-level gate. Tests: OptimisticTransactionSuite +5 cases -- OPTIMIZE-style dataChange=false winner vs (a) a concurrent partition read and (b) a concurrent whole-table read (the unpartitioned / Liquid Clustering scenario), each flag on -> reconcile / off -> legacy abort; plus a safety-floor case proving a genuine dataChange=true append still raises ConcurrentAppendException with the flag on. Full OptimisticTransactionSuite 148/148 green. Co-Authored-By: Claude Opus 4.8 --- .../spark/sql/delta/ConflictChecker.scala | 16 +- .../sql/delta/sources/DeltaSQLConf.scala | 11 ++ .../delta/OptimisticTransactionSuite.scala | 153 ++++++++++++++++++ 3 files changed, 179 insertions(+), 1 deletion(-) diff --git a/spark/src/main/scala/org/apache/spark/sql/delta/ConflictChecker.scala b/spark/src/main/scala/org/apache/spark/sql/delta/ConflictChecker.scala index d5cd7be1beb..de6f66dacb8 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 @@ -1137,8 +1137,22 @@ private[delta] class ConflictChecker( Seq.empty } + // Files added with `dataChange = false` (e.g. OPTIMIZE compaction / Z-ORDER outputs) + // only rearrange rows that already exist in the table; they introduce no new logical + // rows a concurrent reader could have missed. Because OPTIMIZE is not a blind append + // (isBlindAppend = false), its outputs otherwise fall into `changedDataAddedFiles` and + // make a concurrent non-blind writer raise a spurious ConcurrentAppendException against + // OPTIMIZE. When enabled, drop them so only genuinely new data is checked. All + // FileActions in a commit share one `dataChange` value (see `trackConsistentDataChange`). + val addedFilesWithoutNoDataChange = + if (spark.conf.get(DeltaSQLConf.DELTA_CONFLICT_DETECTION_EXCLUDE_NO_DATA_CHANGE_ADDS)) { + addedFilesToCheckForConflicts.filter(_.dataChange) + } else { + addedFilesToCheckForConflicts + } + val fileMatchingPartitionReadPredicates = - getFirstFileMatchingPartitionPredicates(addedFilesToCheckForConflicts) + getFirstFileMatchingPartitionPredicates(addedFilesWithoutNoDataChange) if (fileMatchingPartitionReadPredicates.nonEmpty) { throw DeltaErrors.concurrentAppendException( 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..14d57c5ed49 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 @@ -2240,6 +2240,17 @@ trait DeltaSQLConfBase extends DeltaSQLConfUtils { .booleanConf .createWithDefault(true) + val DELTA_CONFLICT_DETECTION_EXCLUDE_NO_DATA_CHANGE_ADDS = + buildConf("conflictDetection.excludeNoDataChangeAddedFiles.enabled") + .doc("When enabled, files added by a winning transaction with dataChange = false " + + "(e.g. OPTIMIZE compaction / Z-ORDER outputs) are excluded from the added-files " + + "conflict check. Such files only rearrange rows that already exist in the table and " + + "introduce no new logical rows, so a concurrent non-blind writer should not raise a " + + "ConcurrentAppendException against them. Off by default to preserve existing behavior.") + .internal() + .booleanConf + .createWithDefault(false) + val DELTA_UNIFORM_ICEBERG_TABLE_V3_ENABLED = buildConf("uniform.iceberg.v3.enabled") .internal() diff --git a/spark/src/test/scala/org/apache/spark/sql/delta/OptimisticTransactionSuite.scala b/spark/src/test/scala/org/apache/spark/sql/delta/OptimisticTransactionSuite.scala index 9dc8fd5ebe9..39182e0730a 100644 --- a/spark/src/test/scala/org/apache/spark/sql/delta/OptimisticTransactionSuite.scala +++ b/spark/src/test/scala/org/apache/spark/sql/delta/OptimisticTransactionSuite.scala @@ -804,6 +804,159 @@ class OptimisticTransactionSuite RemoveFile("b", None, partitionValues = Map(partCol -> "1"))) ) + // Case: an OPTIMIZE-style winner that only rearranges existing rows (dataChange = false, + // isBlindAppend unset -> its adds fall into `changedDataAddedFiles`) must not raise a + // spurious ConcurrentAppendException against a concurrent non-blind writer that reads the + // same partition. Gated by conflictDetection.excludeNoDataChangeAddedFiles.enabled. + for (excludeNoDataChange <- BOOLEAN_DOMAIN) { + test("dataChange = false append (OPTIMIZE-style) vs concurrent partition read, " + + s"excludeNoDataChangeAddedFiles = $excludeNoDataChange") { + withSQLConf( + DeltaSQLConf.DELTA_CONFLICT_DETECTION_EXCLUDE_NO_DATA_CHANGE_ADDS.key -> + excludeNoDataChange.toString) { + withTempDir { tempDir => + val partCol = "part" + val log = DeltaLog.forTable(spark, tempDir.getCanonicalPath) + + // Partitioned table with one file per partition. + log.startTransaction().commit(Seq( + Metadata( + schemaString = new StructType() + .add(partCol, IntegerType) + .add("value", IntegerType).json, + partitionColumns = Seq(partCol)) + ), ManualUpdate) + log.startTransaction().commit( + Seq(AddFile("a", Map(partCol -> "0"), 1, 1, dataChange = true), + AddFile("b", Map(partCol -> "1"), 1, 1, dataChange = true)), + ManualUpdate) + + // txn1 (loser): a non-blind writer that reads partition 0. + val newData = Seq(AddFile("x", Map(partCol -> "0"), 1, 1, dataChange = true)) + val txn = log.startTransaction() + val addFiles = txn.filterFiles(newData) + + // txn2 (winner): OPTIMIZE-style compaction of partition 0 -- its output carries + // dataChange = false (rearranges existing rows, introduces no new logical rows). + log.startTransaction().commit( + Seq(AddFile("y", Map(partCol -> "0"), 1, 1, dataChange = false)), ManualUpdate) + + // txn1 commits: removes the file it read and appends its new data. + def commitTxn1(): Unit = txn.commit(addFiles.map(_.remove) ++ newData, ManualUpdate) + + if (excludeNoDataChange) { + // dataChange = false output is excluded from the append check -> no false conflict. + commitTxn1() + val files = log.update().allFiles.collect() + // partition 0 now holds 'y' (OPTIMIZE output) + 'x' (txn1); 'a' was removed. + assert(files.count(_.partitionValues.get(partCol).contains("0")) == 2) + assert(files.length == 3) + } else { + // Legacy behavior: OPTIMIZE's output is treated as changed data -> spurious abort. + intercept[ConcurrentAppendException] { + commitTxn1() + } + } + } + } + } + } + + // Safety floor: with the exclusion enabled, a winner that adds genuinely new data + // (dataChange = true) to a partition the loser read must STILL raise a + // ConcurrentAppendException. The fix only drops dataChange = false rearrangements, never + // real appends -- so it can never mask a genuine conflict, regardless of which operation + // produced the concurrent commit. + test("dataChange = true append still conflicts with a concurrent partition read, " + + "excludeNoDataChangeAddedFiles = true") { + withSQLConf( + DeltaSQLConf.DELTA_CONFLICT_DETECTION_EXCLUDE_NO_DATA_CHANGE_ADDS.key -> "true") { + withTempDir { tempDir => + val partCol = "part" + val log = DeltaLog.forTable(spark, tempDir.getCanonicalPath) + + log.startTransaction().commit(Seq( + Metadata( + schemaString = new StructType() + .add(partCol, IntegerType) + .add("value", IntegerType).json, + partitionColumns = Seq(partCol)) + ), ManualUpdate) + log.startTransaction().commit( + Seq(AddFile("a", Map(partCol -> "0"), 1, 1, dataChange = true), + AddFile("b", Map(partCol -> "1"), 1, 1, dataChange = true)), + ManualUpdate) + + // txn1 (loser): a non-blind writer that reads partition 0. + val newData = Seq(AddFile("x", Map(partCol -> "0"), 1, 1, dataChange = true)) + val txn = log.startTransaction() + val addFiles = txn.filterFiles(newData) + + // txn2 (winner): appends genuinely new data (dataChange = true) to partition 0. + log.startTransaction().commit( + Seq(AddFile("z", Map(partCol -> "0"), 1, 1, dataChange = true)), ManualUpdate) + + // Even with the exclusion enabled, a real append must still conflict. + intercept[ConcurrentAppendException] { + txn.commit(addFiles.map(_.remove) ++ newData, ManualUpdate) + } + } + } + } + + // Unpartitioned whole-table read -- the Liquid Clustering / auto-compaction scenario. A + // clustered table has no partitions, so a read-then-append writer reads the whole table and + // a concurrent OPTIMIZE (whose compacted output is dataChange = false) collides with that + // read on every commit. This is the case the exclusion is most needed for. + for (excludeNoDataChange <- BOOLEAN_DOMAIN) { + test("dataChange = false append (OPTIMIZE-style) vs concurrent whole-table read, " + + s"excludeNoDataChangeAddedFiles = $excludeNoDataChange") { + withSQLConf( + DeltaSQLConf.DELTA_CONFLICT_DETECTION_EXCLUDE_NO_DATA_CHANGE_ADDS.key -> + excludeNoDataChange.toString) { + withTempDir { tempDir => + val log = DeltaLog.forTable(spark, tempDir.getCanonicalPath) + + // Unpartitioned table with two files. + log.startTransaction().commit(Seq( + Metadata( + schemaString = new StructType() + .add("id", IntegerType) + .add("value", IntegerType).json) + ), ManualUpdate) + log.startTransaction().commit( + Seq(AddFile("a", Map.empty, 1, 1, dataChange = true), + AddFile("b", Map.empty, 1, 1, dataChange = true)), + ManualUpdate) + + // txn1 (loser): an insert-only writer that reads the whole table (no removes) -- + // e.g. reads to dedup/aggregate, then appends. filterFiles() reads all files. + val txn = log.startTransaction() + txn.filterFiles() + val newData = Seq(AddFile("x", Map.empty, 1, 1, dataChange = true)) + + // txn2 (winner): OPTIMIZE compacts the two files into one, dataChange = false. + log.startTransaction().commit( + Seq(AddFile("y", Map.empty, 1, 1, dataChange = false)), ManualUpdate) + + def commitTxn1(): Unit = txn.commit(newData, ManualUpdate) + + if (excludeNoDataChange) { + // OPTIMIZE's dataChange = false output is not in the append check -> no conflict. + commitTxn1() + val files = log.update().allFiles.collect() + // 'a', 'b', 'y' (OPTIMIZE output) and 'x' (txn1) all present; nothing removed. + assert(files.map(_.path).toSet == Set("a", "b", "y", "x")) + } else { + intercept[ConcurrentAppendException] { + commitTxn1() + } + } + } + } + } + } + for (enableNormalization <- BOOLEAN_DOMAIN) { test("filterFiles for timestamp partitions with different string formats, " + s"enableNormalization = $enableNormalization") { From 32c940775737adb240a4fce70e00fdf74f1cf9aa Mon Sep 17 00:00:00 2001 From: Eunjin Song Date: Wed, 29 Jul 2026 20:12:10 -0700 Subject: [PATCH 2/3] [Spark] Add end-to-end OPTIMIZE conflict tests and tighten fix scope Validate the exclusion against a *realistic* OPTIMIZE that both removes the compacted inputs and adds the dataChange=false output (the earlier tests only modeled the added file): - full OPTIMIZE vs an insert-only reader that registered only a partition read predicate (filterFiles(newFiles) does not populate readFiles) -> the separate removed-files check cannot fire, so with the flag on the txn FULLY reconciles (not merely swapping ConcurrentAppendException for ConcurrentDeleteRead). This is the exact insert-only shape reported in #326 / #626 / #1305. - full OPTIMIZE vs a whole-table reader (readFiles populated) -> still raises ConcurrentDeleteReadException with the flag on, proving the added-files fix does not (and must not) silence a genuine read/remove overlap. Fully reconciling that case is the harder row-level-concurrency problem, out of scope here. Dropped the earlier unpartitioned 'whole-table read reconciles' test: its winner added a dataChange=false file without removing anything, which overstated the fix (a real OPTIMIZE removes, and a whole-table reader then hits the delete-read check as shown above). Full OptimisticTransactionSuite 148/148 green. Co-Authored-By: Claude Opus 4.8 --- .../delta/OptimisticTransactionSuite.scala | 104 ++++++++++-------- 1 file changed, 61 insertions(+), 43 deletions(-) diff --git a/spark/src/test/scala/org/apache/spark/sql/delta/OptimisticTransactionSuite.scala b/spark/src/test/scala/org/apache/spark/sql/delta/OptimisticTransactionSuite.scala index 39182e0730a..d0d73ad9865 100644 --- a/spark/src/test/scala/org/apache/spark/sql/delta/OptimisticTransactionSuite.scala +++ b/spark/src/test/scala/org/apache/spark/sql/delta/OptimisticTransactionSuite.scala @@ -904,54 +904,72 @@ class OptimisticTransactionSuite } } - // Unpartitioned whole-table read -- the Liquid Clustering / auto-compaction scenario. A - // clustered table has no partitions, so a read-then-append writer reads the whole table and - // a concurrent OPTIMIZE (whose compacted output is dataChange = false) collides with that - // read on every commit. This is the case the exclusion is most needed for. - for (excludeNoDataChange <- BOOLEAN_DOMAIN) { - test("dataChange = false append (OPTIMIZE-style) vs concurrent whole-table read, " + - s"excludeNoDataChangeAddedFiles = $excludeNoDataChange") { - withSQLConf( - DeltaSQLConf.DELTA_CONFLICT_DETECTION_EXCLUDE_NO_DATA_CHANGE_ADDS.key -> - excludeNoDataChange.toString) { - withTempDir { tempDir => - val log = DeltaLog.forTable(spark, tempDir.getCanonicalPath) + // End-to-end with a *realistic* OPTIMIZE that both removes the compacted inputs AND adds the + // dataChange = false output. The loser is an insert-only writer that registered only a + // partition read PREDICATE (filterFiles(newFiles) does not populate readFiles), so the + // separate removed-files check cannot fire. With the flag on the fix FULLY reconciles -- it + // is not merely swapping ConcurrentAppendException for ConcurrentDeleteReadException. This is + // the exact shape of the insert-only conflict reported in issues #326 / #626 / PR #1305. + test("full OPTIMIZE (removes + dataChange=false adds) vs insert-only partition reader " + + "reconciles, excludeNoDataChangeAddedFiles = true") { + withSQLConf( + DeltaSQLConf.DELTA_CONFLICT_DETECTION_EXCLUDE_NO_DATA_CHANGE_ADDS.key -> "true") { + withTempDir { tempDir => + val partCol = "part" + val log = DeltaLog.forTable(spark, tempDir.getCanonicalPath) + log.startTransaction().commit(Seq(Metadata( + schemaString = new StructType() + .add(partCol, IntegerType).add("value", IntegerType).json, + partitionColumns = Seq(partCol))), ManualUpdate) + val fileA = AddFile("a", Map(partCol -> "0"), 1, 1, dataChange = true) + log.startTransaction().commit( + Seq(fileA, AddFile("b", Map(partCol -> "1"), 1, 1, dataChange = true)), ManualUpdate) - // Unpartitioned table with two files. - log.startTransaction().commit(Seq( - Metadata( - schemaString = new StructType() - .add("id", IntegerType) - .add("value", IntegerType).json) - ), ManualUpdate) - log.startTransaction().commit( - Seq(AddFile("a", Map.empty, 1, 1, dataChange = true), - AddFile("b", Map.empty, 1, 1, dataChange = true)), - ManualUpdate) + val newData = Seq(AddFile("x", Map(partCol -> "0"), 1, 1, dataChange = true)) + val txn = log.startTransaction() + txn.filterFiles(newData) // partition-0 read predicate only, no readFiles - // txn1 (loser): an insert-only writer that reads the whole table (no removes) -- - // e.g. reads to dedup/aggregate, then appends. filterFiles() reads all files. - val txn = log.startTransaction() - txn.filterFiles() - val newData = Seq(AddFile("x", Map.empty, 1, 1, dataChange = true)) + // realistic OPTIMIZE of partition 0: remove 'a', add 'y', all dataChange = false. + log.startTransaction().commit(Seq( + AddFile("y", Map(partCol -> "0"), 1, 1, dataChange = false), + fileA.removeWithTimestamp(dataChange = false)), ManualUpdate) - // txn2 (winner): OPTIMIZE compacts the two files into one, dataChange = false. - log.startTransaction().commit( - Seq(AddFile("y", Map.empty, 1, 1, dataChange = false)), ManualUpdate) + txn.commit(newData, ManualUpdate) // insert-only: append 'x', remove nothing + assert(log.update().allFiles.collect().map(_.path).toSet == Set("b", "y", "x")) + } + } + } - def commitTxn1(): Unit = txn.commit(newData, ManualUpdate) + // Scope boundary: the same realistic full OPTIMIZE vs a WHOLE-TABLE reader (filterFiles() + // populates readFiles = {a,b}). Here OPTIMIZE removes files the reader actually read, so the + // SEPARATE removed-files check (checkForDeletedFilesAgainstCurrentTxnReadFiles) fires with + // ConcurrentDeleteReadException. The added-files fix does NOT -- and must not -- silence a + // genuine read/remove overlap; fully reconciling that case is the harder row-level- + // concurrency problem and is out of scope for this change. + test("full OPTIMIZE (removes + dataChange=false adds) vs whole-table reader still raises " + + "ConcurrentDeleteReadException, excludeNoDataChangeAddedFiles = true") { + withSQLConf( + DeltaSQLConf.DELTA_CONFLICT_DETECTION_EXCLUDE_NO_DATA_CHANGE_ADDS.key -> "true") { + withTempDir { tempDir => + val log = DeltaLog.forTable(spark, tempDir.getCanonicalPath) + log.startTransaction().commit(Seq(Metadata( + schemaString = new StructType() + .add("id", IntegerType).add("value", IntegerType).json)), ManualUpdate) + val fileA = AddFile("a", Map.empty, 1, 1, dataChange = true) + val fileB = AddFile("b", Map.empty, 1, 1, dataChange = true) + log.startTransaction().commit(Seq(fileA, fileB), ManualUpdate) - if (excludeNoDataChange) { - // OPTIMIZE's dataChange = false output is not in the append check -> no conflict. - commitTxn1() - val files = log.update().allFiles.collect() - // 'a', 'b', 'y' (OPTIMIZE output) and 'x' (txn1) all present; nothing removed. - assert(files.map(_.path).toSet == Set("a", "b", "y", "x")) - } else { - intercept[ConcurrentAppendException] { - commitTxn1() - } - } + val txn = log.startTransaction() + txn.filterFiles() // whole-table read -> readFiles = {a, b} + val newData = Seq(AddFile("x", Map.empty, 1, 1, dataChange = true)) + + log.startTransaction().commit(Seq( + AddFile("y", Map.empty, 1, 1, dataChange = false), + fileA.removeWithTimestamp(dataChange = false), + fileB.removeWithTimestamp(dataChange = false)), ManualUpdate) + + intercept[ConcurrentDeleteReadException] { + txn.commit(newData, ManualUpdate) } } } From 4f801badadb99c3b8ed532c4d82a86d0aa5da122 Mon Sep 17 00:00:00 2001 From: Eunjin Song Date: Wed, 29 Jul 2026 20:35:43 -0700 Subject: [PATCH 3/3] Extend no-data-change exclusion to removed-files read check (insert-only) Broaden the exclusion so an append-only writer no longer raises a spurious ConcurrentDeleteReadException when a concurrent OPTIMIZE (dataChange = false) merely relocated the files it read. A dataChange = false RemoveFile deletes no logical rows -- every row survives under a new file boundary -- so it cannot invalidate a read done by a transaction that itself adds no RemoveFile and no deletion vector. The exclusion is gated on the current transaction being append-only: a DML loser (delete/update) reads files precisely to rewrite them, so a concurrent relocation of a read file must still conflict, else its RemoveFile/DV would target a file that no longer exists. Genuine deletes commit dataChange = true and are always kept, so real delete/read conflicts still fire. Rename the flag DELTA_CONFLICT_DETECTION_EXCLUDE_NO_DATA_CHANGE_ADDS -> DELTA_CONFLICT_DETECTION_EXCLUDE_NO_DATA_CHANGE_FILES (conflictDetection.excludeNoDataChangeFiles.enabled) to reflect that it now covers both added and removed files; still default-off. Tests (OptimisticTransactionSuite, 150/150): - whole-table APPEND-ONLY reader vs full OPTIMIZE -> reconciles (was: raised) - DML (delete) loser vs OPTIMIZE relocation -> still ConcurrentDeleteRead - genuine dataChange=true delete vs append-only reader -> still conflicts Co-Authored-By: Claude Opus 4.8 --- .../spark/sql/delta/ConflictChecker.scala | 30 ++++- .../sql/delta/sources/DeltaSQLConf.scala | 16 +-- .../delta/OptimisticTransactionSuite.scala | 104 +++++++++++++++--- 3 files changed, 124 insertions(+), 26 deletions(-) diff --git a/spark/src/main/scala/org/apache/spark/sql/delta/ConflictChecker.scala b/spark/src/main/scala/org/apache/spark/sql/delta/ConflictChecker.scala index de6f66dacb8..bb832666816 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 @@ -1145,7 +1145,7 @@ private[delta] class ConflictChecker( // OPTIMIZE. When enabled, drop them so only genuinely new data is checked. All // FileActions in a commit share one `dataChange` value (see `trackConsistentDataChange`). val addedFilesWithoutNoDataChange = - if (spark.conf.get(DeltaSQLConf.DELTA_CONFLICT_DETECTION_EXCLUDE_NO_DATA_CHANGE_ADDS)) { + if (spark.conf.get(DeltaSQLConf.DELTA_CONFLICT_DETECTION_EXCLUDE_NO_DATA_CHANGE_FILES)) { addedFilesToCheckForConflicts.filter(_.dataChange) } else { addedFilesToCheckForConflicts @@ -1170,10 +1170,34 @@ private[delta] class ConflictChecker( */ protected def checkForDeletedFilesAgainstCurrentTxnReadFiles(): Unit = { recordTime("checked-deletes") { + // A RemoveFile committed with `dataChange = false` (e.g. OPTIMIZE compaction / Z-ORDER) + // only relocates rows that still exist in the table under a different file boundary; it + // deletes no logical rows. So it cannot invalidate a read done by a purely append-only + // current transaction — every row the loser read is still present, just in a different + // file. When enabled we therefore drop `dataChange = false` removes from this check, but + // ONLY when the current transaction is itself append-only (adds no RemoveFile and no + // AddFile carrying a deletion vector). A transaction that deletes/updates rows (DML) reads + // files precisely to rewrite them; if its read file was concurrently relocated it must + // still conflict, otherwise its RemoveFile / DV would target a file that no longer exists + // and rows could be lost or resurrected. Genuine deletes commit `dataChange = true` and + // are always kept, so real delete/read conflicts still fire. + val currentTxnIsAppendOnly = currentTransactionInfo.actions.forall { + case _: RemoveFile => false + case a: AddFile => a.deletionVector == null + case _ => true + } + val relevantRemovedFiles = + if (spark.conf.get(DeltaSQLConf.DELTA_CONFLICT_DETECTION_EXCLUDE_NO_DATA_CHANGE_FILES) && + currentTxnIsAppendOnly) { + winningCommitSummary.removedFiles.filter(_.dataChange) + } else { + winningCommitSummary.removedFiles + } + // 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 + val deleteReadOverlap = relevantRemovedFiles .find(r => readFilePaths.contains(r.path)) if (deleteReadOverlap.nonEmpty) { val partitionOpt = getPrettyPartitionMessage(readFilePaths(deleteReadOverlap.get.path)) @@ -1183,7 +1207,7 @@ private[delta] class ConflictChecker( winningCommitVersion, partitionOpt) } - if (winningCommitSummary.removedFiles.nonEmpty && currentTransactionInfo.readWholeTable) { + if (relevantRemovedFiles.nonEmpty && currentTransactionInfo.readWholeTable) { throw DeltaErrors.concurrentDeleteReadException( winningCommitSummary.commitInfo, getTableNameOrPath, 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 14d57c5ed49..95be73dad7e 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 @@ -2240,13 +2240,15 @@ trait DeltaSQLConfBase extends DeltaSQLConfUtils { .booleanConf .createWithDefault(true) - val DELTA_CONFLICT_DETECTION_EXCLUDE_NO_DATA_CHANGE_ADDS = - buildConf("conflictDetection.excludeNoDataChangeAddedFiles.enabled") - .doc("When enabled, files added by a winning transaction with dataChange = false " + - "(e.g. OPTIMIZE compaction / Z-ORDER outputs) are excluded from the added-files " + - "conflict check. Such files only rearrange rows that already exist in the table and " + - "introduce no new logical rows, so a concurrent non-blind writer should not raise a " + - "ConcurrentAppendException against them. Off by default to preserve existing behavior.") + val DELTA_CONFLICT_DETECTION_EXCLUDE_NO_DATA_CHANGE_FILES = + buildConf("conflictDetection.excludeNoDataChangeFiles.enabled") + .doc("When enabled, AddFile/RemoveFile actions committed by a winning transaction with " + + "dataChange = false (e.g. OPTIMIZE compaction / Z-ORDER outputs) are excluded from the " + + "file-level conflict checks. Such files only rearrange rows that already exist in the " + + "table and add/delete no logical rows, so (1) a concurrent non-blind writer should not " + + "raise a ConcurrentAppendException against a no-data-change add, and (2) an append-only " + + "writer should not raise a ConcurrentDeleteReadException when a no-data-change remove " + + "merely relocated rows it read. Off by default to preserve existing behavior.") .internal() .booleanConf .createWithDefault(false) diff --git a/spark/src/test/scala/org/apache/spark/sql/delta/OptimisticTransactionSuite.scala b/spark/src/test/scala/org/apache/spark/sql/delta/OptimisticTransactionSuite.scala index d0d73ad9865..3126420b8e2 100644 --- a/spark/src/test/scala/org/apache/spark/sql/delta/OptimisticTransactionSuite.scala +++ b/spark/src/test/scala/org/apache/spark/sql/delta/OptimisticTransactionSuite.scala @@ -807,12 +807,12 @@ class OptimisticTransactionSuite // Case: an OPTIMIZE-style winner that only rearranges existing rows (dataChange = false, // isBlindAppend unset -> its adds fall into `changedDataAddedFiles`) must not raise a // spurious ConcurrentAppendException against a concurrent non-blind writer that reads the - // same partition. Gated by conflictDetection.excludeNoDataChangeAddedFiles.enabled. + // same partition. Gated by conflictDetection.excludeNoDataChangeFiles.enabled. for (excludeNoDataChange <- BOOLEAN_DOMAIN) { test("dataChange = false append (OPTIMIZE-style) vs concurrent partition read, " + - s"excludeNoDataChangeAddedFiles = $excludeNoDataChange") { + s"excludeNoDataChangeFiles = $excludeNoDataChange") { withSQLConf( - DeltaSQLConf.DELTA_CONFLICT_DETECTION_EXCLUDE_NO_DATA_CHANGE_ADDS.key -> + DeltaSQLConf.DELTA_CONFLICT_DETECTION_EXCLUDE_NO_DATA_CHANGE_FILES.key -> excludeNoDataChange.toString) { withTempDir { tempDir => val partCol = "part" @@ -868,9 +868,9 @@ class OptimisticTransactionSuite // real appends -- so it can never mask a genuine conflict, regardless of which operation // produced the concurrent commit. test("dataChange = true append still conflicts with a concurrent partition read, " + - "excludeNoDataChangeAddedFiles = true") { + "excludeNoDataChangeFiles = true") { withSQLConf( - DeltaSQLConf.DELTA_CONFLICT_DETECTION_EXCLUDE_NO_DATA_CHANGE_ADDS.key -> "true") { + DeltaSQLConf.DELTA_CONFLICT_DETECTION_EXCLUDE_NO_DATA_CHANGE_FILES.key -> "true") { withTempDir { tempDir => val partCol = "part" val log = DeltaLog.forTable(spark, tempDir.getCanonicalPath) @@ -911,9 +911,9 @@ class OptimisticTransactionSuite // is not merely swapping ConcurrentAppendException for ConcurrentDeleteReadException. This is // the exact shape of the insert-only conflict reported in issues #326 / #626 / PR #1305. test("full OPTIMIZE (removes + dataChange=false adds) vs insert-only partition reader " + - "reconciles, excludeNoDataChangeAddedFiles = true") { + "reconciles, excludeNoDataChangeFiles = true") { withSQLConf( - DeltaSQLConf.DELTA_CONFLICT_DETECTION_EXCLUDE_NO_DATA_CHANGE_ADDS.key -> "true") { + DeltaSQLConf.DELTA_CONFLICT_DETECTION_EXCLUDE_NO_DATA_CHANGE_FILES.key -> "true") { withTempDir { tempDir => val partCol = "part" val log = DeltaLog.forTable(spark, tempDir.getCanonicalPath) @@ -940,16 +940,17 @@ class OptimisticTransactionSuite } } - // Scope boundary: the same realistic full OPTIMIZE vs a WHOLE-TABLE reader (filterFiles() - // populates readFiles = {a,b}). Here OPTIMIZE removes files the reader actually read, so the - // SEPARATE removed-files check (checkForDeletedFilesAgainstCurrentTxnReadFiles) fires with - // ConcurrentDeleteReadException. The added-files fix does NOT -- and must not -- silence a - // genuine read/remove overlap; fully reconciling that case is the harder row-level- - // concurrency problem and is out of scope for this change. - test("full OPTIMIZE (removes + dataChange=false adds) vs whole-table reader still raises " + - "ConcurrentDeleteReadException, excludeNoDataChangeAddedFiles = true") { + // Full insert-only story: the same realistic full OPTIMIZE vs a WHOLE-TABLE reader + // (filterFiles() populates readFiles = {a,b}) that is itself APPEND-ONLY. OPTIMIZE relocates + // the files the reader read, but with dataChange = false -- every row is still present under a + // new file boundary -- so an append-only writer read nothing that was logically deleted. With + // the flag on, the removed-files check (checkForDeletedFilesAgainstCurrentTxnReadFiles) drops + // those dataChange = false removes and the writer FULLY reconciles. This closes the whole- + // table insert-only case, not just the predicate-only one above. + test("full OPTIMIZE (removes + dataChange=false adds) vs whole-table APPEND-ONLY reader " + + "reconciles, excludeNoDataChangeFiles = true") { withSQLConf( - DeltaSQLConf.DELTA_CONFLICT_DETECTION_EXCLUDE_NO_DATA_CHANGE_ADDS.key -> "true") { + DeltaSQLConf.DELTA_CONFLICT_DETECTION_EXCLUDE_NO_DATA_CHANGE_FILES.key -> "true") { withTempDir { tempDir => val log = DeltaLog.forTable(spark, tempDir.getCanonicalPath) log.startTransaction().commit(Seq(Metadata( @@ -968,6 +969,77 @@ class OptimisticTransactionSuite fileA.removeWithTimestamp(dataChange = false), fileB.removeWithTimestamp(dataChange = false)), ManualUpdate) + // Append-only writer: removes nothing, adds no DV -> reconciles despite the whole-table + // read overlapping OPTIMIZE's dataChange = false removes. + txn.commit(newData, ManualUpdate) + assert(log.update().allFiles.collect().map(_.path).toSet == Set("y", "x")) + } + } + } + + // Boundary: the append-only exclusion must NOT extend to a DML (delete/update) loser. A writer + // that reads a file precisely to rewrite it commits a RemoveFile (or a DV) against that path; + // if the file was concurrently relocated its RemoveFile now targets a file that no longer + // exists, which would lose or resurrect rows. So a non-append-only loser overlapping an + // OPTIMIZE remove must STILL raise ConcurrentDeleteReadException even with the flag on. This + // is exactly why the exclusion is gated on the current transaction being append-only. + test("full OPTIMIZE (dataChange=false removes) vs concurrent DELETE loser still raises " + + "ConcurrentDeleteReadException, excludeNoDataChangeFiles = true") { + withSQLConf( + DeltaSQLConf.DELTA_CONFLICT_DETECTION_EXCLUDE_NO_DATA_CHANGE_FILES.key -> "true") { + withTempDir { tempDir => + val log = DeltaLog.forTable(spark, tempDir.getCanonicalPath) + log.startTransaction().commit(Seq(Metadata( + schemaString = new StructType() + .add("id", IntegerType).add("value", IntegerType).json)), ManualUpdate) + val fileA = AddFile("a", Map.empty, 1, 1, dataChange = true) + val fileB = AddFile("b", Map.empty, 1, 1, dataChange = true) + log.startTransaction().commit(Seq(fileA, fileB), ManualUpdate) + + val txn = log.startTransaction() + txn.filterFiles() // whole-table read -> readFiles = {a, b} + + // OPTIMIZE relocates 'a' only (dataChange = false); leaves 'b' in place. + log.startTransaction().commit(Seq( + AddFile("y", Map.empty, 1, 1, dataChange = false), + fileA.removeWithTimestamp(dataChange = false)), ManualUpdate) + + // DML loser: genuinely deletes 'b' (dataChange = true remove) -> not append-only. The + // exclusion does not apply, so its read of the relocated 'a' still fails. 'b' is disjoint + // from what OPTIMIZE removed, so this exercises the read/remove path, not delete/delete. + intercept[ConcurrentDeleteReadException] { + txn.commit(Seq(fileB.remove), ManualUpdate) + } + } + } + } + + // Safety floor for removes: a winner that genuinely DELETES rows commits its RemoveFile with + // dataChange = true. Those are never excluded (`filter(_.dataChange)` keeps them), so an + // append-only reader whose read file was really deleted must STILL raise + // ConcurrentDeleteReadException even with the flag on. The exclusion silences only relocations + // (dataChange = false), never real deletions. + test("genuine delete (dataChange=true remove) vs append-only reader still raises " + + "ConcurrentDeleteReadException, excludeNoDataChangeFiles = true") { + withSQLConf( + DeltaSQLConf.DELTA_CONFLICT_DETECTION_EXCLUDE_NO_DATA_CHANGE_FILES.key -> "true") { + withTempDir { tempDir => + val log = DeltaLog.forTable(spark, tempDir.getCanonicalPath) + log.startTransaction().commit(Seq(Metadata( + schemaString = new StructType() + .add("id", IntegerType).add("value", IntegerType).json)), ManualUpdate) + val fileA = AddFile("a", Map.empty, 1, 1, dataChange = true) + val fileB = AddFile("b", Map.empty, 1, 1, dataChange = true) + log.startTransaction().commit(Seq(fileA, fileB), ManualUpdate) + + val txn = log.startTransaction() + txn.filterFiles() // whole-table read -> readFiles = {a, b} + val newData = Seq(AddFile("x", Map.empty, 1, 1, dataChange = true)) + + // Winner genuinely deletes 'a' (dataChange = true remove), not a relocation. + log.startTransaction().commit(Seq(fileA.remove), ManualUpdate) + + // Append-only reader, but the removed file was really deleted -> must still conflict. intercept[ConcurrentDeleteReadException] { txn.commit(newData, ManualUpdate) }