diff --git a/spark/src/main/scala/org/apache/spark/sql/delta/actions/actions.scala b/spark/src/main/scala/org/apache/spark/sql/delta/actions/actions.scala
index a2cc42da7ea..0a35523de02 100644
--- a/spark/src/main/scala/org/apache/spark/sql/delta/actions/actions.scala
+++ b/spark/src/main/scala/org/apache/spark/sql/delta/actions/actions.scala
@@ -1193,6 +1193,56 @@ case class RemoveFile(
 }
 // scalastyle:on
 
+object RemoveFile {
+  /**
+   * Misc tombstone-level metadata. Clients may safely ignore any of these tags; they must never
+   * affect correctness (an ignored tag only forgoes an optimization, e.g. a conflict reconcile).
+   */
+  object Tags {
+    /**
+     * [[COMPACTED_INTO]] / [[COMPACTION_INFO]]: recorded together on a source file removed by a
+     * compaction OPTIMIZE, describing where that source's rows landed in the compacted output, so
+     * the conflict checker can remap a concurrent deletion vector between the source and the output
+     * instead of aborting. The value format is modeled on what Databricks Runtime records;
+     * interoperating with a DBR-written OPTIMIZE on a shared table is best-effort and not a
+     * verified guarantee (an ignored or unrecognized tag only forgoes the reconcile -- never wrong
+     * data).
+     *
+     *  - [[COMPACTED_INTO]]: JSON array holding the single output path the source compacted into,
+     *    `["<output>.parquet"]` (matching the AddFile.path in the same commit).
+     *  - [[COMPACTION_INFO]]: JSON array holding the single run this source contributed,
+     *    `[{"rowOffsetInTarget": <outputStart>, "sourceNumPhysicalRecords": <physical>}]`. The
+     *    source's live rows land contiguously starting at physical offset `outputStart` of the
+     *    output, in source order. `sourceNumPhysicalRecords` is a PHYSICAL count; the live run
+     *    length is `sourceNumPhysicalRecords - |sourceDV|`, where `sourceDV` is the DV already on
+     *    this same tombstone (the DV the OPTIMIZE read). The physical count keeps the entry
+     *    self-consistent with the tombstone's own DV.
+     *
+     * Kept on the (short-lived) tombstone rather than the output AddFile (which snapshot
+     * reconstruction replays on every read); tombstone retention outlives the conflict window.
+     * Persisted (not stripped before commit) so a concurrent DML that LOSES to this OPTIMIZE can
+     * read the composition from the committed tombstone. Written whenever OPTIMIZE conflict
+     * reconciliation is enabled. O(1) per removed source.
+     */
+    val COMPACTED_INTO = "compactedInto"
+    val COMPACTION_INFO = "compactionInfo"
+  }
+}
+
+/**
+ * The per-source entry recorded in a compaction OPTIMIZE's `compactionInfo` tombstone tag: where
+ * that source's rows landed in the compacted output. The format is modeled on Databricks Runtime's
+ * (see [[RemoveFile.Tags.COMPACTION_INFO]]); reconciling against a DBR-written tag on a shared
+ * table is best-effort, not a verified guarantee. Every field is optional and unknown fields are
+ * ignored, so a foreign writer's schema drift degrades to a safe abort rather than a wrong result.
+ */
+@JsonIgnoreProperties(ignoreUnknown = true)
+private[delta] case class CompactionInfoEntry(
+    @JsonDeserialize(contentAs = classOf[java.lang.Long])
+    rowOffsetInTarget: Option[Long] = None,
+    @JsonDeserialize(contentAs = classOf[java.lang.Long])
+    sourceNumPhysicalRecords: Option[Long] = None)
+
 /**
  * A change file containing CDC data for the Delta version it's within. Non-CDC readers should
  * ignore this, CDC readers should scan all ChangeFiles in a version rather than computing
diff --git a/spark/src/main/scala/org/apache/spark/sql/delta/commands/OptimizeTableCommand.scala b/spark/src/main/scala/org/apache/spark/sql/delta/commands/OptimizeTableCommand.scala
index d40f2972043..dbe76ba300c 100644
--- a/spark/src/main/scala/org/apache/spark/sql/delta/commands/OptimizeTableCommand.scala
+++ b/spark/src/main/scala/org/apache/spark/sql/delta/commands/OptimizeTableCommand.scala
@@ -19,23 +19,25 @@ package org.apache.spark.sql.delta.commands
 import java.util.ConcurrentModificationException
 
 import scala.collection.mutable.ArrayBuffer
+import scala.util.control.NonFatal
 
 import org.apache.spark.sql.delta.skipping.MultiDimClustering
 import org.apache.spark.sql.delta.skipping.clustering.{ClusteredTableUtils, ClusteringColumnInfo}
 import org.apache.spark.sql.delta._
+import org.apache.spark.sql.delta.ClassicColumnConversions._
 import org.apache.spark.sql.delta.DeltaOperations.Operation
-import org.apache.spark.sql.delta.actions.{Action, AddFile, DeletionVectorDescriptor, FileAction, RemoveFile}
+import org.apache.spark.sql.delta.actions.{Action, AddFile, CompactionInfoEntry, DeletionVectorDescriptor, FileAction, RemoveFile}
 import org.apache.spark.sql.delta.commands.optimize._
-import org.apache.spark.sql.delta.files.SQLMetricsReporting
+import org.apache.spark.sql.delta.files.{SourceCompositionAccumulator, SourceCompositionCaptureExec, SQLMetricsReporting}
 import org.apache.spark.sql.delta.logging.DeltaLogKeys
 import org.apache.spark.sql.delta.schema.{SchemaUtils, UnsupportedDataTypeInfo}
 import org.apache.spark.sql.delta.sources.DeltaSQLConf
-import org.apache.spark.sql.delta.util.BinPackingUtils
+import org.apache.spark.sql.delta.util.{BinPackingUtils, DeltaFileOperations, JsonUtils}
 
 import org.apache.spark.SparkContext
 import org.apache.spark.SparkContext.SPARK_JOB_GROUP_ID
 import org.apache.spark.internal.MDC
-import org.apache.spark.sql.{AnalysisException, Encoders, Row, SparkSession}
+import org.apache.spark.sql.{AnalysisException, DataFrame, Encoders, Row, SparkSession}
 import org.apache.spark.sql.catalyst.TableIdentifier
 import org.apache.spark.sql.catalyst.analysis.{UnresolvedAttribute, UnresolvedTable}
 import org.apache.spark.sql.catalyst.expressions.{Attribute, AttributeReference, Expression}
@@ -43,6 +45,7 @@ import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, UnaryNode}
 import org.apache.spark.sql.execution.command.RunnableCommand
 import org.apache.spark.sql.execution.metric.SQLMetric
 import org.apache.spark.sql.execution.metric.SQLMetrics.createMetric
+import org.apache.spark.sql.internal.SQLConf
 import org.apache.spark.sql.types._
 import org.apache.spark.util.{SystemClock, ThreadUtils}
 import org.apache.spark.sql.catalyst.catalog.CatalogTable
@@ -519,9 +522,40 @@ class OptimizeExecutor(
       bin: Seq[AddFile],
       maxFileSize: Long): Seq[FileAction] = {
     val baseTablePath = txn.deltaLog.dataPath
-
-    var input = txn.deltaLog.createDataFrame(txn.snapshot, bin, actionTypeOpt = Some("Optimize"))
+    // Compaction conflict-reconciliation (optimize.conflictReconciliation.enabled): observe the
+    // source composition of the compacted output at write time (SourceCompositionCaptureExec) so a
+    // concurrent DML's deletion vector can be remapped onto it instead of aborting. Compaction
+    // only, never clustering (a clustering pass permutes rows, so no offset mapping exists).
+    val reconcileEnabled = sparkSession.sessionState.conf
+      .getConf(DeltaSQLConf.DELTA_OPTIMIZE_CONFLICT_RECONCILIATION_ENABLED)
+    val useRepartition = sparkSession.sessionState.conf
+      .getConf(DeltaSQLConf.DELTA_OPTIMIZE_REPARTITION_ENABLED)
+    // Capture the source composition only on the coalesce path for compaction: coalesce
+    // preserves each source file's read order, so rows stay contiguous in the output
+    // (observed, not imposed -- no sort, no helper column) and row-range offsets exist.
+    // Repartition shuffles rows and a clustering pass permutes them, so neither is captured.
+    val captureReconcile =
+      reconcileEnabled && !isMultiDimClustering && !useRepartition
+
+    // Read the bin. The reconcile-capture path pins the read so each source file lands whole in
+    // one contiguous run (see readCompactionSourceWithWholeFilePins); vanilla OPTIMIZE just reads
+    // on the current session.
+    var input = if (captureReconcile) {
+      readCompactionSourceWithWholeFilePins(txn, bin, maxFileSize)
+    } else {
+      txn.deltaLog.createDataFrame(txn.snapshot, bin, actionTypeOpt = Some("Optimize"))
+    }
     input = RowTracking.preserveRowTrackingColumns(input, txn.snapshot)
+
+    val captureAccOpt: Option[SourceCompositionAccumulator] =
+      if (captureReconcile) {
+        val acc = new SourceCompositionAccumulator
+        sparkSession.sparkContext.register(acc)
+        Some(acc)
+      } else {
+        None
+      }
+
     val repartitionDF = if (isMultiDimClustering) {
       val totalSize = bin.map(_.size).sum
       val approxNumFiles = Math.max(1, totalSize / maxFileSize).toInt
@@ -531,8 +565,6 @@ class OptimizeExecutor(
         clusteringColumns,
         optimizeStrategy.curve)
     } else {
-      val useRepartition = sparkSession.sessionState.conf.getConf(
-        DeltaSQLConf.DELTA_OPTIMIZE_REPARTITION_ENABLED)
       if (useRepartition) {
         input.repartition(numPartitions = 1)
       } else {
@@ -549,18 +581,158 @@ class OptimizeExecutor(
       description)
 
     val binInfo = optimizeStrategy.initNewBin
-    val addFiles = txn.writeFiles(repartitionDF, None, isOptimize = true, Nil).collect {
+    val addFiles = txn.writeFiles(repartitionDF, None, isOptimize = true, Nil,
+        sourceCompositionCapture = captureAccOpt).collect {
       case a: AddFile => optimizeStrategy.tagAddFile(a, binInfo)
       case other =>
         throw new IllegalStateException(
           s"Unexpected action $other with type ${other.getClass}. File compaction job output" +
               s"should only have AddFiles")
     }
-    val removeFiles = bin.map(f => f.removeWithTimestamp(operationTimestamp, dataChange = false))
+    // Tombstones for the removed sources. Only the RLC reconciliation path tags each source with
+    // where its rows landed in the compacted output (so a concurrent DML's DV can be remapped by
+    // offset instead of aborting); vanilla OPTIMIZE uses plain untagged tombstones and skips the
+    // reconcile helper entirely.
+    val removeFiles = if (captureReconcile) {
+      buildRemoveFilesWithCompactionCompositionTags(
+        txn, bin, addFiles, captureAccOpt, operationTimestamp)
+    } else {
+      bin.map(_.removeWithTimestamp(operationTimestamp, dataChange = false))
+    }
     val updates = addFiles ++ removeFiles
     updates
   }
 
+  /**
+   * Read `bin` for a compaction OPTIMIZE on the reconciliation-capture path, pinning the read so
+   * each source file lands whole in a single partition.
+   *
+   * Each source file must be read whole so no source is split across partitions -- coalesce(1)
+   * then lands each source as one contiguous run, which is what the capture's one-run-per-file
+   * gate needs. Spark has no "do not split" toggle for Parquet; splitting is governed by
+   *   maxSplitBytes = min(maxPartitionBytes, max(openCostInBytes, totalBytes / minPartitionNum))
+   * so pin BOTH maxPartitionBytes (>= the compaction target) and minPartitionNum = 1: then
+   * maxSplitBytes >= every source (each <= the target), so no source splits. (A split that still
+   * somehow slips through just fails the capture gate and aborts -- never wrong data.) Both confs
+   * go on a CLONED session so the override is isolated from other queries -- and from the other
+   * bins compacting concurrently -- that share this SparkSession: createDataFrame binds the scan
+   * relation to SparkSession.active and Spark reads these confs from the captured session, so the
+   * clone need only be active while the relation is built, then the previous active session is
+   * restored.
+   */
+  private def readCompactionSourceWithWholeFilePins(
+      txn: OptimisticTransaction,
+      bin: Seq[AddFile],
+      maxFileSize: Long): DataFrame = {
+    val readSession = sparkSession.cloneSession()
+    readSession.conf.set(SQLConf.FILES_MAX_PARTITION_BYTES.key, maxFileSize)
+    readSession.conf.set(SQLConf.FILES_MIN_PARTITION_NUM.key, "1")
+    val prevActive = SparkSession.getActiveSession
+    SparkSession.setActiveSession(readSession)
+    try {
+      txn.deltaLog.createDataFrame(txn.snapshot, bin, actionTypeOpt = Some("Optimize"))
+    } finally {
+      prevActive.fold(SparkSession.clearActiveSession())(SparkSession.setActiveSession)
+    }
+  }
+
+  /**
+   * Build the removed-source tombstones for a compaction OPTIMIZE, tagging each with its
+   * `compactedInto` / `compactionInfo` composition -- where that source's rows landed in the single
+   * compacted output -- so a concurrent DML's deletion vector can be remapped by offset instead of
+   * aborting (see [[RemoveFile.Tags.COMPACTION_INFO]]). The composition is persisted, as Databricks
+   * Runtime does on every compaction OPTIMIZE: consumed in-memory when THIS OPTIMIZE loses to a
+   * concurrent DML, and read back from the committed tombstone when a concurrent DML LOSES to this
+   * OPTIMIZE. The tag format is modeled on the one Databricks Runtime writes; reconciling against a
+   * DBR-written tag on a shared table is best-effort, not a verified guarantee.
+   *
+   * Falls back to plain untagged tombstones -- so the conflict aborts as it does today -- unless
+   * the capture is present (reconciliation was enabled) AND trustworthy: exactly one output file
+   * from exactly one write partition, and each source contributed exactly one captured run covering
+   * the whole bin (a sanity gate against retries / speculation / splits).
+   *
+   * Each source's `compactionInfo` records `sourceNumPhysicalRecords` (the live rows the write saw
+   * PLUS the source's read-time DV cardinality), not the live count, so the tags stay O(1) per
+   * source regardless of how fragmented a source DV is; the conflict checker recovers the live run
+   * length by subtracting the tombstone's own DV, and rebuilds the read-time gaps only on a real
+   * conflict.
+   */
+  private def buildRemoveFilesWithCompactionCompositionTags(
+      txn: OptimisticTransaction,
+      bin: Seq[AddFile],
+      addFiles: Seq[AddFile],
+      captureAccOpt: Option[SourceCompositionAccumulator],
+      operationTimestamp: Long): Seq[RemoveFile] = {
+    // The fallback: plain untagged tombstones, exactly as vanilla OPTIMIZE writes them, whenever
+    // the capture is off or the trustworthiness gate below rejects it (the loser aborts as today).
+    def untagged: Seq[RemoveFile] =
+      bin.map(_.removeWithTimestamp(operationTimestamp, dataChange = false))
+
+    try captureAccOpt match {
+      case Some(acc) if addFiles.size == 1 && acc.value.size() == 1 &&
+          bin.forall(_.numLogicalRecords.isDefined) =>
+        val runs = acc.value.get(0)
+        val captured = (0 until runs.size()).map(runs.get(_).count).sum
+        val expected = bin.flatMap(_.numLogicalRecords).sum
+        // Map each run's absolute source path (from the holder) back to the table-relative AddFile
+        // path, so the recorded keys match the RemoveFiles / winning DV updates (both relative) at
+        // conflict-resolution time.
+        val nameToAddFile = generateCandidateFileMap(txn.deltaLog.dataPath, bin)
+        val tablePath = txn.deltaLog.dataPath
+        val compactedIntoJson = JsonUtils.toJson(Seq(addFiles.head.path))
+        // Walk runs in output (write) order, accumulating each source's start offset in the output.
+        var outputPos = 0L
+        val tagByPath: Seq[Option[(String, Map[String, String])]] = (0 until runs.size()).map { i =>
+          val r = runs.get(i)
+          val start = outputPos
+          outputPos += r.count
+          // A null source file (the holder was empty for some rows) can't be mapped back to an
+          // AddFile; yield None so the whole capture is treated as unreconcilable below (no NPE in
+          // absolutePath), and the loser aborts as it does today.
+          if (r.sourceFile == null) {
+            None
+          } else {
+            val abs =
+              DeltaFileOperations.absolutePath(tablePath.toString, r.sourceFile).toString
+            nameToAddFile.get(abs).map { add =>
+              // `r.count` is the live rows the write saw; the physical count adds back the source's
+              // read-time DV (the DV carried onto this source's tombstone below).
+              val sourceDvCardinality = Option(add.deletionVector).map(_.cardinality).getOrElse(0L)
+              val physical = r.count + sourceDvCardinality
+              val compactionInfoJson =
+                JsonUtils.toJson(Seq(CompactionInfoEntry(Some(start), Some(physical))))
+              add.path -> Map(
+                RemoveFile.Tags.COMPACTED_INTO -> compactedIntoJson,
+                RemoveFile.Tags.COMPACTION_INFO -> compactionInfoJson)
+            }
+          }
+        }
+        // Each source file mapped and produced exactly one captured run covering the whole bin.
+        val oneRunPerFile = runs.size() == bin.size &&
+          (0 until runs.size()).map(runs.get(_).sourceFile).distinct.size == bin.size
+        if (captured == expected && tagByPath.forall(_.isDefined) && oneRunPerFile) {
+          val tags = tagByPath.flatten.toMap
+          bin.map { f =>
+            val r = f.removeWithTimestamp(operationTimestamp, dataChange = false)
+            tags.get(f.path).fold(r)(
+              _.foldLeft(r) { case (tagged, (k, v)) => tagged.copyWithTag(k, v) })
+          }
+        } else {
+          untagged
+        }
+      case _ =>
+        untagged
+    } catch {
+      case NonFatal(e) =>
+        // Composition capture is a pure optimization enabler, never required for OPTIMIZE
+        // correctness: on any failure building the tags (an unmappable source path, a
+        // serialization error) fall back to plain untagged tombstones so the already-written
+        // OPTIMIZE still commits and a concurrent loser aborts exactly as it does today.
+        logWarning(log"Compaction composition capture failed; writing untagged tombstones", e)
+        untagged
+    }
+  }
+
   /**
    * Attempts to commit the given actions to the log. In the case of a concurrent update,
    * the given function will be invoked with a new transaction to allow custom conflict
diff --git a/spark/src/main/scala/org/apache/spark/sql/delta/files/DeltaFileFormatWriter.scala b/spark/src/main/scala/org/apache/spark/sql/delta/files/DeltaFileFormatWriter.scala
index bf7868c1ab4..419fe730809 100644
--- a/spark/src/main/scala/org/apache/spark/sql/delta/files/DeltaFileFormatWriter.scala
+++ b/spark/src/main/scala/org/apache/spark/sql/delta/files/DeltaFileFormatWriter.scala
@@ -61,12 +61,12 @@ object DeltaFileFormatWriter extends Logging {
    * A variable used in tests to check whether the output ordering of the query matches the
    * required ordering of the write command.
    */
-  private var outputOrderingMatched: Boolean = false
+  private[delta] var outputOrderingMatched: Boolean = false
 
   /**
    * A variable used in tests to check the final executed plan.
    */
-  private var executedPlan: Option[SparkPlan] = None
+  private[delta] var executedPlan: Option[SparkPlan] = None
 
   // scalastyle:off argcount
   /**
diff --git a/spark/src/main/scala/org/apache/spark/sql/delta/files/SourceCompositionCaptureExec.scala b/spark/src/main/scala/org/apache/spark/sql/delta/files/SourceCompositionCaptureExec.scala
new file mode 100644
index 00000000000..44abdd5f47c
--- /dev/null
+++ b/spark/src/main/scala/org/apache/spark/sql/delta/files/SourceCompositionCaptureExec.scala
@@ -0,0 +1,188 @@
+/*
+ * 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.files
+
+import scala.collection.mutable
+
+import org.apache.spark.rdd.{InputFileBlockHolder, RDD}
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions.{Attribute, SortOrder}
+import org.apache.spark.sql.execution.{SparkPlan, UnaryExecNode}
+import org.apache.spark.sql.vectorized.ColumnarBatch
+import org.apache.spark.unsafe.types.UTF8String
+import org.apache.spark.util.{AccumulatorV2, CompletionIterator}
+
+/** A contiguous run of `count` live rows from one source file, landing at consecutive positions in
+ *  a compaction output in write order. `count` is the rows the write actually saw (the scan has
+ *  already applied any compaction-time deletion vector), so a DV'd source is still one run of its
+ *  live rows; the physical row_index of each live row is reconstructed at conflict-resolution time
+ *  from the source's read-time deletion vector, not recorded here. */
+case class SourceRun(sourceFile: String, count: Long)
+
+/**
+ * Accumulates, per write task, the ordered [[SourceRun]]s observed while a compaction OPTIMIZE
+ * writes its output. Each successful task contributes one entry (the runs it saw, in write order).
+ * The driver expects exactly one entry (a single-output compaction bin, one partition); anything
+ * else (speculation, a split into multiple files) is treated as unreconcilable and the tag is
+ * dropped, so the loser simply aborts as it does today.
+ */
+class SourceCompositionAccumulator
+  extends AccumulatorV2[Seq[SourceRun], java.util.List[java.util.List[SourceRun]]] {
+
+  private val partitionRuns = new java.util.ArrayList[java.util.List[SourceRun]]()
+
+  override def isZero: Boolean = partitionRuns.isEmpty
+
+  override def copy(): SourceCompositionAccumulator = {
+    val c = new SourceCompositionAccumulator
+    c.partitionRuns.addAll(partitionRuns)
+    c
+  }
+
+  override def reset(): Unit = partitionRuns.clear()
+
+  override def add(runs: Seq[SourceRun]): Unit = {
+    val list = new java.util.ArrayList[SourceRun](runs.length)
+    runs.foreach(list.add)
+    partitionRuns.add(list)
+  }
+
+  override def merge(
+      other: AccumulatorV2[Seq[SourceRun], java.util.List[java.util.List[SourceRun]]]): Unit =
+    partitionRuns.addAll(other.value)
+
+  override def value: java.util.List[java.util.List[SourceRun]] = partitionRuns
+}
+
+/**
+ * Per write-task state shared by the row and columnar execution paths: reads the scan's
+ * thread-local file identity ([[InputFileBlockHolder]]) and folds consecutive same-file units --
+ * a single row, or a whole columnar batch -- into one [[SourceRun]], in observed write order.
+ */
+private class RunTracker {
+  private val runs = mutable.ArrayBuffer.empty[SourceRun]
+  // Holder instance for the current file (stable per file, so `eq` fast-paths the same-file hot
+  // path); the path string is materialized once per boundary, not per unit.
+  private var curFileUtf: UTF8String = null
+  private var curFile: String = null
+  private var curCount = 0L
+
+  private def closeRun(): Unit = if (curCount > 0) runs += SourceRun(curFile, curCount)
+
+  /**
+   * Fold `delta` more rows of the current scan file into the open run, starting a new run when the
+   * thread-local file identity changes. `delta` is 1 for a row, or the batch row count for a batch.
+   */
+  def observe(delta: Long): Unit = {
+    val sfUtf = InputFileBlockHolder.getInputFilePath
+    val sameFile = (sfUtf eq curFileUtf) || (sfUtf != null && sfUtf.equals(curFileUtf))
+    if (sameFile) {
+      curCount += delta
+    } else {
+      closeRun()
+      curFileUtf = sfUtf
+      curFile = if (sfUtf == null || sfUtf.numBytes() == 0) null else sfUtf.toString
+      curCount = delta
+    }
+  }
+
+  /** Close the final open run and return the ordered runs this task observed. */
+  def finish(): Seq[SourceRun] = {
+    closeRun()
+    runs.toSeq
+  }
+}
+
+/**
+ * A write-stage operator for OPTIMIZE compaction conflict-reconciliation, injected into the write
+ * plan (like [[DeltaOptimizedWriterExec]]). It records, per source file, how many rows that file
+ * contributed to the compaction output and in what order, with no per-row helper column.
+ *
+ * The source file identity is read per row from [[InputFileBlockHolder]] (the scan's thread-local,
+ * the same one `input_file_name()` reads); a per-file counter tracks each file's live row count.
+ * Rows are emitted unchanged: there is no extra column to strip (no row copy) and no
+ * `_metadata.row_index` materialization (which would drag in the DV-aware scan cost). On a
+ * contiguous coalesce read each file's live rows land in one output segment, so `(sourceFile,
+ * count)` in write order fully describes the layout and the driver derives the output offsets. A
+ * source that had a compaction-time deletion vector is still one run (of its live rows); its
+ * physical row_index gaps are reconstructed at conflict time, so no DV is read here.
+ *
+ * The operator handles both execution modes. In row mode the identity is read once per row; when
+ * the child produces columnar batches (a vectorized / native execution backend), it stays columnar
+ * -- reading the identity once per batch and passing the batch through unchanged -- so it adds no
+ * columnar-to-row transition (which would materialize every batch just to observe it). A batch
+ * holds rows from a single scan file, so one read per batch is exact. Either mode folds units into
+ * runs through the same [[RunTracker]], so the two paths are identical by construction.
+ *
+ * Runs flush to the accumulator on successful task completion; failed attempts flush nothing. The
+ * [[InputFileBlockHolder]] read is only valid when the scan and this operator run in the same task
+ * with no shuffle between them (the coalesce path, the default). On the repartition path the holder
+ * is empty after the shuffle, so no file is recorded and the loser aborts as it does today.
+ */
+case class SourceCompositionCaptureExec(
+    child: SparkPlan,
+    acc: SourceCompositionAccumulator,
+    childOutputOrdering: Seq[SortOrder] = Nil) extends UnaryExecNode {
+
+  override def output: Seq[Attribute] = child.output
+
+  // For a partitioned OPTIMIZE, DeltaFileFormatWriter's requiredOrdering is the partition column,
+  // which is constant within a single compaction bin, so the output IS trivially ordered by it.
+  // Reporting that ordering keeps `orderingMatched` true and stops the writer from inserting a
+  // SortExec ABOVE this operator, which would reorder rows and invalidate the captured write-order
+  // offsets. Falls back to the child's ordering when unset (the unpartitioned case, where the
+  // writer's requiredOrdering is already empty).
+  override def outputOrdering: Seq[SortOrder] =
+    if (childOutputOrdering.nonEmpty) childOutputOrdering else child.outputOrdering
+
+  override def doExecute(): RDD[InternalRow] = {
+    val accumulator = acc
+    child.execute().mapPartitions { iter =>
+      val tracker = new RunTracker
+      // One row observed per element; rows pass through unchanged (no helper column, no copy).
+      val mapped = iter.map { row => tracker.observe(1L); row }
+      // Flush the observed runs once the writer has consumed the whole partition. Using
+      // CompletionIterator (rather than a task-completion listener) runs the flush inside the task
+      // body, so the accumulator update is collected and propagated to the driver.
+      CompletionIterator[InternalRow, Iterator[InternalRow]](
+        mapped, accumulator.add(tracker.finish()))
+    }
+  }
+
+  // Stay columnar when the child is (a vectorized / native execution backend): a row-only operator
+  // would force a columnar-to-row transition here just to observe the file identity.
+  override def supportsColumnar: Boolean = child.supportsColumnar
+
+  override protected def doExecuteColumnar(): RDD[ColumnarBatch] = {
+    val accumulator = acc
+    child.executeColumnar().mapPartitions { iter =>
+      val tracker = new RunTracker
+      val mapped = iter.map { batch =>
+        // A columnar batch holds rows from a single scan file, so the file identity is read once
+        // per batch (not per row) and the whole batch's row count extends the current run.
+        val n = batch.numRows()
+        if (n > 0) tracker.observe(n.toLong)
+        batch // pass through unchanged
+      }
+      CompletionIterator[ColumnarBatch, Iterator[ColumnarBatch]](
+        mapped, accumulator.add(tracker.finish()))
+    }
+  }
+
+  override protected def withNewChildInternal(newChild: SparkPlan): SourceCompositionCaptureExec =
+    copy(child = newChild)
+}
diff --git a/spark/src/main/scala/org/apache/spark/sql/delta/files/TransactionalWrite.scala b/spark/src/main/scala/org/apache/spark/sql/delta/files/TransactionalWrite.scala
index 2680bc18330..d8e60b04dfa 100644
--- a/spark/src/main/scala/org/apache/spark/sql/delta/files/TransactionalWrite.scala
+++ b/spark/src/main/scala/org/apache/spark/sql/delta/files/TransactionalWrite.scala
@@ -407,7 +407,23 @@ trait TransactionalWrite extends DeltaLogging { self: OptimisticTransactionImpl
       inputData: Dataset[_],
       writeOptions: Option[DeltaOptions],
       isOptimize: Boolean,
-      additionalConstraints: Seq[Constraint]): Seq[FileAction] = {
+      additionalConstraints: Seq[Constraint]): Seq[FileAction] =
+    writeFiles(inputData, writeOptions, isOptimize, additionalConstraints,
+      sourceCompositionCapture = None)
+
+  /**
+   * [[writeFiles]] plus OPTIMIZE compaction conflict-reconciliation: when
+   * `sourceCompositionCapture` is set, a `SourceCompositionCaptureExec` is injected to observe the
+   * output's source composition (file identity + per-file row count) into that accumulator. No
+   * helper columns are added; rows pass through unchanged. Kept as a separate overload so the
+   * public [[writeFiles]] signature above is undisturbed for its many callers.
+   */
+  def writeFiles(
+      inputData: Dataset[_],
+      writeOptions: Option[DeltaOptions],
+      isOptimize: Boolean,
+      additionalConstraints: Seq[Constraint],
+      sourceCompositionCapture: Option[SourceCompositionAccumulator]): Seq[FileAction] = {
     hasWritten = true
 
     val spark = inputData.sparkSession
@@ -457,12 +473,25 @@ trait TransactionalWrite extends DeltaLogging { self: OptimisticTransactionImpl
       val checkInvariants = DeltaInvariantCheckerExec(spark, empty2NullPlan, constraints)
       // No need to plan optimized write if the write command is OPTIMIZE, which aims to produce
       // evenly-balanced data files already.
-      val physicalPlan = if (!isOptimize &&
+      val basePlan = if (!isOptimize &&
         shouldOptimizeWrite(writeOptions, spark.sessionState.conf)) {
         DeltaOptimizedWriterExec(checkInvariants, metadata.partitionColumns, deltaLog)
       } else {
         checkInvariants
       }
+      // OPTIMIZE compaction conflict-reconciliation: observe the source composition (file identity
+      // from InputFileBlockHolder + per-file row count) while rows pass through unchanged. No
+      // helper column to strip; the driver derives physical offsets from the per-file counts.
+      val physicalPlan = sourceCompositionCapture match {
+        // Report the partition-column ordering (constant within a compaction bin) so the writer's
+        // required ordering is satisfied and no SortExec is inserted above the capture, which would
+        // reorder rows and break the recorded write-order offsets. Empty (Nil) for unpartitioned
+        // tables, where the writer's required ordering is already empty.
+        case Some(acc) =>
+          SourceCompositionCaptureExec(
+            basePlan, acc, partitioningColumns.map(SortOrder(_, Ascending)))
+        case None => basePlan
+      }
 
       val statsTrackers: ListBuffer[WriteJobStatsTracker] = ListBuffer()
 
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 1277a464b86..5e3883d2c45 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
@@ -547,6 +547,22 @@ trait DeltaSQLConfBase extends DeltaSQLConfUtils {
       .booleanConf
       .createWithDefault(false)
 
+  val DELTA_OPTIMIZE_CONFLICT_RECONCILIATION_ENABLED =
+    buildConf("optimize.conflictReconciliation.enabled")
+      .internal()
+      .doc(
+        """When enabled, a compaction OPTIMIZE that conflicts with a concurrent row-level DML
+          |(DELETE/UPDATE) reconciles instead of aborting: it remaps the concurrent deletion
+          |vector from each removed source file onto the compacted output file by offset
+          |arithmetic (output position = source-file offset + physical row index) and unions it
+          |into the output's deletion vector. Compaction only (order-preserving); reclustering, and
+          |sources that already carried a deletion vector at read time, are left to abort. Relies on
+          |the OPTIMIZE having recorded per-output source composition; if absent (e.g. a native
+          |write bypassed it) the conflict aborts. Only active when deletion vectors are
+          |writable.""".stripMargin)
+      .booleanConf
+      .createWithDefault(false)
+
   val DELTA_PROTOCOL_DEFAULT_WRITER_VERSION =
     buildConf("properties.defaults.minWriterVersion")
       .doc("The default writer protocol version to create new tables with, unless a feature " +
diff --git a/spark/src/test/scala/org/apache/spark/sql/delta/files/SourceCompositionCaptureExecSuite.scala b/spark/src/test/scala/org/apache/spark/sql/delta/files/SourceCompositionCaptureExecSuite.scala
new file mode 100644
index 00000000000..f1cd48f3199
--- /dev/null
+++ b/spark/src/test/scala/org/apache/spark/sql/delta/files/SourceCompositionCaptureExecSuite.scala
@@ -0,0 +1,329 @@
+/*
+ * 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.files
+
+import java.io.File
+
+import scala.jdk.CollectionConverters._
+
+import org.apache.spark.sql.delta.DeltaLog
+import org.apache.spark.sql.delta.actions.{Action, AddFile, CompactionInfoEntry, RemoveFile}
+import org.apache.spark.sql.delta.sources.DeltaSQLConf
+import org.apache.spark.sql.delta.test.DeltaSQLCommandTest
+import org.apache.spark.sql.delta.util.JsonUtils
+import org.apache.hadoop.fs.Path
+import org.apache.parquet.hadoop.ParquetFileReader
+import org.apache.parquet.hadoop.util.HadoopInputFile
+
+import org.apache.spark.rdd.{InputFileBlockHolder, RDD}
+import org.apache.spark.sql.{QueryTest, Row}
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions.{Attribute, AttributeReference}
+import org.apache.spark.sql.execution.LeafExecNode
+import org.apache.spark.sql.execution.vectorized.OnHeapColumnVector
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.types.IntegerType
+import org.apache.spark.sql.vectorized.{ColumnarBatch, ColumnVector}
+
+/**
+ * Tests for the source-composition capture a compaction OPTIMIZE performs.
+ *
+ * The columnar path of [[SourceCompositionCaptureExec]] is exercised directly with a stub columnar
+ * child, since a columnar execution backend is not available in the OSS test harness; both the row
+ * and columnar paths fold units into runs through the same `RunTracker`. The end-to-end tests then
+ * run a real compaction OPTIMIZE and assert what the write side persists on the removed-source
+ * tombstones: contiguous per-source composition tags when the capture is trustworthy, and a safe
+ * fall back to plain untagged tombstones (so a losing DML aborts as today) when it is not.
+ */
+class SourceCompositionCaptureExecSuite extends QueryTest with DeltaSQLCommandTest {
+
+  test("columnar path folds one run per source file across batches, in write order") {
+    val acc = new SourceCompositionAccumulator
+    spark.sparkContext.register(acc)
+    // Two batches from fileA then one from fileB, as a coalesced scan would feed them.
+    val child = FakeColumnarScan(Seq(("fileA", 3), ("fileA", 2), ("fileB", 4)))
+    // Force the columnar RDD: batches pass through and runs flush to the accumulator on completion.
+    SourceCompositionCaptureExec(child, acc).executeColumnar().foreach(_ => ())
+
+    assert(acc.value.size() == 1, "a single partition contributes exactly one entry")
+    val runs = acc.value.get(0).asScala.toSeq
+    // fileA's two batches fold into one run of 5; fileB starts a new run at the boundary.
+    assert(runs == Seq(SourceRun("fileA", 5), SourceRun("fileB", 4)))
+  }
+
+  test("interleaved source batches surface as separate runs (the mixed shape the gate rejects)") {
+    val acc = new SourceCompositionAccumulator
+    spark.sparkContext.register(acc)
+    // fileA, fileB, then fileA again -- the interleaving a split-and-packed scan can produce when a
+    // single source file is broken into row-group splits that pack non-adjacently.
+    val child = FakeColumnarScan(Seq(("fileA", 3), ("fileB", 4), ("fileA", 2)))
+    SourceCompositionCaptureExec(child, acc).executeColumnar().foreach(_ => ())
+
+    assert(acc.value.size() == 1)
+    val runs = acc.value.get(0).asScala.toSeq
+    // fileA is NOT folded across fileB: it appears as two runs. A downstream one-run-per-file gate
+    // therefore sees fileA twice and declines to record a (mixed) composition -- reconcile aborts.
+    assert(runs == Seq(SourceRun("fileA", 3), SourceRun("fileB", 4), SourceRun("fileA", 2)))
+  }
+
+  test("supportsColumnar mirrors the child so the operator stays columnar-transparent") {
+    val acc = new SourceCompositionAccumulator
+    assert(SourceCompositionCaptureExec(FakeColumnarScan(Nil), acc).supportsColumnar)
+    assert(!SourceCompositionCaptureExec(FakeRowScan(), acc).supportsColumnar)
+  }
+
+  test("compaction OPTIMIZE tags each multi-row-group source as one contiguous run") {
+    withTempDir { dir =>
+      val path = dir.getCanonicalPath
+      val hadoopConf = spark.sparkContext.hadoopConfiguration
+      val prevBlockSize = hadoopConf.get("parquet.block.size")
+      // A tiny row-group size so each source file is written as MANY row groups -- the multi-piece
+      // read that would defeat capture if a source were split across partitions. The pinned read
+      // must still land each source whole, folding its row groups into one contiguous run.
+      hadoopConf.set("parquet.block.size", "1024")
+      try {
+        // Two differently sized sources so contiguity is observable regardless of write order.
+        spark.range(0, 3000).repartition(1).write.format("delta").mode("append").save(path)
+        spark.range(3000, 5000).repartition(1).write.format("delta").mode("append").save(path)
+      } finally {
+        if (prevBlockSize == null) hadoopConf.unset("parquet.block.size")
+        else hadoopConf.set("parquet.block.size", prevBlockSize)
+      }
+
+      // Precondition: each source really is multi-row-group (otherwise the test proves nothing).
+      val rowGroups = rowGroupCountsPerFile(dir)
+      assert(rowGroups.size == 2 && rowGroups.forall(_ > 1),
+        s"expected two multi-row-group sources, got $rowGroups")
+
+      // A hostile ambient split size: 8 KiB is smaller than either source (~11.6 KiB / ~17.6 KiB),
+      // so each is broken into a full split plus a row-bearing remainder. Under coalesce(1)'s
+      // descending-length split packing that remainder is read after the other source's head, so
+      // WITHOUT the pins the two sources interleave -- each would surface to the capture as more
+      // than one run, the one-run-per-file gate would decline, and the tags asserted below would be
+      // absent. The capture path pins the read against exactly this: it clones the session and sets
+      // maxPartitionBytes to the compaction target (>= every source) and minPartitionNum = 1, so
+      // each source reads whole in one partition and its row groups fold into one contiguous run.
+      // Drop the pins in readCompactionSourceWithWholeFilePins and this test fails.
+      withSQLConf(
+          DeltaSQLConf.DELTA_OPTIMIZE_CONFLICT_RECONCILIATION_ENABLED.key -> "true",
+          SQLConf.FILES_MAX_PARTITION_BYTES.key -> "8192") {
+        sql(s"OPTIMIZE delta.`$path`")
+      }
+
+      val actions = optimizeCommitActions(path)
+      val adds = actions.collect { case a: AddFile => a }
+      val removes = actions.collect { case r: RemoveFile => r }
+      assert(adds.size == 1, "the bin compacts into a single output file")
+      assert(removes.size == 2, "both sources are removed")
+
+      // Every source is tagged and points at the one output.
+      val output = adds.head.path
+      assert(removes.forall { r =>
+        r.getTag(RemoveFile.Tags.COMPACTED_INTO)
+          .map(JsonUtils.fromJson[Seq[String]]).contains(Seq(output))
+      }, "each source must record the output it compacted into")
+
+      // (offset, physicalCount) for each source, in output order. No source DVs here, so
+      // physical == live.
+      val runs = removes
+        .map(r => compactionInfo(r).get.head)
+        .map(e => (e.rowOffsetInTarget.get, e.sourceNumPhysicalRecords.get))
+        .sortBy(_._1)
+      // The runs tile the output contiguously from offset 0, with no gaps or overlaps.
+      assert(runs.head._1 == 0L, s"first run must start at offset 0: $runs")
+      assert(runs(1)._1 == runs(0)._1 + runs(0)._2, s"runs are not contiguous: $runs")
+      assert(runs.map(_._2).sum == 5000L, s"runs must cover every output row: $runs")
+      assert(runs.map(_._2).toSet == Set(2000L, 3000L), s"unexpected run sizes: $runs")
+    }
+  }
+
+  test("a source without row-count stats fails the gate -> untagged tombstones (aborts as today)") {
+    withTempDir { dir =>
+      val path = dir.getCanonicalPath
+      // Write the sources with stats collection OFF, so their AddFiles carry no numLogicalRecords
+      // -- one of the trustworthiness conditions the capture gate requires.
+      withSQLConf(DeltaSQLConf.DELTA_COLLECT_STATS.key -> "false") {
+        spark.range(0, 2000).repartition(1).write.format("delta").mode("append").save(path)
+        spark.range(2000, 4000).repartition(1).write.format("delta").mode("append").save(path)
+      }
+      // Sanity: the sources indeed lack the stat the gate checks.
+      val deltaLog = DeltaLog.forTable(spark, path)
+      assert(deltaLog.update().allFiles.collect().forall(_.numLogicalRecords.isEmpty),
+        "sources must have no row-count stats for this test to exercise the gate")
+
+      withSQLConf(DeltaSQLConf.DELTA_OPTIMIZE_CONFLICT_RECONCILIATION_ENABLED.key -> "true") {
+        sql(s"OPTIMIZE delta.`$path`")
+      }
+
+      val actions = optimizeCommitActions(path)
+      val adds = actions.collect { case a: AddFile => a }
+      val removes = actions.collect { case r: RemoveFile => r }
+      assert(adds.size == 1 && removes.size == 2, "the sources are still compacted")
+      // Capture ran (reconcile was on) but the missing stats make it untrustworthy: the write side
+      // must fall back to plain untagged tombstones, so a losing DML aborts exactly as it does
+      // today -- never a bogus offset remap.
+      assert(removes.forall(_.getTag(RemoveFile.Tags.COMPACTED_INTO).isEmpty))
+      assert(removes.forall(_.getTag(RemoveFile.Tags.COMPACTION_INFO).isEmpty))
+      // The compaction itself is otherwise a normal OPTIMIZE: data is intact.
+      checkAnswer(spark.read.format("delta").load(path), (0 until 4000).map(i => Row(i.toLong)))
+    }
+  }
+
+  test("repartition OPTIMIZE writes no composition tags (capture is coalesce-only)") {
+    withTempDir { dir =>
+      val path = dir.getCanonicalPath
+      spark.range(0, 2000).repartition(1).write.format("delta").mode("append").save(path)
+      spark.range(2000, 4000).repartition(1).write.format("delta").mode("append").save(path)
+
+      // Reconcile is on, but the repartition compaction path shuffles rows into the output, so no
+      // source keeps a contiguous row range -- an offset composition would be meaningless (and a
+      // DV remapped by it would corrupt data). The capture gate excludes this path, so the sources
+      // must be removed with plain untagged tombstones, exactly as vanilla OPTIMIZE writes.
+      withSQLConf(
+          DeltaSQLConf.DELTA_OPTIMIZE_CONFLICT_RECONCILIATION_ENABLED.key -> "true",
+          DeltaSQLConf.DELTA_OPTIMIZE_REPARTITION_ENABLED.key -> "true") {
+        sql(s"OPTIMIZE delta.`$path`")
+      }
+
+      val actions = optimizeCommitActions(path)
+      val adds = actions.collect { case a: AddFile => a }
+      val removes = actions.collect { case r: RemoveFile => r }
+      assert(adds.size == 1 && removes.size == 2, "the sources are still compacted")
+      assert(removes.forall(_.getTag(RemoveFile.Tags.COMPACTED_INTO).isEmpty),
+        "the repartition path must not record where sources landed -- rows were shuffled")
+      assert(removes.forall(_.getTag(RemoveFile.Tags.COMPACTION_INFO).isEmpty),
+        "the repartition path must not record where sources landed -- rows were shuffled")
+      checkAnswer(spark.read.format("delta").load(path), (0 until 4000).map(i => Row(i.toLong)))
+    }
+  }
+
+  test("ZORDER OPTIMIZE writes no composition tags (rows are z-ordered, not contiguous)") {
+    withTempDir { dir =>
+      val path = dir.getCanonicalPath
+      spark.range(0, 2000).repartition(1).write.format("delta").mode("append").save(path)
+      spark.range(2000, 4000).repartition(1).write.format("delta").mode("append").save(path)
+
+      // A ZORDER pass reorders rows onto a space-filling curve, so no source keeps a contiguous
+      // row range and an offset composition would be meaningless. The capture gate excludes the
+      // multi-dimensional-clustering path, so the sources must be removed with plain untagged
+      // tombstones.
+      withSQLConf(
+          DeltaSQLConf.DELTA_OPTIMIZE_CONFLICT_RECONCILIATION_ENABLED.key -> "true",
+          DeltaSQLConf.DELTA_OPTIMIZE_ZORDER_COL_STAT_CHECK.key -> "false") {
+        sql(s"OPTIMIZE delta.`$path` ZORDER BY (id)")
+      }
+
+      val actions = optimizeCommitActions(path)
+      val removes = actions.collect { case r: RemoveFile => r }
+      assert(actions.exists(_.isInstanceOf[AddFile]) && removes.nonEmpty,
+        "the z-order must rewrite files")
+      assert(removes.forall(_.getTag(RemoveFile.Tags.COMPACTED_INTO).isEmpty),
+        "z-order must not record a row-range composition -- rows were permuted")
+      assert(removes.forall(_.getTag(RemoveFile.Tags.COMPACTION_INFO).isEmpty),
+        "z-order must not record a row-range composition -- rows were permuted")
+      checkAnswer(spark.read.format("delta").load(path), (0 until 4000).map(i => Row(i.toLong)))
+    }
+  }
+
+  test("CLUSTER BY OPTIMIZE writes no composition tags (clustering permutes rows)") {
+    withTable("clustered_optimize_src") {
+      withTempDir { dir =>
+        val path = dir.getCanonicalPath
+        sql(s"CREATE TABLE clustered_optimize_src (id LONG) USING delta " +
+          s"CLUSTER BY (id) LOCATION '$path'")
+        sql("INSERT INTO clustered_optimize_src SELECT id FROM range(0, 2000)")
+        sql("INSERT INTO clustered_optimize_src SELECT id FROM range(2000, 4000)")
+
+        // A clustering pass reorders rows into ZCubes, so no source keeps a contiguous row range.
+        // Same gate as ZORDER (isMultiDimClustering): the sources must be removed with plain
+        // untagged tombstones.
+        withSQLConf(DeltaSQLConf.DELTA_OPTIMIZE_CONFLICT_RECONCILIATION_ENABLED.key -> "true") {
+          sql("OPTIMIZE clustered_optimize_src")
+        }
+
+        val actions = optimizeCommitActions(path)
+        val removes = actions.collect { case r: RemoveFile => r }
+        assert(actions.exists(_.isInstanceOf[AddFile]) && removes.nonEmpty,
+          "the clustering pass must rewrite files")
+        assert(removes.forall(_.getTag(RemoveFile.Tags.COMPACTED_INTO).isEmpty),
+          "clustering must not record a row-range composition -- rows were permuted")
+        assert(removes.forall(_.getTag(RemoveFile.Tags.COMPACTION_INFO).isEmpty),
+          "clustering must not record a row-range composition -- rows were permuted")
+        checkAnswer(spark.table("clustered_optimize_src"), (0 until 4000).map(i => Row(i.toLong)))
+      }
+    }
+  }
+
+  /** Actions committed by the single OPTIMIZE at the table's current (latest) version. */
+  private def optimizeCommitActions(path: String): Seq[Action] = {
+    val deltaLog = DeltaLog.forTable(spark, path)
+    val optimizeVersion = deltaLog.update().version
+    deltaLog.getChanges(startVersion = optimizeVersion, catalogTableOpt = None).next()._2
+  }
+
+  /** The compaction composition recorded on a source tombstone, if it was tagged. */
+  private def compactionInfo(r: RemoveFile): Option[Seq[CompactionInfoEntry]] =
+    r.getTag(RemoveFile.Tags.COMPACTION_INFO).map(JsonUtils.fromJson[Seq[CompactionInfoEntry]])
+
+  /** Number of Parquet row groups in each data file physically present under the table dir. */
+  private def rowGroupCountsPerFile(dir: File): Seq[Int] = {
+    // scalastyle:off deltahadoopconfiguration
+    val conf = spark.sessionState.newHadoopConf()
+    // scalastyle:on deltahadoopconfiguration
+    dir.listFiles().filter(_.getName.endsWith(".parquet")).toSeq.map { f =>
+      val input = HadoopInputFile.fromPath(new Path(f.getAbsolutePath), conf)
+      val reader = ParquetFileReader.open(input)
+      try reader.getRowGroups.size() finally reader.close()
+    }
+  }
+}
+
+/**
+ * A leaf that emits one columnar batch per `(file, rowCount)` spec, setting the scan's thread-local
+ * file identity before each batch just as a real file scan does. One partition, in spec order.
+ */
+private case class FakeColumnarScan(specs: Seq[(String, Int)]) extends LeafExecNode {
+  override def output: Seq[Attribute] = Seq(AttributeReference("v", IntegerType)())
+
+  override def supportsColumnar: Boolean = true
+
+  override protected def doExecute(): RDD[InternalRow] =
+    throw new UnsupportedOperationException("columnar only")
+
+  override protected def doExecuteColumnar(): RDD[ColumnarBatch] = {
+    val specsLocal = specs
+    sparkContext.parallelize(Seq(specsLocal), numSlices = 1).flatMap { batchSpecs =>
+      batchSpecs.iterator.map { case (file, n) =>
+        InputFileBlockHolder.set(file, 0L, (n * 4).toLong)
+        val vec = new OnHeapColumnVector(math.max(n, 1), IntegerType)
+        var i = 0
+        while (i < n) {
+          vec.putInt(i, 1)
+          i += 1
+        }
+        new ColumnarBatch(Array[ColumnVector](vec), n)
+      }
+    }
+  }
+}
+
+/** A leaf that supports only the row path (`supportsColumnar` defaults to false). */
+private case class FakeRowScan() extends LeafExecNode {
+  override def output: Seq[Attribute] = Seq(AttributeReference("v", IntegerType)())
+
+  override protected def doExecute(): RDD[InternalRow] = sparkContext.emptyRDD[InternalRow]
+}