From 6a7985dcc5087049413f1c393e560f7dce2be452 Mon Sep 17 00:00:00 2001 From: Eunjin Song Date: Mon, 3 Aug 2026 23:35:51 -0700 Subject: [PATCH 1/2] [Spark] Reconcile OPTIMIZE vs concurrent DML via lazy offset DV remap When a compaction OPTIMIZE loses a commit race to a concurrent row-level DML (DELETE/UPDATE), remap the winner's deletion vector onto the OPTIMIZE output using the persisted source composition, instead of aborting the OPTIMIZE. - ConflictChecker.resolveOptimizeConflicts reads each removed source's compactedInto / compactionInfo tombstone tag (via parseOptimizeSourceComposition) to recover (outputPath, outputStart, liveCount), then rebases the winning DML's new deletions onto the compacted output by contiguous offset + live rank. O(1) per source. - Conservative and safe-by-abort: only a pure compaction, only single-run contiguous placements, output bound checked; any absent or malformed tag or unmodeled shape falls back to today's abort, never a wrong result. - Gated by optimize.conflictReconciliation.enabled (default off). Builds on the capture/persist layer (compactedInto / compactionInfo tags, CompactionInfoEntry). Co-Authored-By: Claude Opus 4.8 --- .../spark/sql/delta/ConflictChecker.scala | 55 +- .../OptimizeConflictReconciliation.scala | 203 ++++++++ .../OptimizeConflictReconciliationSuite.scala | 489 ++++++++++++++++++ 3 files changed, 746 insertions(+), 1 deletion(-) create mode 100644 spark/src/main/scala/org/apache/spark/sql/delta/OptimizeConflictReconciliation.scala create mode 100644 spark/src/test/scala/org/apache/spark/sql/delta/OptimizeConflictReconciliationSuite.scala diff --git a/spark/src/main/scala/org/apache/spark/sql/delta/ConflictChecker.scala b/spark/src/main/scala/org/apache/spark/sql/delta/ConflictChecker.scala index 9933b68715e..8c2695d3819 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 @@ -20,6 +20,7 @@ package org.apache.spark.sql.delta import java.util.concurrent.TimeUnit import scala.collection.mutable +import scala.util.Try import org.apache.spark.sql.delta.DeltaOperations.{OP_SET_TBLPROPERTIES, ROW_TRACKING_BACKFILL_OPERATION_NAME, ROW_TRACKING_UNBACKFILL_OPERATION_NAME} import org.apache.spark.sql.delta.RowId.RowTrackingMetadataDomain @@ -31,6 +32,7 @@ import org.apache.spark.sql.delta.sources.DeltaSourceUtils import org.apache.spark.sql.delta.sources.DeltaSQLConf import org.apache.spark.sql.delta.util.DeltaSparkPlanUtils.CheckDeterministicOptions import org.apache.spark.sql.delta.util.FileNames +import org.apache.spark.sql.delta.util.JsonUtils import io.delta.storage.commit.UpdatedActions import io.delta.storage.commit.uccommitcoordinator.UCCommitCoordinatorClient import io.delta.storage.commit.uniform.UniformMetadata @@ -78,6 +80,9 @@ private[delta] case class CurrentTransactionInfo( * * TODO: We might want to cluster all non-file actions at the front, for similar reasons. */ + // Nothing to strip: the OPTIMIZE `compactedInto` / `compactionInfo` composition tags on a removed + // source's tombstone are persisted (as Databricks Runtime persists them), so a concurrent DML + // that LOSES to this OPTIMIZE can read the composition from the committed tombstone. lazy val finalActionsToCommit: Seq[Action] = commitInfo ++: actions private var newMetadata: Option[Metadata] = None @@ -221,7 +226,8 @@ private[delta] class ConflictChecker( protected val winningCommitSummary: WinningCommitSummary, isolationLevel: IsolationLevel) extends DeltaLogging with ConflictCheckerPredicateElimination - with RowLevelConcurrencyResolution { + with RowLevelConcurrencyResolution + with OptimizeConflictReconciliation { protected val winningCommitVersion = winningCommitSummary.commitVersion protected val startTimeMs = System.currentTimeMillis() @@ -307,6 +313,11 @@ private[delta] class ConflictChecker( // base row IDs. resolveRowLevelConflicts() + // Compaction OPTIMIZE vs concurrent row-level DML: remap the concurrent deletion vector from + // each removed source file onto the compacted output file (offset arithmetic) instead of + // aborting. Compaction only; reclustering / already-DV'd sources / missing composition abort. + resolveOptimizeConflicts() + // Data file checks. checkForAddedFilesThatShouldHaveBeenReadByCurrentTxn() checkForDeletedFilesAgainstCurrentTxnReadFiles() @@ -1550,6 +1561,48 @@ private[delta] class ConflictChecker( } private[delta] object ConflictChecker extends DeltaLogging { + + /** + * Parse the composition a compaction OPTIMIZE recorded on a removed source `r`'s tombstone, + * normalized to `(outputPath, outputStart, liveCount)`: `r`'s `liveCount` live rows landed + * contiguously at output positions `[outputStart, outputStart + liveCount)` of `outputPath`, in + * physical order. Used by [[ConflictChecker.resolveOptimizeConflicts]] to remap a concurrent + * deletion vector onto the output. + * + * Reads the `compactedInto` / `compactionInfo` tags (see [[RemoveFile.Tags.COMPACTION_INFO]]), a + * format modeled on Databricks Runtime's (cross-engine reconciliation is best-effort, not a + * verified guarantee). `sourceNumPhysicalRecords` is a PHYSICAL count; the live count is derived + * here as + * `physical - |Do|`, where `|Do|` (the OPTIMIZE-read DV) is the cardinality carried on this same + * tombstone (`r.deletionVector`). + * + * Returns None -- so the caller falls back to today's abort, never a wrong result -- when the + * tags are absent, malformed, point at an output not in `outputPaths`, or describe a source split + * across more than one output run, which the contiguous offset remap does not model. + */ + private[delta] def parseOptimizeSourceComposition( + r: RemoveFile, + outputPaths: Set[String]): Option[(String, Long, Long)] = Try { + (r.getTag(RemoveFile.Tags.COMPACTED_INTO), r.getTag(RemoveFile.Tags.COMPACTION_INFO)) match { + case (Some(intoJson), Some(infoJson)) => + val outputs = JsonUtils.fromJson[Seq[String]](intoJson) + val entries = JsonUtils.fromJson[Seq[CompactionInfoEntry]](infoJson) + (outputs, entries) match { + case (Seq(outputPath), Seq(entry)) + if outputPaths.contains(outputPath) && + entry.rowOffsetInTarget.isDefined && entry.sourceNumPhysicalRecords.isDefined => + // Physical count -> live count via the read-time DV cardinality carried on this + // tombstone. Single-run only (one output, one entry); anything else -> None. + val readTimeDvCardinality = + Option(r.deletionVector).map(_.cardinality).getOrElse(0L) + Some((outputPath, entry.rowOffsetInTarget.get, + entry.sourceNumPhysicalRecords.get - readTimeDvCardinality)) + case _ => None + } + case _ => None + } + }.toOption.flatten + /** * Returns an iterator that validates all [[AddFile]] and [[RemoveFile]] actions in * `actions` share a consistent `dataChange` value. [[AddCDCFile]] is excluded because diff --git a/spark/src/main/scala/org/apache/spark/sql/delta/OptimizeConflictReconciliation.scala b/spark/src/main/scala/org/apache/spark/sql/delta/OptimizeConflictReconciliation.scala new file mode 100644 index 00000000000..5e90df82320 --- /dev/null +++ b/spark/src/main/scala/org/apache/spark/sql/delta/OptimizeConflictReconciliation.scala @@ -0,0 +1,203 @@ +/* + * Copyright (2021) The Delta Lake Project Authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.spark.sql.delta + +import scala.collection.mutable + +import org.apache.spark.sql.delta.actions.{AddFile, RemoveFile} +import org.apache.spark.sql.delta.commands.DeletionVectorUtils +import org.apache.spark.sql.delta.deletionvectors.RoaringBitmapArray +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 + +/** + * OPTIMIZE-vs-concurrent-DML reconciliation for the [[ConflictChecker]]: when a compaction OPTIMIZE + * and a concurrent row-level DELETE/UPDATE touch the same source file, remap the DML's deletion + * vector onto the compacted output (offset arithmetic) instead of aborting. + * + * Mixed into [[ConflictChecker]] as a self-typed trait alongside [[RowLevelConcurrencyResolution]], + * whose `readDeletionVectorOrEmpty` / `writeMergedDeletionVector` / `rowLevelResolvedPaths` / + * `winningOperationName` members it reuses (both traits share the same `ConflictChecker` + * self-type). Lives in its own file to keep ConflictChecker focused on file-level conflict + * detection. The static composition parser stays in `object ConflictChecker` as + * [[ConflictChecker.parseOptimizeSourceComposition]]. + */ +trait OptimizeConflictReconciliation extends DeltaLogging { self: ConflictChecker => + + private lazy val optimizeReconciliationEnabled: Boolean = + spark.conf.get(DeltaSQLConf.DELTA_OPTIMIZE_CONFLICT_RECONCILIATION_ENABLED) && + DeletionVectorUtils.deletionVectorsWritable( + currentTransactionInfo.protocol, currentTransactionInfo.metadata) + + /** + * Compaction OPTIMIZE vs a concurrent row-level DML. An OPTIMIZE that removed a source file `F` + * and compacted it into an output `O` conflicts with a concurrent DELETE/UPDATE that added a + * deletion vector to `F`. Instead of aborting, remap the concurrent DV onto `O`: the compaction + * write recorded, on each removed source's tombstone, that the source's live rows landed at + * output positions `[outputStart, outputStart + liveCount)` in physical order (the + * `compactedInto` / `compactionInfo` composition tags; see [[RemoveFile.Tags.COMPACTION_INFO]]). + * A deleted physical row `i` in `F` lands at `outputStart + liveRank(i)`, where `liveRank(i)` is + * `i` minus the rows already deleted below it at read time; that mapped position is unioned into + * `O`'s deletion vector. + * + * The read-time gaps are reconstructed here (from `F`'s read-time DV) rather than encoded in the + * tag at write time, so a fragmented source DV cannot bloat the tag. Conservative by design: only + * for tagged (compaction) sources, only if every newly-deleted row maps within the source's live + * run, and only if every conflicting source is remappable; otherwise fall through to the standard + * checks (abort). Only the winner's incremental (post-read) deletions are remapped. Handles the + * direction where OPTIMIZE is the current (losing) transaction, reading the composition from its + * own in-memory tombstones. + */ + protected def resolveOptimizeConflicts(): Unit = { + if (!optimizeReconciliationEnabled) return + + // Winning transaction's DV updates (a concurrent DELETE/UPDATE added a DV to a file). + 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 + + // This OPTIMIZE's removed sources (some tagged with their composition) and its outputs by path. + val currentRemoveByPath = currentTransactionInfo.actions.collect { + case r: RemoveFile => r.path -> r + }.toMap + val currentAddByPath = currentTransactionInfo.actions.collect { + case a: AddFile => a.path -> a + }.toMap + + // source path -> (output, liveCount, outputStart): the source's `liveCount` live rows landed + // contiguously at output positions [outputStart, outputStart + liveCount) in physical order, + // read from the source tombstone's composition tags via the shared parser (the output AddFile + // is resolved by the recorded path). + val outputPaths = currentAddByPath.keySet + val srcToRun = mutable.Map.empty[String, (AddFile, Long, Long)] + for ((src, r) <- currentRemoveByPath) { + ConflictChecker.parseOptimizeSourceComposition(r, outputPaths).foreach { + case (outputPath, outputStart, liveCount) => + currentAddByPath.get(outputPath).foreach { out => + srcToRun(src) = (out, liveCount, outputStart) + } + } + } + if (srcToRun.isEmpty) return + + val sharedPaths = winningDvUpdates.keySet + .intersect(srcToRun.keySet) + .intersect(currentRemoveByPath.keySet) + if (sharedPaths.isEmpty) return + + recordTime("resolved-optimize-conflicts") { + val dvStore = DeletionVectorStore.createInstance(deltaLog.newDeltaHadoopConf()) + val tablePath = deltaLog.dataPath + + // Accumulate the remapped DV per output file, starting from its existing DV. + val outputDv = mutable.Map.empty[String, RoaringBitmapArray] + val resolvedSources = mutable.Set.empty[String] + var allResolvable = true + + for (src <- sharedPaths if allResolvable) { + // `F`'s deletion vector when OPTIMIZE read it. If non-empty, those rows were already gone + // from the compacted output, so the recorded run covers only `F`'s live rows and a deleted + // physical row's output offset must be discounted by the read-time deletions below it. + val readTimeDv = + readDeletionVectorOrEmpty(dvStore, currentRemoveByPath(src).deletionVector, tablePath) + val winnerDv = + readDeletionVectorOrEmpty(dvStore, winningDvUpdates(src).deletionVector, tablePath) + // Sorted ascending, so the count of read-time deletions below a physical row is a binary + // search. This is the read-time gap reconstruction deferred from write time to here. + val readTimeDeleted = readTimeDv.toArray + val (out, liveCount, outputStart) = srcToRun(src) + // Defense-in-depth against a corrupt or foreign composition tag: the source's live run must + // fit within the output file's physical record count. An out-of-range range is impossible + // for a real capture, so treat it as unreconcilable (abort) rather than remap out of + // bounds. Cheap: reads the already-parsed stats; a no-op on the happy path. + if (out.numPhysicalRecords.exists(outputStart + liveCount > _)) { + allResolvable = false + } + winnerDv.forEach { i => + // The winner's DV is cumulative. Rows already deleted at read time are not in the output, + // so remap only the winner's NEW deletions; each lands among the source's live rows. + if (!readTimeDv.contains(i)) { + // Live-rank of physical row `i` = i minus the read-time deletions below it; that is + // its offset within the source's contiguous live-row run in the compacted output. + val ins = java.util.Arrays.binarySearch(readTimeDeleted, i) + val liveRank = i - (if (ins < 0) -(ins + 1) else ins) + if (liveRank >= 0 && liveRank < liveCount) { + val acc = outputDv.getOrElseUpdate(out.path, + readDeletionVectorOrEmpty(dvStore, out.deletionVector, tablePath).copy()) + acc.add(outputStart + liveRank) + } else { + // A newly-deleted physical row does not map within the source's live run -> cannot + // remap. + allResolvable = false + } + } + } + if (allResolvable) resolvedSources += src + } + + // Reconcile only if every conflicting source was remappable; a partial remap could leave an + // un-reconciled conflict, so otherwise leave everything to the standard checks (abort). + if (allResolvable && outputDv.nonEmpty) { + // Each compacted output gets the remapped/unioned DV. + val addReplacements: Map[String, AddFile] = outputDv.keys.iterator + .map { p => + // Writes a new DV file as a side effect of conflict resolution. If this commit later + // aborts or retries against another winner, the file is unreferenced and reclaimed by + // VACUUM (same lifecycle as any DV the DML write path persists). See the note on + // `writeMergedDeletionVector`. + val desc = writeMergedDeletionVector(dvStore, tablePath, outputDv(p)) + p -> currentAddByPath(p).copy(deletionVector = desc).withoutTightBoundStats + }.toMap + // Re-point each resolved source's RemoveFile at the winner's post-image (path + winning + // DV). The OPTIMIZE read the pre-winner version, so its (path, no-DV) RemoveFile would not + // match the winner's now-live (path, DV) file and would leave it undeleted (files are + // identified by path AND deletion vector). + val removeReplacements: Map[String, RemoveFile] = + resolvedSources.iterator.map(src => + // dataChange = false: an OPTIMIZE commit is a data-preserving relocation, so the + // reconciled source tombstone must match the dataChange=false OPTIMIZE output (keeps + // streaming/CDC transparent and avoids mixing dataChange values in one commit). + src -> winningDvUpdates(src).removeWithTimestamp(dataChange = false)).toMap + + val newActions = currentTransactionInfo.actions.map { + case a: AddFile if addReplacements.contains(a.path) => addReplacements(a.path) + case r: RemoveFile if removeReplacements.contains(r.path) => removeReplacements(r.path) + case other => other + } + // The reconciled sources are no longer read/delete conflicts for the standard checks. + resolvedSources.foreach(rowLevelResolvedPaths += _) + val newReadFiles = currentTransactionInfo.readFiles + .filterNot(f => resolvedSources.contains(f.path)) + currentTransactionInfo = + currentTransactionInfo.copy(actions = newActions, readFiles = newReadFiles) + + recordDeltaEvent( + deltaLog, + opType = "delta.optimize.conflictReconciliation.remapped", + data = Map( + "winningCommitVersion" -> winningCommitVersion, + "resolvedSources" -> resolvedSources.size, + "outputsRemapped" -> addReplacements.size, + "winningOperation" -> winningOperationName.getOrElse("UNKNOWN"))) + } + } + } +} diff --git a/spark/src/test/scala/org/apache/spark/sql/delta/OptimizeConflictReconciliationSuite.scala b/spark/src/test/scala/org/apache/spark/sql/delta/OptimizeConflictReconciliationSuite.scala new file mode 100644 index 00000000000..8d3b80c5a54 --- /dev/null +++ b/spark/src/test/scala/org/apache/spark/sql/delta/OptimizeConflictReconciliationSuite.scala @@ -0,0 +1,489 @@ +/* + * 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.actions.{DeletionVectorDescriptor, RemoveFile} +import org.apache.spark.sql.delta.concurrency.{PhaseLockingTestMixin, TransactionExecutionTestMixin} +import org.apache.spark.sql.delta.files.{DeltaFileFormatWriter, SourceCompositionCaptureExec} +import org.apache.spark.sql.delta.fuzzer.{OptimisticTransactionPhases, PhaseLockingTransactionExecutionObserver} +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.execution.SortExec +import org.apache.spark.sql.functions.lit +import org.apache.spark.sql.test.SharedSparkSession +import org.apache.spark.util.ThreadUtils + +/** + * Tests for compaction OPTIMIZE vs concurrent row-level DML reconciliation + * (spark.databricks.delta.optimize.conflictReconciliation.enabled): a compaction OPTIMIZE that + * loses to a concurrent DELETE/UPDATE remaps the concurrent deletion vector onto the compacted + * output (offset arithmetic, with read-time DV gaps reconstructed lazily) instead of aborting. + * Also covers the safety fallbacks that must NOT reconcile -- reclustering (ZORDER), the + * repartition path, and the reverse (DELETE-loser) direction -- which abort as they do today. + */ +class OptimizeConflictReconciliationSuite extends QueryTest + with SharedSparkSession + with DeltaSQLCommandTest + with PhaseLockingTestMixin + with TransactionExecutionTestMixin { + + // Deletion vectors on for every table in 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}`" + + /** Multi-file table: id in [0, n) across `files` data files, deletion vectors enabled. */ + private def createMultiFileTable(dir: File, n: Int = 300, files: Int = 3): DeltaLog = { + spark.range(start = 0, end = n, step = 1, numPartitions = files) + .write.format("delta").mode("append").save(dir.getAbsolutePath) + val log = DeltaLog.forTable(spark, dir.getCanonicalPath) + assert(log.update().allFiles.collect().length === files, + s"test table must have $files data files") + log + } + + /** + * Runs `sqlText` with the OPTIMIZE conflict-reconciliation flag set to `reconcile`, plus any + * `extraConf` overrides (e.g. forcing the repartition path or a small max file size). + */ + private def sqlTxn( + sqlText: String, + reconcile: Boolean, + extraConf: Seq[(String, String)] = Nil): () => Array[Row] = + () => { + val confs = (DeltaSQLConf.DELTA_OPTIMIZE_CONFLICT_RECONCILIATION_ENABLED.key -> + reconcile.toString) +: extraConf + withSQLConf(confs: _*) { + sql(sqlText).collect() + } + Array.empty[Row] + } + + /** + * Runs `optimizeSql` as the loser while a concurrent `DELETE id = deleteId` wins (commits during + * the OPTIMIZE's first attempt), and returns the exception the OPTIMIZE ultimately fails with. + * + * OPTIMIZE auto-resolves and RETRIES on abort, so the single-observer `runTxnsWithOrder` helpers + * cannot drive it -- the retry's second commit is an unexpected phase transition. This mirrors + * upstream `OptimizeConflictSuite`: give the retry its own [[PhaseLockingTransactionExecutionObserver]] + * chained via `setNextObserver`, then let the abort surface. + */ + private def runOptimizeLoserExpectingAbort( + dir: File, + optimizeSql: String, + deleteId: Long, + extraConf: Seq[(String, String)] = Nil): SparkException = { + val optimizeFn = sqlTxn(optimizeSql, reconcile = true, extraConf) + val Seq(future) = runFunctionsWithOrderingFromObserver(Seq(optimizeFn)) { + case (optimizeObserver :: Nil) => + val retryObserver = new PhaseLockingTransactionExecutionObserver( + OptimisticTransactionPhases.forName("test-replacement-txn")) + optimizeObserver.setNextObserver(retryObserver, autoAdvance = true) + unblockUntilPreCommit(optimizeObserver) + busyWaitFor(optimizeObserver.phases.preparePhase.hasEntered, timeout) + // Winner commits during OPTIMIZE's first attempt, then OPTIMIZE aborts and retries. + sql(s"DELETE FROM ${tableRef(dir)} WHERE id = $deleteId").collect() + unblockCommit(optimizeObserver) + busyWaitFor(optimizeObserver.phases.commitPhase.hasLeft, timeout) + optimizeObserver.phases.postCommitPhase.exitBarrier.unblock() + unblockAllPhases(retryObserver) + } + intercept[SparkException] { ThreadUtils.awaitResult(future, timeout) } + } + + /** Asserts an awaited transaction future failed with a Delta concurrency conflict (an abort). */ + 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 + + private def deletionVectorCardinalities(log: DeltaLog): Seq[Long] = + log.update().allFiles.collect() + .filter(_.deletionVector != null) + .map(_.deletionVector.cardinality) + .toSeq + + /** + * Partitioned (single partition value, `files` data files) variant of [[createMultiFileTable]]. + * A single distinct partition value keeps OPTIMIZE to one compaction bin, so the write is + * partitioned but still a single output -- the case F1 (sort-above-capture) applies to. + */ + private def createPartitionedMultiFileTable(dir: File, n: Int = 300, files: Int = 3): DeltaLog = { + spark.range(start = 0, end = n, step = 1, numPartitions = files) + .withColumn("part", lit(0)) + .write.format("delta").partitionBy("part").mode("append").save(dir.getAbsolutePath) + val log = DeltaLog.forTable(spark, dir.getCanonicalPath) + assert(log.update().allFiles.collect().length === files, + s"test table must have $files data files") + log + } + + /** RemoveFile actions in the table's latest commit (the reconciled OPTIMIZE commit here). */ + private def lastCommitRemoveFiles(log: DeltaLog): Seq[RemoveFile] = { + val version = log.update().version + log.getChanges(version).flatMap(_._2).collect { case r: RemoveFile => r }.toSeq + } + + test("compaction OPTIMIZE (loser) reconciles a concurrent DELETE by remapping the DV") { + withTempDir { dir => + val log = createMultiFileTable(dir) + // A (loser): OPTIMIZE compacts all files. B (winner): DELETE id=150 commits during A, adding + // a DV to the middle source file. A must remap B's DV onto the compacted output at commit. + val txnA = sqlTxn(s"OPTIMIZE ${tableRef(dir)}", reconcile = true) + val txnB = sqlTxn(s"DELETE FROM ${tableRef(dir)} WHERE id = 150", reconcile = true) + + val (futureA, futureB) = runTxnsWithOrder__A_Start__B__A_End(txnA, txnB) + ThreadUtils.awaitResult(futureB, Duration.Inf) + ThreadUtils.awaitResult(futureA, Duration.Inf) + + // Both committed, correct data: id=150 gone, everything else present. + assert(ids(dir) === (0L until 300L).filterNot(_ == 150L)) + // Compacted to one output file carrying the remapped deletion vector (cardinality 1), proving + // reconciliation (a plain re-compaction retry would leave no DV on the output). + val files = log.update().allFiles.collect() + assert(files.length === 1, s"expected a single compacted file, got ${files.length}") + assert(deletionVectorCardinalities(log) === Seq(1L), + "compacted output should carry the remapped deletion vector") + // F2: every RemoveFile in the reconciled OPTIMIZE commit must be dataChange=false. An OPTIMIZE + // is a data-preserving relocation; a dataChange=true reconciled tombstone would break + // streaming/CDC transparency and mix dataChange values within the one commit. + val removes = lastCommitRemoveFiles(log) + assert(removes.nonEmpty, "reconciled OPTIMIZE commit should contain source tombstones") + assert(removes.forall(!_.dataChange), + "reconciled OPTIMIZE tombstones must all be dataChange=false") + } + } + + test("compaction OPTIMIZE (loser) reconciles a DELETE when the source already has a DV") { + withTempDir { dir => + val log = createMultiFileTable(dir) + // Pre-existing (compaction-time) DV: delete id=100 (physical row 0 of the middle file). This + // commits before OPTIMIZE reads, so that file's live rows start at physical index 1 and the + // read-time gap must be reconstructed at conflict time. + sql(s"DELETE FROM ${tableRef(dir)} WHERE id = 100") + assert(deletionVectorCardinalities(log) === Seq(1L), "pre-existing DV expected on one file") + + // A (loser): OPTIMIZE compacts all files (purging the pre-existing DV). B (winner): DELETE + // id=150 commits during A, giving the middle file a cumulative DV {id100, id150}. A remaps + // only B's NEW deletion (id150) onto the output, discounting the read-time gap from id100. + val txnA = sqlTxn(s"OPTIMIZE ${tableRef(dir)}", reconcile = true) + val txnB = sqlTxn(s"DELETE FROM ${tableRef(dir)} WHERE id = 150", reconcile = true) + + val (futureA, futureB) = runTxnsWithOrder__A_Start__B__A_End(txnA, txnB) + ThreadUtils.awaitResult(futureB, Duration.Inf) + ThreadUtils.awaitResult(futureA, Duration.Inf) + + // Both committed, correct data: id=100 and id=150 gone, everything else present. + assert(ids(dir) === (0L until 300L).filterNot(x => x == 100L || x == 150L)) + // Compacted to one output file whose DV carries only the new id=150 (cardinality 1); id=100 + // was already excluded from the output, so it is not in the remapped DV. + val files = log.update().allFiles.collect() + assert(files.length === 1, s"expected a single compacted file, got ${files.length}") + assert(deletionVectorCardinalities(log) === Seq(1L), + "compacted output should carry the remapped DV for the new delete only") + } + } + + test("reconciliation remaps a delete to a non-DV file in a bin that also holds a DV'd file") { + withTempDir { dir => + val log = createMultiFileTable(dir) + // Pre-existing DV on the middle file (id=100) drops its live count to 99, which shifts the + // LAST file's output offset. A concurrent delete of id=250 (last file, no DV) then has to + // land at the shifted offset (199+50, not 200+50): catches an off-by-one in the cumulative + // offset if physical instead of live counts were used for the DV'd file. + sql(s"DELETE FROM ${tableRef(dir)} WHERE id = 100") + val txnA = sqlTxn(s"OPTIMIZE ${tableRef(dir)}", reconcile = true) + val txnB = sqlTxn(s"DELETE FROM ${tableRef(dir)} WHERE id = 250", reconcile = true) + + val (futureA, futureB) = runTxnsWithOrder__A_Start__B__A_End(txnA, txnB) + ThreadUtils.awaitResult(futureB, Duration.Inf) + ThreadUtils.awaitResult(futureA, Duration.Inf) + + assert(ids(dir) === (0L until 300L).filterNot(x => x == 100L || x == 250L)) + val files = log.update().allFiles.collect() + assert(files.length === 1, s"expected a single compacted file, got ${files.length}") + assert(deletionVectorCardinalities(log) === Seq(1L), + "compacted output DV should carry only the concurrent delete of id=250") + } + } + + test("reconciliation remaps across multiple read-time DV gaps in one source") { + withTempDir { dir => + val log = createMultiFileTable(dir) + // Fragment the middle file's read-time DV: delete physical rows 0, 1 and 5 (id 100, 101, 105) + // before OPTIMIZE reads. The output tag records only the file's 97 live rows (no per-gap + // segments), so remapping the winner's later delete must reconstruct all three gaps from the + // read-time DV and discount them from the physical offset. This is the fragmented-DV case the + // lazy reconstruction is built for: a delete at physical row 50 lands at live-rank 50-3=47. + sql(s"DELETE FROM ${tableRef(dir)} WHERE id IN (100, 101, 105)") + assert(deletionVectorCardinalities(log) === Seq(3L), "pre-existing 3-row DV expected") + + val txnA = sqlTxn(s"OPTIMIZE ${tableRef(dir)}", reconcile = true) + val txnB = sqlTxn(s"DELETE FROM ${tableRef(dir)} WHERE id = 150", reconcile = true) + + val (futureA, futureB) = runTxnsWithOrder__A_Start__B__A_End(txnA, txnB) + ThreadUtils.awaitResult(futureB, Duration.Inf) + ThreadUtils.awaitResult(futureA, Duration.Inf) + + // If the gap discount were wrong, a different physical row would be masked and a different id + // would go missing, so the exact surviving set validates the multi-gap rank arithmetic. + assert(ids(dir) === (0L until 300L).filterNot(x => Set(100L, 101L, 105L, 150L).contains(x))) + val files = log.update().allFiles.collect() + assert(files.length === 1, s"expected a single compacted file, got ${files.length}") + assert(deletionVectorCardinalities(log) === Seq(1L), + "compacted output DV should carry only the remapped new delete (id=150)") + } + } + + test("compaction OPTIMIZE (loser) reconciles a concurrent UPDATE by remapping the DV") { + withTempDir { dir => + val log = createMultiFileTable(dir) + // UPDATE with DVs masks the old row (a DV on the source file, which the loser remaps onto the + // compacted output) and appends the new value in a fresh image file. Same remap path as a + // DELETE, but exercising an UPDATE as the winner. + val txnA = sqlTxn(s"OPTIMIZE ${tableRef(dir)}", reconcile = true) + val txnB = sqlTxn(s"UPDATE ${tableRef(dir)} SET id = id + 1000 WHERE id = 150", + reconcile = true) + + val (futureA, futureB) = runTxnsWithOrder__A_Start__B__A_End(txnA, txnB) + ThreadUtils.awaitResult(futureB, Duration.Inf) + ThreadUtils.awaitResult(futureA, Duration.Inf) + + // Old id=150 masked, new id=1150 present, everything else intact. + assert(ids(dir) === ((0L until 300L).filterNot(_ == 150L) :+ 1150L).sorted) + // The compacted output carries the remapped DV (cardinality 1); the winner's fresh image + // file has none. + assert(deletionVectorCardinalities(log) === Seq(1L), + "compacted output should carry the remapped DV for the updated row") + } + } + + test("reconciliation targets the correct output among multiple compaction bins") { + withTempDir { dir => + val log = createMultiFileTable(dir, n = 400, files = 4) + // Force two 2-file bins (maxFileSize = the two largest files) so OPTIMIZE emits two compacted + // outputs. The remap must land the winner's DV on the output whose bin held the deleted row, + // computing that output's offsets independently of the other output. + val sizes = log.update().allFiles.collect().map(_.size).sorted.reverse + val maxFileSize = (sizes(0) + sizes(1)).toString + val txnA = sqlTxn(s"OPTIMIZE ${tableRef(dir)}", reconcile = true, + Seq(DeltaSQLConf.DELTA_OPTIMIZE_MAX_FILE_SIZE.key -> maxFileSize)) + val txnB = sqlTxn(s"DELETE FROM ${tableRef(dir)} WHERE id = 150", reconcile = true) + + val (futureA, futureB) = runTxnsWithOrder__A_Start__B__A_End(txnA, txnB) + ThreadUtils.awaitResult(futureB, Duration.Inf) + ThreadUtils.awaitResult(futureA, Duration.Inf) + + assert(ids(dir) === (0L until 400L).filterNot(_ == 150L)) + val files = log.update().allFiles.collect() + assert(files.length === 2, s"expected two compacted outputs, got ${files.length}") + // Exactly one of the two outputs carries the remapped DV (the bin that held id=150). + assert(deletionVectorCardinalities(log) === Seq(1L), + "only the bin that held the deleted row should carry a remapped DV") + } + } + + test("reclustering (ZORDER) OPTIMIZE loser is not reconciled and aborts") { + withTempDir { dir => + createMultiFileTable(dir) + // ZORDER permutes rows across files, so its output carries no composition tag: the offset + // remap cannot apply and the loser must abort. Reconciliation must never engage on a + // row-permuting rewrite, or it would mask the wrong physical rows. + val e = runOptimizeLoserExpectingAbort( + dir, s"OPTIMIZE ${tableRef(dir)} ZORDER BY (id)", deleteId = 150) + assertConcurrentModificationException(e) + // The winner's delete stands and nothing was corrupted by a bad remap. + assert(ids(dir) === (0L until 300L).filterNot(_ == 150L)) + } + } + + test("repartition-path OPTIMIZE loser is not reconciled and aborts") { + withTempDir { dir => + createMultiFileTable(dir) + // On the repartition path a shuffle reorders rows across files, so no source composition is + // captured (the operator is not even injected) -> no tag -> the loser aborts (as today). + val e = runOptimizeLoserExpectingAbort( + dir, s"OPTIMIZE ${tableRef(dir)}", deleteId = 150, + extraConf = Seq(DeltaSQLConf.DELTA_OPTIMIZE_REPARTITION_ENABLED.key -> "true")) + assertConcurrentModificationException(e) + assert(ids(dir) === (0L until 300L).filterNot(_ == 150L)) + } + } + + test("reverse direction: a losing DELETE against a compacted file still aborts") { + withTempDir { dir => + val log = createMultiFileTable(dir) + // Only the OPTIMIZE-loser direction is reconciled here. When OPTIMIZE wins (commits first) and + // the DELETE loses, the delete's target file was compacted away; DELETE-loser reconciliation + // is a separate follow-up, so the loser aborts. + val txnDelete = sqlTxn(s"DELETE FROM ${tableRef(dir)} WHERE id = 150", reconcile = true) + val txnOptimize = sqlTxn(s"OPTIMIZE ${tableRef(dir)}", reconcile = true) + + // A = DELETE (starts first, ends last = loser); B = OPTIMIZE (commits in between = winner). + val (futureA, futureB) = runTxnsWithOrder__A_Start__B__A_End(txnDelete, txnOptimize) + ThreadUtils.awaitResult(futureB, Duration.Inf) + val e = intercept[SparkException] { ThreadUtils.awaitResult(futureA, Duration.Inf) } + assertConcurrentModificationException(e) + // OPTIMIZE succeeded and compacted to one file; the delete did not apply. + assert(ids(dir) === (0L until 300L)) + val files = log.update().allFiles.collect() + assert(files.length === 1, s"expected a single compacted file, got ${files.length}") + } + } + + test("partitioned compaction OPTIMIZE (loser) reconciles a concurrent DELETE") { + withTempDir { dir => + val log = createPartitionedMultiFileTable(dir) + // The partitioned case F1 fixes: the writer's required ordering is the partition column, so + // the capture must report that (constant-within-bin) ordering, or a SortExec is inserted above + // the capture and the recorded offsets become scan order rather than physical write order. + // End-to-end this reconciles like the unpartitioned case: both commit, exactly id=150 gone. + val txnA = sqlTxn(s"OPTIMIZE ${tableRef(dir)}", reconcile = true) + val txnB = sqlTxn(s"DELETE FROM ${tableRef(dir)} WHERE id = 150", reconcile = true) + + val (futureA, futureB) = runTxnsWithOrder__A_Start__B__A_End(txnA, txnB) + ThreadUtils.awaitResult(futureB, Duration.Inf) + ThreadUtils.awaitResult(futureA, Duration.Inf) + + assert(ids(dir) === (0L until 300L).filterNot(_ == 150L)) + val files = log.update().allFiles.collect() + assert(files.length === 1, s"expected a single compacted file, got ${files.length}") + assert(deletionVectorCardinalities(log) === Seq(1L), + "compacted output should carry the remapped deletion vector") + } + } + + test("partitioned OPTIMIZE capture reports the partition ordering (no sort above the capture)") { + withTempDir { dir => + createPartitionedMultiFileTable(dir) + // Deterministic, structural proof of the sort-skip. The behavioral partitioned test above is + // non-deterministic (equal partition keys in a bin + an unstable sort); this single-bin table + // compacts to exactly one file, so the executed write plan is deterministic. With the fix the + // capture reports the constant partition-column ordering, the writer's required ordering is + // satisfied, and no SortExec is inserted above the capture, so the recorded offsets are the + // physical write order. We assert on the captured plan directly rather than on the indirect + // `outputOrderingMatched` proxy (which passes vacuously when the required ordering is empty, + // and never checks the capture is present). + DeltaFileFormatWriter.executedPlan = None + withSQLConf(DeltaSQLConf.DELTA_OPTIMIZE_CONFLICT_RECONCILIATION_ENABLED.key -> "true") { + sql(s"OPTIMIZE ${tableRef(dir)}").collect() + } + val plan = DeltaFileFormatWriter.executedPlan.getOrElse( + fail("no executed write plan was captured for the partitioned OPTIMIZE")) + // The capture must be in the write plan at all, else the no-sort assertion below is vacuous. + assert(plan.exists(_.isInstanceOf[SourceCompositionCaptureExec]), + s"expected a SourceCompositionCaptureExec in the write plan, got:\n${plan.treeString}") + // No SortExec may sit above the capture: a sort there would reorder rows after the capture + // recorded them, turning the write-order offsets into scan order. + val sortAboveCapture = plan.exists { + case s: SortExec => s.exists(_.isInstanceOf[SourceCompositionCaptureExec]) + case _ => false + } + assert(!sortAboveCapture, + s"a SortExec was inserted above the capture, corrupting write-order offsets:\n" + + plan.treeString) + // Complementary: the writer must also have found the ordering matched (no sort needed). + assert(DeltaFileFormatWriter.outputOrderingMatched, + "a partitioned OPTIMIZE capture write must not require a sort above the capture operator") + } + } + + // --- Composition-tag parser --- + // The reconcile path reads the source composition through the shared parser + // `ConflictChecker.parseOptimizeSourceComposition`, which decodes the `compactedInto` / + // `compactionInfo` tags (a format shared with Databricks Runtime, so a foreign OPTIMIZE + // reconciles too). These unit-test the parser directly; the end-to-end remap that consumes its + // `(outputPath, outputStart, liveCount)` output is covered by the tests above. + + private def removeWithTags(tags: Map[String, String], dvCardinality: Long = 0L): RemoveFile = { + val dv = + if (dvCardinality > 0L) DeletionVectorDescriptor.EMPTY.copy(cardinality = dvCardinality) + else null + RemoveFile("src.parquet", Some(1L), deletionVector = dv, tags = tags) + } + + private val outputs = Set("out.parquet") + + test("composition tag: parses a single-run compaction entry") { + // No source DV, so sourceNumPhysicalRecords (90) == liveCount. + val r = removeWithTags(Map( + "compactedInto" -> """["out.parquet"]""", + "compactionInfo" -> """[{"rowOffsetInTarget":100,"sourceNumPhysicalRecords":90}]""")) + assert(ConflictChecker.parseOptimizeSourceComposition(r, outputs) === + Some(("out.parquet", 100L, 90L))) + } + + test("composition tag: derives liveCount from the physical count minus the read-time DV") { + // physical 100, read-time DV cardinality 10 on the same tombstone -> liveCount 90. + val r = removeWithTags( + Map( + "compactedInto" -> """["out.parquet"]""", + "compactionInfo" -> """[{"rowOffsetInTarget":100,"sourceNumPhysicalRecords":100}]"""), + dvCardinality = 10L) + assert(ConflictChecker.parseOptimizeSourceComposition(r, outputs) === + Some(("out.parquet", 100L, 90L))) + } + + test("composition tag: tolerates unknown fields in a compaction entry (schema drift)") { + val info = + """[{"rowOffsetInTarget":100,"sourceNumPhysicalRecords":90,"sourceDeletionVector":null}]""" + val r = removeWithTags(Map( + "compactedInto" -> """["out.parquet"]""", + "compactionInfo" -> info)) + assert(ConflictChecker.parseOptimizeSourceComposition(r, outputs) === + Some(("out.parquet", 100L, 90L))) + } + + test("composition tag: falls back (None) on shapes the contiguous offset remap cannot model") { + // Output not among this commit's added files. + assert(ConflictChecker.parseOptimizeSourceComposition( + removeWithTags(Map( + "compactedInto" -> """["elsewhere.parquet"]""", + "compactionInfo" -> """[{"rowOffsetInTarget":0,"sourceNumPhysicalRecords":90}]""")), + outputs).isEmpty) + // Source split across more than one output run (multi-entry) -- not modeled. + assert(ConflictChecker.parseOptimizeSourceComposition( + removeWithTags(Map( + "compactedInto" -> """["out.parquet"]""", + "compactionInfo" -> + ("""[{"rowOffsetInTarget":0,"sourceNumPhysicalRecords":40},""" + + """{"rowOffsetInTarget":40,"sourceNumPhysicalRecords":50}]"""))), + outputs).isEmpty) + // Malformed JSON. + assert(ConflictChecker.parseOptimizeSourceComposition( + removeWithTags(Map("compactedInto" -> """["out.parquet"]""", "compactionInfo" -> "not json")), + outputs).isEmpty) + // No composition tags at all. + assert( + ConflictChecker.parseOptimizeSourceComposition(removeWithTags(Map.empty), outputs).isEmpty) + } +} From 050a63d0fae32eac996d44196a0e4889205e4147 Mon Sep 17 00:00:00 2001 From: Eunjin Song Date: Fri, 7 Aug 2026 22:39:32 -0700 Subject: [PATCH 2/2] [RLC] Fail-safe forward OPTIMIZE reconciliation on DV I/O errors resolveOptimizeConflicts reads, remaps and rewrites deletion vectors as driver-side object-store I/O during conflict detection. A failure there (unreadable/corrupt DV, transient I/O) would surface an unexpected error out of the conflict checker instead of the retryable Concurrent* exception callers expect. Reconciliation is a pure optimization over the conservative abort: the transaction is mutated only on the success path. Wrap the work in try/catch(NonFatal) so any DV read/merge/write failure leaves the transaction untouched and the standard file-level checks abort cleanly. Co-Authored-By: Claude Opus 4.8 --- .../sql/delta/OptimizeConflictReconciliation.scala | 10 +++++++++- 1 file changed, 9 insertions(+), 1 deletion(-) diff --git a/spark/src/main/scala/org/apache/spark/sql/delta/OptimizeConflictReconciliation.scala b/spark/src/main/scala/org/apache/spark/sql/delta/OptimizeConflictReconciliation.scala index 5e90df82320..154e1834b9e 100644 --- a/spark/src/main/scala/org/apache/spark/sql/delta/OptimizeConflictReconciliation.scala +++ b/spark/src/main/scala/org/apache/spark/sql/delta/OptimizeConflictReconciliation.scala @@ -17,6 +17,7 @@ package org.apache.spark.sql.delta import scala.collection.mutable +import scala.util.control.NonFatal import org.apache.spark.sql.delta.actions.{AddFile, RemoveFile} import org.apache.spark.sql.delta.commands.DeletionVectorUtils @@ -103,7 +104,10 @@ trait OptimizeConflictReconciliation extends DeltaLogging { self: ConflictChecke .intersect(currentRemoveByPath.keySet) if (sharedPaths.isEmpty) return - recordTime("resolved-optimize-conflicts") { + // Reconcile is a pure optimization over the conservative abort: mutate the transaction only + // on the success path below, so any DV read/merge/write failure leaves it untouched and the + // standard file-level checks abort cleanly rather than surfacing an unexpected error. + try recordTime("resolved-optimize-conflicts") { val dvStore = DeletionVectorStore.createInstance(deltaLog.newDeltaHadoopConf()) val tablePath = deltaLog.dataPath @@ -198,6 +202,10 @@ trait OptimizeConflictReconciliation extends DeltaLogging { self: ConflictChecke "outputsRemapped" -> addReplacements.size, "winningOperation" -> winningOperationName.getOrElse("UNKNOWN"))) } + } catch { + case NonFatal(e) => + logWarning(log"OPTIMIZE-vs-DML conflict reconciliation failed; leaving all conflicts " + + log"for the standard checks to arbitrate", e) } } }