From 05acb4151bb28f67a99ec140a072b14ca0a25ebb Mon Sep 17 00:00:00 2001 From: Eunjin Song Date: Mon, 3 Aug 2026 23:32:34 -0700 Subject: [PATCH 1/6] [Spark] Capture and persist compaction OPTIMIZE source composition When OPTIMIZE conflict reconciliation is enabled, record on each removed source file's tombstone where that source's live rows landed in the compacted output, so a later conflict check can remap a deletion vector between the source and the output instead of aborting. - SourceCompositionCaptureExec: a write-stage operator that observes, per output partition, each source file's contiguous run of rows via the input-file identity and a per-file row count (row and vectorized execution paths), accumulating (sourceFile, outputStart, liveCount) runs without imposing a sort. - The OptimizeExecutor coalesce/compaction path injects the capture (compaction only: repartition and a clustering pass permute rows, so no contiguous offset mapping exists) and writes the composition as the RemoveFile compactedInto / compactionInfo tombstone tags. - RemoveFile.Tags.COMPACTED_INTO / COMPACTION_INFO + CompactionInfoEntry define the on-disk tag format (modeled on what Databricks Runtime records; cross-engine interop is best-effort, not a verified guarantee). Persisted, not stripped: an ignored tag only forgoes a reconcile, never affects correctness. This is the capture/persist layer only; the conflict-time reconcile that consumes the tags lands in a follow-up. Co-Authored-By: Claude Opus 4.8 --- .../spark/sql/delta/actions/actions.scala | 50 +++++ .../delta/commands/OptimizeTableCommand.scala | 176 ++++++++++++++-- .../delta/files/DeltaFileFormatWriter.scala | 4 +- .../files/SourceCompositionCaptureExec.scala | 188 ++++++++++++++++++ .../sql/delta/files/TransactionalWrite.scala | 40 +++- .../sql/delta/sources/DeltaSQLConf.scala | 16 ++ .../SourceCompositionCaptureExecSuite.scala | 94 +++++++++ 7 files changed, 543 insertions(+), 25 deletions(-) create mode 100644 spark/src/main/scala/org/apache/spark/sql/delta/files/SourceCompositionCaptureExec.scala create mode 100644 spark/src/test/scala/org/apache/spark/sql/delta/files/SourceCompositionCaptureExecSuite.scala 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, + * `[".parquet"]` (matching the AddFile.path in the same commit). + * - [[COMPACTION_INFO]]: JSON array holding the single run this source contributed, + * `[{"rowOffsetInTarget": , "sourceNumPhysicalRecords": }]`. 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..ace43113f64 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 @@ -23,14 +23,15 @@ import scala.collection.mutable.ArrayBuffer 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 @@ -43,6 +44,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 +521,58 @@ 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 + + // When capturing the source composition (coalesce compaction path), read each source file + // 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 gate and aborts -- never wrong data.) Both confs go on a + // CLONED session so the override is isolated from other queries sharing 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 we restore the previous active session. + var input = if (captureReconcile) { + 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) + } + } 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,13 +582,8 @@ class OptimizeExecutor( clusteringColumns, optimizeStrategy.curve) } else { - val useRepartition = sparkSession.sessionState.conf.getConf( - DeltaSQLConf.DELTA_OPTIMIZE_REPARTITION_ENABLED) - if (useRepartition) { - input.repartition(numPartitions = 1) - } else { - input.coalesce(numPartitions = 1) - } + if (useRepartition) input.repartition(numPartitions = 1) + else input.coalesce(numPartitions = 1) } val partitionDesc = partition.toSeq.map(entry => entry._1 + "=" + entry._2).mkString(",") @@ -549,18 +595,120 @@ 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)) + // Record each removed source's placement in the compacted output (only when reconciliation is + // enabled and the capture is trustworthy) so the conflict checker can remap a concurrent DML + // deletion vector by offset instead of aborting; see buildCompactionCompositionTags. Empty when + // capture was off or the gate rejected it -- then the sources below are tombstoned untagged and + // the loser aborts as it does today. + val srcCompositionTag: Map[String, Map[String, String]] = + buildCompactionCompositionTags(txn, bin, addFiles, captureAccOpt) + + // Fast path: no composition tag (capture off, or the gate above rejected it) -> build the + // RemoveFiles exactly as vanilla OPTIMIZE does, with no per-file tag lookup. + val removeFiles = if (srcCompositionTag.isEmpty) { + bin.map(_.removeWithTimestamp(operationTimestamp, dataChange = false)) + } else { + bin.map { f => + val r = f.removeWithTimestamp(operationTimestamp, dataChange = false) + srcCompositionTag.get(f.path) match { + case Some(tags) => + // Persist the composition (as Databricks Runtime does on every compaction OPTIMIZE): it + // is consumed in-memory when THIS OPTIMIZE loses to a concurrent DML, and read from the + // committed tombstone when a concurrent DML LOSES to this OPTIMIZE. + tags.foldLeft(r) { case (tagged, (k, v)) => tagged.copyWithTag(k, v) } + case None => r + } + } + } val updates = addFiles ++ removeFiles updates } + /** + * Build the `compactedInto` / `compactionInfo` composition tags for a compaction OPTIMIZE's + * removed sources: a map from each source's table-relative AddFile path to the tag pair recording + * where that source's rows landed in the single compacted output. Written on the source's + * tombstone (see [[RemoveFile.Tags.COMPACTION_INFO]]) so a concurrent DML's deletion vector + * can be remapped by offset instead of aborting; the value format matches Databricks Runtime, + * so the two engines can reconcile against each other on a shared table. + * + * Returns empty -- so the caller tombstones every source untagged and the conflict falls back to + * today's abort -- 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 buildCompactionCompositionTags( + txn: OptimisticTransaction, + bin: Seq[AddFile], + addFiles: Seq[AddFile], + captureAccOpt: Option[SourceCompositionAccumulator]): Map[String, Map[String, String]] = { + 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 outputPath = addFiles.head.path + val compactedIntoJson = JsonUtils.toJson(Seq(outputPath)) + // Walk runs in output (write) order, accumulating each source's start offset in the output. + var outputPos = 0L + val perFile: 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 && perFile.forall(_.isDefined) && oneRunPerFile) { + perFile.map(_.get).toMap + } else { + Map.empty[String, Map[String, String]] + } + case _ => + Map.empty[String, Map[String, String]] + } + } + /** * 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..5168709e78f 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 @@ -105,7 +105,8 @@ trait TransactionalWrite extends DeltaLogging { self: OptimisticTransactionImpl protected def normalizeData( deltaLog: DeltaLog, options: Option[DeltaOptions], - data: DataFrame): (QueryExecution, Seq[Attribute], Seq[Constraint], Set[String]) = { + data: DataFrame) + : (QueryExecution, Seq[Attribute], Seq[Constraint], Set[String]) = { val (normalizedSchema, output, constraints, trackHighWaterMarks) = normalizeSchema( deltaLog, options, data) @@ -126,7 +127,8 @@ trait TransactionalWrite extends DeltaLogging { self: OptimisticTransactionImpl protected def normalizeSchema( deltaLog: DeltaLog, options: Option[DeltaOptions], - data: DataFrame): (DataFrame, Seq[Attribute], Seq[Constraint], Set[String]) = { + data: DataFrame) + : (DataFrame, Seq[Attribute], Seq[Constraint], Set[String]) = { val normalizedData = SchemaUtils.normalizeColumnNames( deltaLog, metadata.schema, data ) @@ -407,7 +409,11 @@ trait TransactionalWrite extends DeltaLogging { self: OptimisticTransactionImpl inputData: Dataset[_], writeOptions: Option[DeltaOptions], isOptimize: Boolean, - additionalConstraints: Seq[Constraint]): Seq[FileAction] = { + additionalConstraints: Seq[Constraint], + // OPTIMIZE compaction conflict-reconciliation: when set, a SourceCompositionCaptureExec is + // injected to observe the output's source composition (file identity + per-file row count) + // into this accumulator. No helper columns are added; rows pass through unchanged. + sourceCompositionCapture: Option[SourceCompositionAccumulator] = None): Seq[FileAction] = { hasWritten = true val spark = inputData.sparkSession @@ -416,21 +422,24 @@ trait TransactionalWrite extends DeltaLogging { self: OptimisticTransactionImpl val (queryExecution, output, generatedColumnConstraints, trackFromData) = normalizeData(deltaLog, writeOptions, data) + // The capture path adds no helper columns; rows pass through unchanged, so the write output is + // exactly the normalized output. + val writeOutput = output // Use the track set from the transaction if set, // otherwise use the track set from `normalizeData()`. val trackIdentityHighWaterMarks = trackHighWaterMarks.getOrElse(trackFromData) - val partitioningColumns = getPartitioningColumns(partitionSchema, output) + val partitioningColumns = getPartitioningColumns(partitionSchema, writeOutput) val committer = getCommitter(outputPath) - val (statsDataSchema, _) = getStatsSchema(output, partitionSchema) + val (statsDataSchema, _) = getStatsSchema(writeOutput, partitionSchema) // If Statistics Collection is enabled, then create a stats tracker that will be injected during // the FileFormatWriter.write call below and will collect per-file stats using // StatisticsCollection - val (optionalStatsTracker, _) = getOptionalStatsTrackerAndStatsCollection(output, outputPath, - partitionSchema, data) + val (optionalStatsTracker, _) = getOptionalStatsTrackerAndStatsCollection( + writeOutput, outputPath, partitionSchema, data) val constraints = @@ -450,19 +459,32 @@ trait TransactionalWrite extends DeltaLogging { self: OptimisticTransactionImpl val outputSpec = FileFormatWriter.OutputSpec( outputPath.toString, Map.empty, - output) + writeOutput) val empty2NullPlan = convertEmptyToNullIfNeeded(queryExecution.executedPlan, partitioningColumns, constraints) 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..ebfafe9e02f --- /dev/null +++ b/spark/src/test/scala/org/apache/spark/sql/delta/files/SourceCompositionCaptureExecSuite.scala @@ -0,0 +1,94 @@ +/* + * 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.jdk.CollectionConverters._ + +import org.apache.spark.rdd.{InputFileBlockHolder, RDD} +import org.apache.spark.sql.QueryTest +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.test.SharedSparkSession +import org.apache.spark.sql.types.IntegerType +import org.apache.spark.sql.vectorized.{ColumnarBatch, ColumnVector} + +/** + * Unit tests for [[SourceCompositionCaptureExec]]'s columnar execution path. The row path is + * covered end to end by `OptimizeConflictReconciliationSuite`; here the columnar path is exercised + * directly with a stub columnar child, since a columnar execution backend is not available in the + * OSS test harness. Both paths fold units into runs through the same `RunTracker`. + */ +class SourceCompositionCaptureExecSuite extends QueryTest with SharedSparkSession { + + 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("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) + } +} + +/** + * 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] +} From a6820fa12c945f1b998af51813eed140132e6eb9 Mon Sep 17 00:00:00 2001 From: Eunjin Song Date: Tue, 4 Aug 2026 08:31:29 -0700 Subject: [PATCH 2/6] [RLC] Capture path review polish: helper returns RemoveFiles, writeFiles overload Address review feedback on the OPTIMIZE source-composition capture/write path: - buildRemoveFilesWithCompactionCompositionTags now returns the tagged Seq[RemoveFile] directly (inner `untagged` fallback), dropping the Map.empty fast/slow-branch plumbing at the call site. - Keep the public 4-arg writeFiles signature undisturbed; add a 5-arg overload (no default arg) carrying sourceCompositionCapture and the SourceCompositionCaptureExec injection. - Drop the vestigial `val writeOutput = output` alias; revert normalizeData/normalizeSchema signatures to their original single-line form. - Add a columnar-path unit test for interleaved source batches (the "mixed" shape the one-run-per-file gate rejects). Co-Authored-By: Claude Opus 4.8 --- .../delta/commands/OptimizeTableCommand.scala | 81 +++++++++---------- .../sql/delta/files/TransactionalWrite.scala | 39 +++++---- .../SourceCompositionCaptureExecSuite.scala | 15 ++++ 3 files changed, 75 insertions(+), 60 deletions(-) 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 ace43113f64..29829f7130c 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 @@ -603,47 +603,30 @@ class OptimizeExecutor( s"Unexpected action $other with type ${other.getClass}. File compaction job output" + s"should only have AddFiles") } - // Record each removed source's placement in the compacted output (only when reconciliation is - // enabled and the capture is trustworthy) so the conflict checker can remap a concurrent DML - // deletion vector by offset instead of aborting; see buildCompactionCompositionTags. Empty when - // capture was off or the gate rejected it -- then the sources below are tombstoned untagged and - // the loser aborts as it does today. - val srcCompositionTag: Map[String, Map[String, String]] = - buildCompactionCompositionTags(txn, bin, addFiles, captureAccOpt) - - // Fast path: no composition tag (capture off, or the gate above rejected it) -> build the - // RemoveFiles exactly as vanilla OPTIMIZE does, with no per-file tag lookup. - val removeFiles = if (srcCompositionTag.isEmpty) { - bin.map(_.removeWithTimestamp(operationTimestamp, dataChange = false)) - } else { - bin.map { f => - val r = f.removeWithTimestamp(operationTimestamp, dataChange = false) - srcCompositionTag.get(f.path) match { - case Some(tags) => - // Persist the composition (as Databricks Runtime does on every compaction OPTIMIZE): it - // is consumed in-memory when THIS OPTIMIZE loses to a concurrent DML, and read from the - // committed tombstone when a concurrent DML LOSES to this OPTIMIZE. - tags.foldLeft(r) { case (tagged, (k, v)) => tagged.copyWithTag(k, v) } - case None => r - } - } - } + // Build the removed-source tombstones, tagging each with where its rows landed in the compacted + // output when reconciliation is enabled and the capture is trustworthy, so the conflict checker + // can remap a concurrent DML deletion vector by offset instead of aborting; otherwise plain + // untagged tombstones as vanilla OPTIMIZE writes (see the helper below). + val removeFiles = buildRemoveFilesWithCompactionCompositionTags( + txn, bin, addFiles, captureAccOpt, operationTimestamp) val updates = addFiles ++ removeFiles updates } /** - * Build the `compactedInto` / `compactionInfo` composition tags for a compaction OPTIMIZE's - * removed sources: a map from each source's table-relative AddFile path to the tag pair recording - * where that source's rows landed in the single compacted output. Written on the source's - * tombstone (see [[RemoveFile.Tags.COMPACTION_INFO]]) so a concurrent DML's deletion vector - * can be remapped by offset instead of aborting; the value format matches Databricks Runtime, - * so the two engines can reconcile against each other on a shared table. + * 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. * - * Returns empty -- so the caller tombstones every source untagged and the conflict falls back to - * today's abort -- 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). + * 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 @@ -651,11 +634,17 @@ class OptimizeExecutor( * length by subtracting the tombstone's own DV, and rebuilds the read-time gaps only on a real * conflict. */ - private def buildCompactionCompositionTags( + private def buildRemoveFilesWithCompactionCompositionTags( txn: OptimisticTransaction, bin: Seq[AddFile], addFiles: Seq[AddFile], - captureAccOpt: Option[SourceCompositionAccumulator]): Map[String, Map[String, String]] = { + 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)) + captureAccOpt match { case Some(acc) if addFiles.size == 1 && acc.value.size() == 1 && bin.forall(_.numLogicalRecords.isDefined) => @@ -667,11 +656,10 @@ class OptimizeExecutor( // conflict-resolution time. val nameToAddFile = generateCandidateFileMap(txn.deltaLog.dataPath, bin) val tablePath = txn.deltaLog.dataPath - val outputPath = addFiles.head.path - val compactedIntoJson = JsonUtils.toJson(Seq(outputPath)) + 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 perFile: Seq[Option[(String, Map[String, String])]] = (0 until runs.size()).map { i => + 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 @@ -699,13 +687,18 @@ class OptimizeExecutor( // 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 && perFile.forall(_.isDefined) && oneRunPerFile) { - perFile.map(_.get).toMap + 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 { - Map.empty[String, Map[String, String]] + untagged } case _ => - Map.empty[String, Map[String, String]] + untagged } } 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 5168709e78f..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 @@ -105,8 +105,7 @@ trait TransactionalWrite extends DeltaLogging { self: OptimisticTransactionImpl protected def normalizeData( deltaLog: DeltaLog, options: Option[DeltaOptions], - data: DataFrame) - : (QueryExecution, Seq[Attribute], Seq[Constraint], Set[String]) = { + data: DataFrame): (QueryExecution, Seq[Attribute], Seq[Constraint], Set[String]) = { val (normalizedSchema, output, constraints, trackHighWaterMarks) = normalizeSchema( deltaLog, options, data) @@ -127,8 +126,7 @@ trait TransactionalWrite extends DeltaLogging { self: OptimisticTransactionImpl protected def normalizeSchema( deltaLog: DeltaLog, options: Option[DeltaOptions], - data: DataFrame) - : (DataFrame, Seq[Attribute], Seq[Constraint], Set[String]) = { + data: DataFrame): (DataFrame, Seq[Attribute], Seq[Constraint], Set[String]) = { val normalizedData = SchemaUtils.normalizeColumnNames( deltaLog, metadata.schema, data ) @@ -405,15 +403,27 @@ trait TransactionalWrite extends DeltaLogging { self: OptimisticTransactionImpl * @param isOptimize Whether the operation writing this is Optimize or not. * @param additionalConstraints Additional constraints on the write. */ + def writeFiles( + inputData: Dataset[_], + writeOptions: Option[DeltaOptions], + isOptimize: Boolean, + 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], - // OPTIMIZE compaction conflict-reconciliation: when set, a SourceCompositionCaptureExec is - // injected to observe the output's source composition (file identity + per-file row count) - // into this accumulator. No helper columns are added; rows pass through unchanged. - sourceCompositionCapture: Option[SourceCompositionAccumulator] = None): Seq[FileAction] = { + sourceCompositionCapture: Option[SourceCompositionAccumulator]): Seq[FileAction] = { hasWritten = true val spark = inputData.sparkSession @@ -422,24 +432,21 @@ trait TransactionalWrite extends DeltaLogging { self: OptimisticTransactionImpl val (queryExecution, output, generatedColumnConstraints, trackFromData) = normalizeData(deltaLog, writeOptions, data) - // The capture path adds no helper columns; rows pass through unchanged, so the write output is - // exactly the normalized output. - val writeOutput = output // Use the track set from the transaction if set, // otherwise use the track set from `normalizeData()`. val trackIdentityHighWaterMarks = trackHighWaterMarks.getOrElse(trackFromData) - val partitioningColumns = getPartitioningColumns(partitionSchema, writeOutput) + val partitioningColumns = getPartitioningColumns(partitionSchema, output) val committer = getCommitter(outputPath) - val (statsDataSchema, _) = getStatsSchema(writeOutput, partitionSchema) + val (statsDataSchema, _) = getStatsSchema(output, partitionSchema) // If Statistics Collection is enabled, then create a stats tracker that will be injected during // the FileFormatWriter.write call below and will collect per-file stats using // StatisticsCollection - val (optionalStatsTracker, _) = getOptionalStatsTrackerAndStatsCollection( - writeOutput, outputPath, partitionSchema, data) + val (optionalStatsTracker, _) = getOptionalStatsTrackerAndStatsCollection(output, outputPath, + partitionSchema, data) val constraints = @@ -459,7 +466,7 @@ trait TransactionalWrite extends DeltaLogging { self: OptimisticTransactionImpl val outputSpec = FileFormatWriter.OutputSpec( outputPath.toString, Map.empty, - writeOutput) + output) val empty2NullPlan = convertEmptyToNullIfNeeded(queryExecution.executedPlan, partitioningColumns, constraints) 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 index ebfafe9e02f..72ff542f41c 100644 --- 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 @@ -50,6 +50,21 @@ class SourceCompositionCaptureExecSuite extends QueryTest with SharedSparkSessio 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) From fc28df6dad98892e8386ea2c6e5d3cf1c48e1c49 Mon Sep 17 00:00:00 2001 From: Eunjin Song Date: Tue, 4 Aug 2026 09:57:04 -0700 Subject: [PATCH 3/6] [RLC] Extract capture-path read pinning into a helper; add e2e capture tests Extract the reconcile-capture read (clone session, pin FILES_MAX_PARTITION_BYTES to the compaction target + FILES_MIN_PARTITION_NUM=1, run createDataFrame under the pinned active session) into readCompactionSourceWithWholeFilePins so the vanilla OPTIMIZE route keeps its plain createDataFrame inline. The non-RLC tombstone branch likewise skips the reconcile helper and writes plain untagged RemoveFiles. Strengthen SourceCompositionCaptureExecSuite (now on DeltaSQLCommandTest) with two end-to-end tests over a real compaction OPTIMIZE: - multi-row-group sources under a hostile ambient split size still yield contiguous per-source compactedInto/compactionInfo tags tiling the output from offset 0; - a source with no row-count stats fails the trustworthiness gate and falls back to plain untagged tombstones (a losing DML aborts as today), data intact. Co-Authored-By: Claude Opus 4.8 --- .../delta/commands/OptimizeTableCommand.scala | 84 ++++++---- .../SourceCompositionCaptureExecSuite.scala | 145 +++++++++++++++++- 2 files changed, 191 insertions(+), 38 deletions(-) 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 29829f7130c..6404a0e39f0 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 @@ -36,7 +36,7 @@ import org.apache.spark.sql.delta.util.{BinPackingUtils, DeltaFileOperations, Js 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} @@ -536,29 +536,11 @@ class OptimizeExecutor( val captureReconcile = reconcileEnabled && !isMultiDimClustering && !useRepartition - // When capturing the source composition (coalesce compaction path), read each source file - // 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 gate and aborts -- never wrong data.) Both confs go on a - // CLONED session so the override is isolated from other queries sharing 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 we restore the previous active session. + // 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) { - 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) - } + readCompactionSourceWithWholeFilePins(txn, bin, maxFileSize) } else { txn.deltaLog.createDataFrame(txn.snapshot, bin, actionTypeOpt = Some("Optimize")) } @@ -582,8 +564,11 @@ class OptimizeExecutor( clusteringColumns, optimizeStrategy.curve) } else { - if (useRepartition) input.repartition(numPartitions = 1) - else input.coalesce(numPartitions = 1) + if (useRepartition) { + input.repartition(numPartitions = 1) + } else { + input.coalesce(numPartitions = 1) + } } val partitionDesc = partition.toSeq.map(entry => entry._1 + "=" + entry._2).mkString(",") @@ -603,16 +588,53 @@ class OptimizeExecutor( s"Unexpected action $other with type ${other.getClass}. File compaction job output" + s"should only have AddFiles") } - // Build the removed-source tombstones, tagging each with where its rows landed in the compacted - // output when reconciliation is enabled and the capture is trustworthy, so the conflict checker - // can remap a concurrent DML deletion vector by offset instead of aborting; otherwise plain - // untagged tombstones as vanilla OPTIMIZE writes (see the helper below). - val removeFiles = buildRemoveFilesWithCompactionCompositionTags( - txn, bin, addFiles, captureAccOpt, operationTimestamp) + // 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 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 index 72ff542f41c..072c1593fbb 100644 --- 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 @@ -16,25 +16,40 @@ 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 +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.test.SharedSparkSession +import org.apache.spark.sql.internal.SQLConf import org.apache.spark.sql.types.IntegerType import org.apache.spark.sql.vectorized.{ColumnarBatch, ColumnVector} /** - * Unit tests for [[SourceCompositionCaptureExec]]'s columnar execution path. The row path is - * covered end to end by `OptimizeConflictReconciliationSuite`; here the columnar path is exercised - * directly with a stub columnar child, since a columnar execution backend is not available in the - * OSS test harness. Both paths fold units into runs through the same `RunTracker`. + * 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 SharedSparkSession { +class SourceCompositionCaptureExecSuite extends QueryTest with DeltaSQLCommandTest { test("columnar path folds one run per source file across batches, in write order") { val acc = new SourceCompositionAccumulator @@ -70,6 +85,122 @@ class SourceCompositionCaptureExecSuite extends QueryTest with SharedSparkSessio 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 (512 bytes) breaks each multi-row-group source into many + // per-row-group splits before the read. The capture path pins the read against exactly this + // -- it clones the session and sets maxPartitionBytes to the compaction target and + // minPartitionNum = 1, so each source is read whole in one partition rather than as scattered + // splits -- and the tags below still come out contiguous per source. + withSQLConf( + DeltaSQLConf.DELTA_OPTIMIZE_CONFLICT_RECONCILIATION_ENABLED.key -> "true", + SQLConf.FILES_MAX_PARTITION_BYTES.key -> "512") { + 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))) + } + } + + /** 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() + } + } } /** From f984c7b740857fee8f99d87959050784401e566d Mon Sep 17 00:00:00 2001 From: Eunjin Song Date: Tue, 4 Aug 2026 10:33:57 -0700 Subject: [PATCH 4/6] [RLC] Test that the repartition OPTIMIZE path writes no composition tags The capture gate excludes the repartition compaction path (useRepartition): repartition(1) shuffles rows into the output, so no source keeps a contiguous row range and an offset composition would be meaningless -- a DV remapped by it would corrupt data. Add a regression test asserting that with reconcile on and optimize.repartition.enabled on, the removed sources carry plain untagged tombstones (no compactedInto / compactionInfo), and the data is intact. Co-Authored-By: Claude Opus 4.8 --- .../SourceCompositionCaptureExecSuite.scala | 28 +++++++++++++++++++ 1 file changed, 28 insertions(+) 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 index 072c1593fbb..81d377bc021 100644 --- 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 @@ -179,6 +179,34 @@ class SourceCompositionCaptureExecSuite extends QueryTest with DeltaSQLCommandTe } } + 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))) + } + } + /** 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) From 657bbce8976abe8b04f9d65802bde947b7d2c5f2 Mon Sep 17 00:00:00 2001 From: Eunjin Song Date: Tue, 4 Aug 2026 11:05:01 -0700 Subject: [PATCH 5/6] test: guard read-pins via interleaving split size; add zorder/cluster-by no-tag tests Make the multi-row-group capture test a genuine regression guard for the read pins in readCompactionSourceWithWholeFilePins. It now sets an ambient FILES_MAX_PARTITION_BYTES (8 KiB) that would break each source into a full split plus a row-bearing remainder and interleave the sources under coalesce(1)'s descending-length packing. With the pins present the command overrides this to the compaction target so each source reads whole and stays one contiguous run (tags written); drop the pins and the sources interleave, the one-run-per-file gate declines, and the test fails. (The earlier 512 B value never interleaved -- many equal splits plus a footer-sized remainder -- so it could not catch pin removal.) Also add two isMultiDimClustering-path contract tests asserting that ZORDER and CLUSTER BY OPTIMIZE write no composition tags, since a clustering/z-order pass permutes rows and no contiguous offset composition exists. Co-Authored-By: Claude Opus 4.8 --- .../SourceCompositionCaptureExecSuite.scala | 73 +++++++++++++++++-- 1 file changed, 67 insertions(+), 6 deletions(-) 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 index 81d377bc021..f1cd48f3199 100644 --- 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 @@ -109,14 +109,18 @@ class SourceCompositionCaptureExecSuite extends QueryTest with DeltaSQLCommandTe assert(rowGroups.size == 2 && rowGroups.forall(_ > 1), s"expected two multi-row-group sources, got $rowGroups") - // A hostile ambient split size (512 bytes) breaks each multi-row-group source into many - // per-row-group splits before the read. The capture path pins the read against exactly this - // -- it clones the session and sets maxPartitionBytes to the compaction target and - // minPartitionNum = 1, so each source is read whole in one partition rather than as scattered - // splits -- and the tags below still come out contiguous per source. + // 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 -> "512") { + SQLConf.FILES_MAX_PARTITION_BYTES.key -> "8192") { sql(s"OPTIMIZE delta.`$path`") } @@ -207,6 +211,63 @@ class SourceCompositionCaptureExecSuite extends QueryTest with DeltaSQLCommandTe } } + 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) From 3a81d9b51e1aacc88a0604416b8ae82f970c9be5 Mon Sep 17 00:00:00 2001 From: Eunjin Song Date: Fri, 7 Aug 2026 22:39:09 -0700 Subject: [PATCH 6/6] [RLC] Fail-safe capture: fall back to untagged tombstones on any error buildRemoveFilesWithCompactionCompositionTags could throw while building the composition tags (an unmappable source path, a JSON serialization error) after the OPTIMIZE output was already written, turning a best-effort optimization enabler into a hard failure of the OPTIMIZE commit. Wrap the whole tag-building match in try/catch(NonFatal): on any failure fall back to plain untagged tombstones -- exactly what vanilla OPTIMIZE writes -- so the already-written OPTIMIZE still commits and a concurrent loser aborts exactly as it does today. This upholds the method's own documented contract that capture is never required for OPTIMIZE correctness. Co-Authored-By: Claude Opus 4.8 --- .../sql/delta/commands/OptimizeTableCommand.scala | 11 ++++++++++- 1 file changed, 10 insertions(+), 1 deletion(-) 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 6404a0e39f0..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,6 +19,7 @@ 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} @@ -667,7 +668,7 @@ class OptimizeExecutor( def untagged: Seq[RemoveFile] = bin.map(_.removeWithTimestamp(operationTimestamp, dataChange = false)) - captureAccOpt match { + try captureAccOpt match { case Some(acc) if addFiles.size == 1 && acc.value.size() == 1 && bin.forall(_.numLogicalRecords.isDefined) => val runs = acc.value.get(0) @@ -721,6 +722,14 @@ class OptimizeExecutor( } 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 } }