From a2ad862ccde0c9586d0f2a9423e27a5ef38f7159 Mon Sep 17 00:00:00 2001 From: Eunjin Song Date: Wed, 8 Jul 2026 13:07:33 -0700 Subject: [PATCH 1/4] [Spark] Reader-side conflict-time data skipping to reduce unnecessary conflicts Conflict detection filters candidate files by partition only; for unpartitioned (incl. liquid-clustered) tables the append check conflicts on any concurrently-added file. This adds column-statistics data skipping: added files whose stats prove they cannot match the current transaction's read predicates are excluded, so operations touching disjoint data ranges no longer falsely conflict. - DataSkippingReaderBase.buildDataSkippingPredicate + filterFilesByDataSkipping reuse the reader's DataFiltersBuilder / verifyStatsForFilter and the shared parseAndDecodeStats, applying 'expr || !verifyStatsForFilter(...)' for one-way safety (skip only when stats prove no match; missing stats keep the file). - ConflictChecker.getFirstFileMatchingPartitionPredicates applies skipping per read predicate and unions survivors (OR across reads; a read's own data predicates are ANDed), and never skips for a whole-table read. Emits a delta.conflictDetection.dataSkipping.filesSkipped event. - Gated by spark.databricks.delta.conflictDetection.dataSkipping.enabled (internal, default off). - Adds ConflictDataSkippingSuite (disjoint / overlapping / disabled / missing-stats / partitioned). Part of #3. Implements #4. Co-Authored-By: Claude Opus 4.8 --- .../spark/sql/delta/ConflictChecker.scala | 31 +++- .../sql/delta/sources/DeltaSQLConf.scala | 11 ++ .../sql/delta/stats/DataSkippingReader.scala | 86 ++++++++- .../sql/delta/ConflictDataSkippingSuite.scala | 171 ++++++++++++++++++ 4 files changed, 292 insertions(+), 7 deletions(-) create mode 100644 spark/src/test/scala/org/apache/spark/sql/delta/ConflictDataSkippingSuite.scala diff --git a/spark/src/main/scala/org/apache/spark/sql/delta/ConflictChecker.scala b/spark/src/main/scala/org/apache/spark/sql/delta/ConflictChecker.scala index d5cd7be1beb..b92392e0c87 100644 --- a/spark/src/main/scala/org/apache/spark/sql/delta/ConflictChecker.scala +++ b/spark/src/main/scala/org/apache/spark/sql/delta/ConflictChecker.scala @@ -994,14 +994,41 @@ private[delta] class ConflictChecker( return None } + // Reader-side data skipping: narrow the candidate files using the read snapshot's column-stats + // skipping. A concurrently-added file whose stats prove it cannot match the read predicates is + // not a conflict -- this is what lets unpartitioned / liquid-clustered tables avoid conflicting + // on every added file below. Data predicates are applied *per read predicate* and the survivors + // unioned (OR across reads), because independent reads have OR, not AND, semantics: a file is a + // candidate if it could match ANY read; it is dropped only if its stats prove it fails EVERY + // read. Skipping is not applied for a whole-table read, where every added file must conflict. + val candidateFiles = + if (spark.conf.get(DeltaSQLConf.DELTA_CONFLICT_DETECTION_DATA_SKIPPING_ENABLED) && + !currentTransactionInfo.readWholeTable) { + val readSnapshot = currentTransactionInfo.readSnapshot + val survivingPaths = currentTransactionInfo.readPredicates.iterator.flatMap { rp => + readSnapshot.filterFilesByDataSkipping(files, rp.dataPredicates) + }.map(_.path).toSet + val remaining = files.filter(f => survivingPaths.contains(f.path)) + if (remaining.size < files.size) { + recordDeltaEvent(deltaLog, + opType = "delta.conflictDetection.dataSkipping.filesSkipped", + data = Map( + "candidateFiles" -> files.size, + "skippedFiles" -> (files.size - remaining.size))) + } + remaining + } else { + files + } + // There is no reason to filter files if the table is not partitioned. if (currentTransactionInfo.readWholeTable || currentTransactionInfo.readSnapshot.metadata.partitionColumns.isEmpty) { - return files.headOption + return candidateFiles.headOption } import org.apache.spark.sql.delta.implicits._ - val filesDf = files.toDF(spark) + val filesDf = candidateFiles.toDF(spark) spark.conf.get(DeltaSQLConf.DELTA_CONFLICT_DETECTION_WIDEN_NONDETERMINISTIC_PREDICATES) match { case DeltaSQLConf.NonDeterministicPredicateWidening.OFF => diff --git a/spark/src/main/scala/org/apache/spark/sql/delta/sources/DeltaSQLConf.scala b/spark/src/main/scala/org/apache/spark/sql/delta/sources/DeltaSQLConf.scala index d4d9c813b6c..f8edb294261 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 @@ -534,6 +534,17 @@ trait DeltaSQLConfBase extends DeltaSQLConfUtils { .booleanConf .createWithDefault(true) + val DELTA_CONFLICT_DETECTION_DATA_SKIPPING_ENABLED = + buildConf("conflictDetection.dataSkipping.enabled") + .internal() + .doc( + """When enabled, conflict detection uses column-statistics data skipping to exclude + |concurrently-added files whose stats prove they cannot match the current transaction's + |read predicates, reducing unnecessary conflicts (especially on unpartitioned tables). + |One-way safe: a file is excluded only when its stats prove no match.""".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/main/scala/org/apache/spark/sql/delta/stats/DataSkippingReader.scala b/spark/src/main/scala/org/apache/spark/sql/delta/stats/DataSkippingReader.scala index 574e0e3008e..05fd8fd6fe1 100644 --- a/spark/src/main/scala/org/apache/spark/sql/delta/stats/DataSkippingReader.scala +++ b/spark/src/main/scala/org/apache/spark/sql/delta/stats/DataSkippingReader.scala @@ -286,22 +286,29 @@ trait DataSkippingReaderBase DeltaSQLConf.DELTA_DATASKIPPING_PARTITION_LIKE_FILTERS_ADDITIONAL_SUPPORTED_EXPRESSIONS) .toSet.flatMap((exprs: String) => exprs.split(",")) - /** Returns a DataFrame expression to obtain a list of files with parsed statistics. */ - private def withStatsInternal0: DataFrame = { - val parsedStats = from_json(col("stats"), statsSchema) + /** + * Parses (and, when the schema contains VariantType, Z85-decodes) the JSON `statsCol` into the + * `statsSchema` struct. Shared by [[withStatsInternal0]] and conflict-detection data skipping + * ([[filterFilesByDataSkipping]]) so both parse stats the same way. + */ + private def parseAndDecodeStats(statsCol: Column): Column = { + val parsedStats = from_json(statsCol, statsSchema) // Only use DecodeNestedZ85EncodedVariant if the schema contains VariantType. // This avoids performance overhead for tables without variant columns. // `DecodeNestedZ85EncodedVariant` is a temporary workaround since the Spark 4.1 from_json // expression has no way to decode a VariantVal from an encoded Z85 string. // TODO: Add Z85 decoding to Variant in Spark 4.2 and use that from_json option here. - val decodedStats = if (SchemaUtils.checkForVariantTypeColumnsRecursively(statsSchema)) { + if (SchemaUtils.checkForVariantTypeColumnsRecursively(statsSchema)) { Column(DecodeNestedZ85EncodedVariant(parsedStats.expr)) } else { parsedStats } - allFiles.withColumn("stats", decodedStats) } + /** Returns a DataFrame expression to obtain a list of files with parsed statistics. */ + private def withStatsInternal0: DataFrame = + allFiles.withColumn("stats", parseAndDecodeStats(col("stats"))) + private lazy val withStatsCache = cacheDS(withStatsInternal0, s"Delta Table State with Stats #$version - $redactedPath") @@ -641,6 +648,75 @@ trait DataSkippingReaderBase files.toSeq -> Seq(DataSize(totalSize), DataSize(partitionSize), DataSize(scanSize)) } + /** + * Builds a single [[DataSkippingPredicate]] (skipping expression + the stats it references) from + * `dataFilters`, mirroring [[filesForScan]]'s eligibility filtering, per-filter construction and + * conjunction fold. Returns None when stats skipping is unavailable or no eligible filter yields + * a predicate. The filters are AND-combined, so callers must pass filters from a single logical + * read (predicates from independent reads have OR, not AND, semantics). + */ + private[delta] def buildDataSkippingPredicate( + dataFilters: Seq[Expression]): Option[DataSkippingPredicate] = { + import DeltaTableUtils._ + if (!useStats) return None + // Mirror filesForScan eligibility: drop subquery / non-deterministic / metadata filters, so we + // never build a skipping predicate that could wrongly exclude a matching file. + val eligibleFilters = dataFilters.filterNot { f => + containsSubquery(f) || !f.deterministic || f.exists { + case MetadataAttribute(_) => true + case _ => false + } + } + val constructDataFilters = new DataFiltersBuilder( + spark = spark, + dataSkippingType = DeltaDataSkippingType.dataSkippingOnlyV1, + getStatsColumnOpt = (s: StatsColumn) => getStatsColumnOpt(s)) + eligibleFilters + .flatMap(f => constructDataFilters(f)) + .reduceOption((skip1, skip2) => DataSkippingPredicate( + skip1.expr && skip2.expr, skip1.referencedStats ++ skip2.referencedStats)) + } + + /** + * Conflict-detection helper (reader-side data skipping): returns the subset of `files` whose + * statistics do NOT prove they fail `dataFilters` -- i.e. the files that could still match and + * therefore must be treated as conflicts. Files with missing/insufficient stats (or when skipping + * is unavailable) are kept. + * + * `dataFilters` must come from a single logical read: they are AND-combined via + * [[buildDataSkippingPredicate]]. Callers with multiple independent reads must invoke this per + * read and union the survivors (OR semantics). + * + * One-way safe: it applies `expr || !verifyStatsForFilter(...)` exactly as + * [[getDataSkippedFiles]], so a file is dropped only when its stats *prove* it cannot match -- a + * real conflict is never turned into a false negative. Returns the original [[AddFile]]s (matched + * by path), stats untouched. + */ + private[delta] def filterFilesByDataSkipping( + files: Seq[AddFile], + dataFilters: Seq[Expression]): Seq[AddFile] = { + import org.apache.spark.sql.delta.implicits._ + if (files.isEmpty || dataFilters.isEmpty || schema.isEmpty) return files + buildDataSkippingPredicate(dataFilters) match { + // No usable data-skipping predicate -> keep all files (conservative). + case None => files + case Some(pred) => + val survivingPaths = recordFrameProfile( + "Delta", "DataSkippingReader.filterFilesByDataSkipping") { + // `expr || !verifyStatsForFilter(...)` keeps any file whose referenced stats are + // missing/NULL (mirrors getDataSkippedFiles): only skip when stats prove no match. + files.toDF(spark) + .withColumn("stats", parseAndDecodeStats(col("stats"))) + .where(pred.expr || !verifyStatsForFilter(pred.referencedStats)) + .select("path") + .collect() + .map(_.getString(0)) + .toSet + } + files.filter(f => survivingPaths.contains(f.path)) + } + } + private def getCorrectDataSkippingType( dataSkippingType: DeltaDataSkippingType): DeltaDataSkippingType = { dataSkippingType diff --git a/spark/src/test/scala/org/apache/spark/sql/delta/ConflictDataSkippingSuite.scala b/spark/src/test/scala/org/apache/spark/sql/delta/ConflictDataSkippingSuite.scala new file mode 100644 index 00000000000..112862ac131 --- /dev/null +++ b/spark/src/test/scala/org/apache/spark/sql/delta/ConflictDataSkippingSuite.scala @@ -0,0 +1,171 @@ +/* + * Copyright (2021) The Delta Lake Project Authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.spark.sql.delta + +import java.io.File + +import scala.concurrent.duration.Duration + +import org.apache.spark.sql.delta.concurrency.PhaseLockingTestMixin +import org.apache.spark.sql.delta.concurrency.TransactionExecutionTestMixin +import org.apache.spark.sql.delta.sources.DeltaSQLConf +import org.apache.spark.sql.delta.test.DeltaSQLCommandTest + +import org.apache.spark.SparkException +import org.apache.spark.sql.{QueryTest, Row} +import org.apache.spark.sql.functions.lit +import org.apache.spark.sql.test.SharedSparkSession +import org.apache.spark.util.ThreadUtils + +/** + * Tests for reader-side conflict-time data skipping + * ([[DeltaSQLConf.DELTA_CONFLICT_DETECTION_DATA_SKIPPING_ENABLED]]). + * + * A concurrently-added file whose column stats prove it cannot match the current transaction's read + * predicates should NOT cause an append conflict, especially on unpartitioned tables, where the + * append check would otherwise conflict on any added file. Skipping must be one-way safe: a file + * with missing stats is always kept, so a real conflict is never missed. + */ +class ConflictDataSkippingSuite extends QueryTest + with SharedSparkSession + with DeltaSQLCommandTest + with PhaseLockingTestMixin + with TransactionExecutionTestMixin { + + private def tableRef(dir: File): String = s"delta.`${dir.getCanonicalPath}`" + + /** + * Unpartitioned table with `id` in [0, 1000) across 10 files (file i covers [100i, 100i+100)), + * committed at Serializable isolation so that concurrent (blind) appends are conflict-checked. + * When `numIndexedCols` is set, stats collection is limited accordingly (0 = no stats). + */ + private def createTable(dir: File, numIndexedCols: Option[Int] = None): Unit = { + spark.range(start = 0, end = 1000, step = 1, numPartitions = 10) + .write.format("delta").mode("append").save(dir.getAbsolutePath) + val extraProps = numIndexedCols + .map(n => s", '${DeltaConfigs.DATA_SKIPPING_NUM_INDEXED_COLS.key}' = '$n'") + .getOrElse("") + sql(s"ALTER TABLE ${tableRef(dir)} SET TBLPROPERTIES " + + s"('${DeltaConfigs.ISOLATION_LEVEL.key}' = 'Serializable'$extraProps)") + } + + /** A DELETE that runs under the given data-skipping setting (evaluated on its commit thread). */ + private def deleteTxn(dir: File, condition: String, dataSkipping: Boolean): () => Array[Row] = + () => { + withSQLConf( + DeltaSQLConf.DELTA_CONFLICT_DETECTION_DATA_SKIPPING_ENABLED.key -> dataSkipping.toString) { + sql(s"DELETE FROM ${tableRef(dir)} WHERE $condition").collect() + } + Array.empty[Row] + } + + /** A blind append of `id` in [start, end), optionally into partition `p`. */ + private def appendTxn( + dir: File, start: Long, end: Long, partition: Option[Int] = None): () => Array[Row] = + () => { + var df = spark.range(start, end).toDF() + partition.foreach(p => df = df.withColumn("p", lit(p))) + df.write.format("delta").mode("append").save(dir.getAbsolutePath) + Array.empty[Row] + } + + /** Expected surviving ids after deleting id<50 and appending [1000, 1100). */ + private val disjointExpected: Seq[Row] = + ((50L until 1000L) ++ (1000L until 1100L)).map(Row(_)) + + private def assertConcurrentAppend(e: SparkException): Unit = + assert(e.getCause.isInstanceOf[io.delta.exceptions.ConcurrentAppendException], + s"Expected ConcurrentAppendException, got: ${e.getCause}") + + test("disjoint data ranges: added file is skipped, no conflict") { + withTempDir { dir => + createTable(dir) + // A (loser) deletes id<50; B (winner) appends [1000,1100), disjoint from A's predicate. + val txnA = deleteTxn(dir, "id < 50", dataSkipping = true) + val txnB = appendTxn(dir, 1000, 1100) + + val (futureA, futureB) = runTxnsWithOrder__A_Start__B__A_End(txnA, txnB) + ThreadUtils.awaitResult(futureA, Duration.Inf) + ThreadUtils.awaitResult(futureB, Duration.Inf) + + // Both committed: id<50 deleted, [1000,1100) appended. + checkAnswer( + spark.read.format("delta").load(dir.getAbsolutePath).select("id"), disjointExpected) + } + } + + test("overlapping data ranges: added file matches the predicate, still conflicts") { + withTempDir { dir => + createTable(dir) + // A's predicate id>=950 overlaps B's appended [1000,1100) file range -> not skippable. + val txnA = deleteTxn(dir, "id >= 950", dataSkipping = true) + val txnB = appendTxn(dir, 1000, 1100) + + val (futureA, futureB) = runTxnsWithOrder__A_Start__B__A_End(txnA, txnB) + ThreadUtils.awaitResult(futureB, Duration.Inf) + val e = intercept[SparkException] { ThreadUtils.awaitResult(futureA, Duration.Inf) } + assertConcurrentAppend(e) + } + } + + test("feature disabled: disjoint data ranges still conflict") { + withTempDir { dir => + createTable(dir) + val txnA = deleteTxn(dir, "id < 50", dataSkipping = false) + val txnB = appendTxn(dir, 1000, 1100) + + val (futureA, futureB) = runTxnsWithOrder__A_Start__B__A_End(txnA, txnB) + ThreadUtils.awaitResult(futureB, Duration.Inf) + val e = intercept[SparkException] { ThreadUtils.awaitResult(futureA, Duration.Inf) } + assertConcurrentAppend(e) + } + } + + test("missing stats: disjoint ranges still conflict (one-way safety)") { + withTempDir { dir => + // No indexed columns -> the appended file has no id stats -> must NOT be skipped. + createTable(dir, numIndexedCols = Some(0)) + val txnA = deleteTxn(dir, "id < 50", dataSkipping = true) + val txnB = appendTxn(dir, 1000, 1100) + + val (futureA, futureB) = runTxnsWithOrder__A_Start__B__A_End(txnA, txnB) + ThreadUtils.awaitResult(futureB, Duration.Inf) + val e = intercept[SparkException] { ThreadUtils.awaitResult(futureA, Duration.Inf) } + assertConcurrentAppend(e) + } + } + + test("partitioned table: data skipping on a non-partition column avoids the conflict") { + withTempDir { dir => + // Partitioned by `p`; the appended file shares the partition but its id range is disjoint. + spark.range(start = 0, end = 1000, step = 1, numPartitions = 10).withColumn("p", lit(0)) + .write.partitionBy("p").format("delta").mode("append").save(dir.getAbsolutePath) + sql(s"ALTER TABLE ${tableRef(dir)} SET TBLPROPERTIES " + + s"('${DeltaConfigs.ISOLATION_LEVEL.key}' = 'Serializable')") + + val txnA = deleteTxn(dir, "id < 50", dataSkipping = true) + val txnB = appendTxn(dir, 1000, 1100, partition = Some(0)) + + val (futureA, futureB) = runTxnsWithOrder__A_Start__B__A_End(txnA, txnB) + ThreadUtils.awaitResult(futureA, Duration.Inf) + ThreadUtils.awaitResult(futureB, Duration.Inf) + + checkAnswer( + spark.read.format("delta").load(dir.getAbsolutePath).select("id"), disjointExpected) + } + } +} From f6b092fd032c6291efc67ba48495225ff786cafa Mon Sep 17 00:00:00 2001 From: Eunjin Song Date: Mon, 3 Aug 2026 11:19:09 -0700 Subject: [PATCH 2/4] Guard conflict-time data skipping on non-empty read predicates Fix a Serializable soundness gap in reader-side conflict-time data skipping. A non-blind-append transaction can record no read predicates (e.g. it removes files without reading the table). With skipping enabled its empty read set produced an empty survivor set, so getFirstFileMatchingPartitionPredicates returned None and a real ConcurrentAppendException was silently suppressed. Skipping is now applied only when the transaction has at least one read predicate; otherwise every concurrently added file stays a conflict candidate, matching behavior with the feature disabled. Add ConflictDataSkippingSuite coverage: - empty read predicates on a non-blind-append txn still conflict (regression test for the fix above); - a whole-table read is never skipped and conflicts with a concurrent append; - filterFilesByDataSkipping AND-combines predicates within a single read. Co-Authored-By: Claude Opus 4.8 --- .../spark/sql/delta/ConflictChecker.scala | 9 +- .../sql/delta/ConflictDataSkippingSuite.scala | 83 ++++++++++++++++++- 2 files changed, 89 insertions(+), 3 deletions(-) diff --git a/spark/src/main/scala/org/apache/spark/sql/delta/ConflictChecker.scala b/spark/src/main/scala/org/apache/spark/sql/delta/ConflictChecker.scala index b92392e0c87..9e7db8589e1 100644 --- a/spark/src/main/scala/org/apache/spark/sql/delta/ConflictChecker.scala +++ b/spark/src/main/scala/org/apache/spark/sql/delta/ConflictChecker.scala @@ -1000,10 +1000,15 @@ private[delta] class ConflictChecker( // on every added file below. Data predicates are applied *per read predicate* and the survivors // unioned (OR across reads), because independent reads have OR, not AND, semantics: a file is a // candidate if it could match ANY read; it is dropped only if its stats prove it fails EVERY - // read. Skipping is not applied for a whole-table read, where every added file must conflict. + // read. Skipping is not applied for a whole-table read, nor when there are no read predicates + // at all: a transaction can be a non-blind-append (e.g. it removed files) yet have no read + // predicates, and in that case every added file must remain a conflict candidate, exactly as + // when the feature is disabled. Guarding on non-empty read predicates avoids an empty survivor + // set silently suppressing a real conflict. val candidateFiles = if (spark.conf.get(DeltaSQLConf.DELTA_CONFLICT_DETECTION_DATA_SKIPPING_ENABLED) && - !currentTransactionInfo.readWholeTable) { + !currentTransactionInfo.readWholeTable && + currentTransactionInfo.readPredicates.nonEmpty) { val readSnapshot = currentTransactionInfo.readSnapshot val survivingPaths = currentTransactionInfo.readPredicates.iterator.flatMap { rp => readSnapshot.filterFilesByDataSkipping(files, rp.dataPredicates) diff --git a/spark/src/test/scala/org/apache/spark/sql/delta/ConflictDataSkippingSuite.scala b/spark/src/test/scala/org/apache/spark/sql/delta/ConflictDataSkippingSuite.scala index 112862ac131..b003835bd13 100644 --- a/spark/src/test/scala/org/apache/spark/sql/delta/ConflictDataSkippingSuite.scala +++ b/spark/src/test/scala/org/apache/spark/sql/delta/ConflictDataSkippingSuite.scala @@ -20,15 +20,18 @@ import java.io.File import scala.concurrent.duration.Duration +import org.apache.spark.sql.delta.actions.AddFile import org.apache.spark.sql.delta.concurrency.PhaseLockingTestMixin import org.apache.spark.sql.delta.concurrency.TransactionExecutionTestMixin import org.apache.spark.sql.delta.sources.DeltaSQLConf import org.apache.spark.sql.delta.test.DeltaSQLCommandTest import org.apache.spark.SparkException -import org.apache.spark.sql.{QueryTest, Row} +import org.apache.spark.sql.{QueryTest, Row, SaveMode} +import org.apache.spark.sql.catalyst.expressions.{AttributeReference, GreaterThanOrEqual, LessThan, Literal} import org.apache.spark.sql.functions.lit import org.apache.spark.sql.test.SharedSparkSession +import org.apache.spark.sql.types.LongType import org.apache.spark.util.ThreadUtils /** @@ -168,4 +171,82 @@ class ConflictDataSkippingSuite extends QueryTest spark.read.format("delta").load(dir.getAbsolutePath).select("id"), disjointExpected) } } + + private def manufacturedAdd(name: String): AddFile = + AddFile(name, Map.empty[String, String], size = 1L, modificationTime = 1L, dataChange = true) + + test("empty read predicates on a non-blind-append txn: added file still conflicts") { + // Soundness of the Serializable blind-append case. A transaction can be a NON-blind append + // (it removes a file, so onlyAddFiles = false) and yet record no read predicates. With data + // skipping enabled such a txn must still conflict-check every concurrently added file, exactly + // as it does with the feature disabled. Before the guard on non-empty read predicates, the + // empty read set produced an empty survivor set and the conflict was silently suppressed. + withTempDir { dir => + createTable(dir) + val log = DeltaLog.forTable(spark, dir.getCanonicalPath) + // An existing file to remove, which makes the loser a non-blind append with no reads. + val existingRemove = log.update().allFiles.collect().head.remove + withSQLConf( + DeltaSQLConf.DELTA_CONFLICT_DETECTION_DATA_SKIPPING_ENABLED.key -> "true") { + val loser = log.startTransaction() + // Winner: a blind append committed while the loser is still open. + log.startTransaction().commit( + Seq(manufacturedAdd("winner.parquet")), DeltaOperations.Write(SaveMode.Append)) + // Loser adds AND removes without reading -> not a blind append, no read predicates. + intercept[io.delta.exceptions.ConcurrentAppendException] { + loser.commit( + Seq(manufacturedAdd("loser.parquet"), existingRemove), + DeltaOperations.Write(SaveMode.Append)) + } + } + } + } + + test("whole-table read is never skipped: conflicts with a concurrent append") { + // A whole-table read must treat every concurrently added file as a conflict, even with data + // skipping enabled -- the feature is deliberately bypassed for readWholeTable. + withTempDir { dir => + createTable(dir) + val log = DeltaLog.forTable(spark, dir.getCanonicalPath) + withSQLConf( + DeltaSQLConf.DELTA_CONFLICT_DETECTION_DATA_SKIPPING_ENABLED.key -> "true") { + val loser = log.startTransaction() + loser.readWholeTable() + log.startTransaction().commit( + Seq(manufacturedAdd("winner.parquet")), DeltaOperations.Write(SaveMode.Append)) + intercept[io.delta.exceptions.ConcurrentAppendException] { + loser.commit( + Seq(manufacturedAdd("loser.parquet")), DeltaOperations.Write(SaveMode.Append)) + } + } + } + } + + test("filterFilesByDataSkipping AND-combines predicates within one read") { + // Predicates from a SINGLE read have AND semantics (OR is only across independent reads, which + // ConflictChecker unions by construction). This exercises that AND directly. + withTempDir { dir => + createTable(dir) + val log = DeltaLog.forTable(spark, dir.getCanonicalPath) + val snapshot = log.update() + val allFiles = snapshot.allFiles.collect().toSeq + assert(allFiles.size == 10) + + // Resolved catalyst predicates on the `id` column; built directly so the literal stays a + // bare Long (a parsed `id >= 200` would wrap it in a Cast, which is not skipping-eligible). + val id = AttributeReference("id", LongType)() + val ge200 = GreaterThanOrEqual(id, Literal(200L)) + val lt300 = LessThan(id, Literal(300L)) + + // Only the [200, 300) file can match `id >= 200 AND id < 300`. OR semantics would keep every + // file with id >= 200 or id < 300, i.e. all 10. + val survivors = snapshot.filterFilesByDataSkipping(allFiles, Seq(ge200, lt300)) + assert(survivors.size == 1, + s"expected exactly the [200,300) file, got ${survivors.map(_.path).sorted}") + + // Sanity: a single predicate keeps its whole matching range (files with max id >= 200). + val geOnly = snapshot.filterFilesByDataSkipping(allFiles, Seq(ge200)) + assert(geOnly.size == 8, s"expected 8 files with id >= 200, got ${geOnly.size}") + } + } } From e2f12cb539879cd2b9a05992816e5a8d759b9c4c Mon Sep 17 00:00:00 2001 From: Eunjin Song Date: Fri, 7 Aug 2026 18:16:18 -0700 Subject: [PATCH 3/4] Address review: single-job multi-read skipping + eval fallback Combine per-read data-skipping into one Spark job and fail safe on evaluation errors, per review comments on #7358. - filterFilesMatchingAnyReadPredicate(files, dataFiltersPerRead): builds one skipping predicate per read and OR-combines them into a single `where`, so all read predicates are evaluated in one Spark job instead of one job per read. A read with no usable predicate matches everything -> short-circuit and keep all files, no job. - filterFilesByDataSkipping is now the single-read case delegating to it. - Wrap build+eval in try/catch(NonFatal): on failure log a warning and return all candidate files (default conflict behavior), so a skipping error can never abort a valid commit. - ConflictChecker calls the new method once instead of flatMap-per-read. - Test: OR-combines independent reads in one call. Co-Authored-By: Claude Opus 4.8 --- .../spark/sql/delta/ConflictChecker.scala | 21 +++-- .../sql/delta/stats/DataSkippingReader.scala | 90 +++++++++++++------ .../sql/delta/ConflictDataSkippingSuite.scala | 30 ++++++- 3 files changed, 103 insertions(+), 38 deletions(-) diff --git a/spark/src/main/scala/org/apache/spark/sql/delta/ConflictChecker.scala b/spark/src/main/scala/org/apache/spark/sql/delta/ConflictChecker.scala index 9e7db8589e1..0201dfe8403 100644 --- a/spark/src/main/scala/org/apache/spark/sql/delta/ConflictChecker.scala +++ b/spark/src/main/scala/org/apache/spark/sql/delta/ConflictChecker.scala @@ -998,22 +998,21 @@ private[delta] class ConflictChecker( // skipping. A concurrently-added file whose stats prove it cannot match the read predicates is // not a conflict -- this is what lets unpartitioned / liquid-clustered tables avoid conflicting // on every added file below. Data predicates are applied *per read predicate* and the survivors - // unioned (OR across reads), because independent reads have OR, not AND, semantics: a file is a - // candidate if it could match ANY read; it is dropped only if its stats prove it fails EVERY - // read. Skipping is not applied for a whole-table read, nor when there are no read predicates - // at all: a transaction can be a non-blind-append (e.g. it removed files) yet have no read - // predicates, and in that case every added file must remain a conflict candidate, exactly as - // when the feature is disabled. Guarding on non-empty read predicates avoids an empty survivor - // set silently suppressing a real conflict. + // OR-combined, because independent reads have OR, not AND, semantics: a file is a candidate if + // it could match ANY read; it is dropped only if its stats prove it fails EVERY read. All reads + // are evaluated in a single Spark job by filterFilesMatchingAnyReadPredicate. Skipping is not + // applied for a whole-table read, nor when there are no read predicates at all: a transaction + // can be a non-blind-append (e.g. it removed files) yet have no read predicates, and in that + // case every added file must remain a conflict candidate, exactly as when the feature is + // disabled. Guarding on non-empty read predicates avoids an empty survivor set silently + // suppressing a real conflict. val candidateFiles = if (spark.conf.get(DeltaSQLConf.DELTA_CONFLICT_DETECTION_DATA_SKIPPING_ENABLED) && !currentTransactionInfo.readWholeTable && currentTransactionInfo.readPredicates.nonEmpty) { val readSnapshot = currentTransactionInfo.readSnapshot - val survivingPaths = currentTransactionInfo.readPredicates.iterator.flatMap { rp => - readSnapshot.filterFilesByDataSkipping(files, rp.dataPredicates) - }.map(_.path).toSet - val remaining = files.filter(f => survivingPaths.contains(f.path)) + val remaining = readSnapshot.filterFilesMatchingAnyReadPredicate( + files, currentTransactionInfo.readPredicates.map(_.dataPredicates).toSeq) if (remaining.size < files.size) { recordDeltaEvent(deltaLog, opType = "delta.conflictDetection.dataSkipping.filesSkipped", diff --git a/spark/src/main/scala/org/apache/spark/sql/delta/stats/DataSkippingReader.scala b/spark/src/main/scala/org/apache/spark/sql/delta/stats/DataSkippingReader.scala index 05fd8fd6fe1..6c9673c55a4 100644 --- a/spark/src/main/scala/org/apache/spark/sql/delta/stats/DataSkippingReader.scala +++ b/spark/src/main/scala/org/apache/spark/sql/delta/stats/DataSkippingReader.scala @@ -20,6 +20,7 @@ package org.apache.spark.sql.delta.stats import java.io.Closeable import scala.collection.mutable.ArrayBuffer +import scala.util.control.NonFatal import org.apache.spark.sql.delta.skipping.clustering.{ClusteredTableUtils, ClusteringColumnInfo} import org.apache.spark.sql.delta.ClassicColumnConversions._ @@ -684,36 +685,73 @@ trait DataSkippingReaderBase * is unavailable) are kept. * * `dataFilters` must come from a single logical read: they are AND-combined via - * [[buildDataSkippingPredicate]]. Callers with multiple independent reads must invoke this per - * read and union the survivors (OR semantics). - * - * One-way safe: it applies `expr || !verifyStatsForFilter(...)` exactly as - * [[getDataSkippedFiles]], so a file is dropped only when its stats *prove* it cannot match -- a - * real conflict is never turned into a false negative. Returns the original [[AddFile]]s (matched - * by path), stats untouched. + * [[buildDataSkippingPredicate]]. This is the one-read case of + * [[filterFilesMatchingAnyReadPredicate]]; callers with multiple independent reads should use + * that method so all reads are evaluated in a single Spark job. */ private[delta] def filterFilesByDataSkipping( files: Seq[AddFile], - dataFilters: Seq[Expression]): Seq[AddFile] = { + dataFilters: Seq[Expression]): Seq[AddFile] = + filterFilesMatchingAnyReadPredicate(files, Seq(dataFilters)) + + /** + * Conflict-detection helper (reader-side data skipping) over several INDEPENDENT reads: returns + * the subset of `files` that could still match ANY read and therefore must be treated as + * conflicts. Each inner `Seq[Expression]` is one logical read's data filters (AND-combined via + * [[buildDataSkippingPredicate]]); the reads are OR-combined, matching read semantics -- a file + * is a candidate if it could match any one of them. + * + * All reads are evaluated in a SINGLE Spark job: the per-read skipping predicates are OR-ed + * together into one `where` clause, rather than filtering per read and unioning the survivors + * (which launched one job per read). Note we cannot instead flatten every read's filters into one + * predicate -- that would AND them (read1 AND read2), the opposite of the OR we need. + * + * One-way safe: each read contributes `expr || !verifyStatsForFilter(...)` exactly as + * [[getDataSkippedFiles]], so a file is dropped only when its stats *prove* it fails EVERY read + * -- a real conflict is never a false negative. Files with missing/insufficient stats, a + * read with no usable skipping predicate (empty / ineligible filters -> matches everything), or a + * table without stats are all kept. Returns the original [[AddFile]]s (matched by path). + * + * Fail-safe: skipping here is a pure optimization over correct (conservative) conflict detection, + * so if building or evaluating the predicate throws we fall back to the default behavior of + * keeping all `files` as conflict candidates rather than failing the commit. + */ + private[delta] def filterFilesMatchingAnyReadPredicate( + files: Seq[AddFile], + dataFiltersPerRead: Seq[Seq[Expression]]): Seq[AddFile] = { import org.apache.spark.sql.delta.implicits._ - if (files.isEmpty || dataFilters.isEmpty || schema.isEmpty) return files - buildDataSkippingPredicate(dataFilters) match { - // No usable data-skipping predicate -> keep all files (conservative). - case None => files - case Some(pred) => - val survivingPaths = recordFrameProfile( - "Delta", "DataSkippingReader.filterFilesByDataSkipping") { - // `expr || !verifyStatsForFilter(...)` keeps any file whose referenced stats are - // missing/NULL (mirrors getDataSkippedFiles): only skip when stats prove no match. - files.toDF(spark) - .withColumn("stats", parseAndDecodeStats(col("stats"))) - .where(pred.expr || !verifyStatsForFilter(pred.referencedStats)) - .select("path") - .collect() - .map(_.getString(0)) - .toSet - } - files.filter(f => survivingPaths.contains(f.path)) + if (files.isEmpty || dataFiltersPerRead.isEmpty || schema.isEmpty) return files + try { + // One skipping predicate per read. A read with no usable predicate (empty or ineligible + // filters) matches every file -> nothing can be skipped, so keep all files without a job. + val perReadPredicates = dataFiltersPerRead.map { dataFilters => + if (dataFilters.isEmpty) None else buildDataSkippingPredicate(dataFilters) + } + if (perReadPredicates.exists(_.isEmpty)) return files + // Survive if the file could match ANY read. Per read, `expr || !verifyStatsForFilter(...)` + // keeps any file whose referenced stats are missing/NULL (mirrors getDataSkippedFiles): only + // skip when stats prove no match. OR the reads so a match against any one keeps the file. + val survivorCondition = perReadPredicates.flatten + .map(pred => pred.expr || !verifyStatsForFilter(pred.referencedStats)) + .reduce(_ || _) + val survivingPaths = recordFrameProfile( + "Delta", "DataSkippingReader.filterFilesMatchingAnyReadPredicate") { + files.toDF(spark) + .withColumn("stats", parseAndDecodeStats(col("stats"))) + .where(survivorCondition) + .select("path") + .collect() + .map(_.getString(0)) + .toSet + } + files.filter(f => survivingPaths.contains(f.path)) + } catch { + case NonFatal(e) => + // Optimization only: never let a skipping failure abort a commit. Fall back to the default + // (feature-off) behavior of treating every added file as a conflict candidate. + logWarning(log"Conflict-time data skipping failed to evaluate; falling back to treating " + + log"all added files as conflict candidates", e) + files } } diff --git a/spark/src/test/scala/org/apache/spark/sql/delta/ConflictDataSkippingSuite.scala b/spark/src/test/scala/org/apache/spark/sql/delta/ConflictDataSkippingSuite.scala index b003835bd13..bf5648ce97e 100644 --- a/spark/src/test/scala/org/apache/spark/sql/delta/ConflictDataSkippingSuite.scala +++ b/spark/src/test/scala/org/apache/spark/sql/delta/ConflictDataSkippingSuite.scala @@ -28,7 +28,7 @@ import org.apache.spark.sql.delta.test.DeltaSQLCommandTest import org.apache.spark.SparkException import org.apache.spark.sql.{QueryTest, Row, SaveMode} -import org.apache.spark.sql.catalyst.expressions.{AttributeReference, GreaterThanOrEqual, LessThan, Literal} +import org.apache.spark.sql.catalyst.expressions.{AttributeReference, Expression, GreaterThanOrEqual, LessThan, Literal} import org.apache.spark.sql.functions.lit import org.apache.spark.sql.test.SharedSparkSession import org.apache.spark.sql.types.LongType @@ -249,4 +249,32 @@ class ConflictDataSkippingSuite extends QueryTest assert(geOnly.size == 8, s"expected 8 files with id >= 200, got ${geOnly.size}") } } + + test("filterFilesMatchingAnyReadPredicate OR-combines independent reads in one call") { + // Independent reads have OR semantics: a file survives if it could match ANY read. This is the + // multi-read path ConflictChecker uses -- all reads are evaluated together in a single job. + withTempDir { dir => + createTable(dir) + val log = DeltaLog.forTable(spark, dir.getCanonicalPath) + val snapshot = log.update() + val allFiles = snapshot.allFiles.collect().toSeq + assert(allFiles.size == 10) + + val id = AttributeReference("id", LongType)() + // Read 1: id in [200, 300) -> the single [200,300) file. Read 2: id >= 800 -> the [800,900) + // and [900,1000) files. Their union (OR across reads) is exactly 3 files. + val read1 = Seq[Expression]( + GreaterThanOrEqual(id, Literal(200L)), LessThan(id, Literal(300L))) + val read2 = Seq[Expression](GreaterThanOrEqual(id, Literal(800L))) + val survivors = + snapshot.filterFilesMatchingAnyReadPredicate(allFiles, Seq(read1, read2)) + assert(survivors.size == 3, + s"expected [200,300) + [800,1000), got ${survivors.map(_.path).sorted}") + + // A read with no usable predicate matches everything, so nothing can be skipped -> all kept. + val allKept = snapshot.filterFilesMatchingAnyReadPredicate(allFiles, Seq(read1, Seq.empty)) + assert(allKept.size == allFiles.size, + s"a read with no predicate must keep all files, got ${allKept.size}") + } + } } From e4ac0dab359c229a0b4a3f0700051ff8e20092a7 Mon Sep 17 00:00:00 2001 From: Eunjin Song Date: Wed, 12 Aug 2026 22:59:17 -0700 Subject: [PATCH 4/4] Address review: isolate conflict-time skipping in a self-typed trait Move buildDataSkippingPredicate / filterFilesByDataSkipping / filterFilesMatchingAnyReadPredicate out of DataSkippingReaderBase into a new self-typed trait ConflictDataSkippingReader (self: DataSkippingReaderBase), mixed into the base. Existing data-skipping code is left untouched: withStatsInternal0 is reverted to its original inline form, so the only change to DataSkippingReader.scala is the one-line trait mixin. parseAndDecodeStats is duplicated (trait-private) rather than shared with withStatsInternal0, keeping the feature a fully isolated additive unit per review feedback. Co-Authored-By: Claude Opus 4.8 --- .../stats/ConflictDataSkippingReader.scala | 172 ++++++++++++++++++ .../sql/delta/stats/DataSkippingReader.scala | 125 +------------ 2 files changed, 178 insertions(+), 119 deletions(-) create mode 100644 spark/src/main/scala/org/apache/spark/sql/delta/stats/ConflictDataSkippingReader.scala diff --git a/spark/src/main/scala/org/apache/spark/sql/delta/stats/ConflictDataSkippingReader.scala b/spark/src/main/scala/org/apache/spark/sql/delta/stats/ConflictDataSkippingReader.scala new file mode 100644 index 00000000000..2305397b7f5 --- /dev/null +++ b/spark/src/main/scala/org/apache/spark/sql/delta/stats/ConflictDataSkippingReader.scala @@ -0,0 +1,172 @@ +/* + * 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.stats + +import scala.util.control.NonFatal + +import org.apache.spark.sql.delta.ClassicColumnConversions._ +import org.apache.spark.sql.delta.DeltaTableUtils +import org.apache.spark.sql.delta.actions.AddFile +import org.apache.spark.sql.delta.expressions.DecodeNestedZ85EncodedVariant +import org.apache.spark.sql.delta.metering.DeltaLogging +import org.apache.spark.sql.delta.schema.SchemaUtils +import org.apache.spark.sql.delta.sources.DeltaSQLConf + +import org.apache.spark.sql.Column +import org.apache.spark.sql.catalyst.expressions._ +import org.apache.spark.sql.functions.{col, from_json} + +/** + * Reader-side data skipping used during conflict detection (row-level concurrency Case 1: writers + * touching disjoint data ranges). + * + * Mixed into [[DataSkippingReaderBase]] as a self-typed trait so this feature is a single isolated + * unit that only ADDS methods -- it does not modify any existing data-skipping code. It is opt-in: + * the caller ([[org.apache.spark.sql.delta.ConflictChecker]]) invokes these methods only behind + * `DeltaSQLConf.DELTA_CONFLICT_DETECTION_DATA_SKIPPING_ENABLED`, and each is one-way safe -- a file + * is dropped only when its stats *prove* it cannot match, so a real conflict is never a false + * negative. + */ +trait ConflictDataSkippingReader extends DeltaLogging { self: DataSkippingReaderBase => + + /** + * Parses (and, when the schema contains VariantType, Z85-decodes) the JSON `statsCol` into the + * `statsSchema` struct, mirroring [[withStatsInternal0]]. Deliberately kept private to this trait + * rather than shared with `withStatsInternal0`, so the existing stats path stays untouched. + */ + private def parseAndDecodeStats(statsCol: Column): Column = { + val parsedStats = from_json(statsCol, statsSchema) + // Only use DecodeNestedZ85EncodedVariant if the schema contains VariantType. + // This avoids performance overhead for tables without variant columns. + // `DecodeNestedZ85EncodedVariant` is a temporary workaround since the Spark 4.1 from_json + // expression has no way to decode a VariantVal from an encoded Z85 string. + // TODO: Add Z85 decoding to Variant in Spark 4.2 and use that from_json option here. + if (SchemaUtils.checkForVariantTypeColumnsRecursively(statsSchema)) { + Column(DecodeNestedZ85EncodedVariant(parsedStats.expr)) + } else { + parsedStats + } + } + + /** + * Builds a single [[DataSkippingPredicate]] (skipping expression + the stats it references) from + * `dataFilters`, mirroring [[filesForScan]]'s eligibility filtering, per-filter construction and + * conjunction fold. Returns None when stats skipping is unavailable or no eligible filter yields + * a predicate. The filters are AND-combined, so callers must pass filters from a single logical + * read (predicates from independent reads have OR, not AND, semantics). + */ + private[delta] def buildDataSkippingPredicate( + dataFilters: Seq[Expression]): Option[DataSkippingPredicate] = { + import DeltaTableUtils._ + // Stats-skipping conf, inlined from DataSkippingReaderBase.useStats (file-private, so not + // visible here). Reading it inline avoids widening that member's visibility. + if (!spark.sessionState.conf.getConf(DeltaSQLConf.DELTA_STATS_SKIPPING)) return None + // Mirror filesForScan eligibility: drop subquery / non-deterministic / metadata filters, so we + // never build a skipping predicate that could wrongly exclude a matching file. + val eligibleFilters = dataFilters.filterNot { f => + containsSubquery(f) || !f.deterministic || f.exists { + case MetadataAttribute(_) => true + case _ => false + } + } + val constructDataFilters = new DataFiltersBuilder( + spark = spark, + dataSkippingType = DeltaDataSkippingType.dataSkippingOnlyV1, + getStatsColumnOpt = (s: StatsColumn) => getStatsColumnOpt(s)) + eligibleFilters + .flatMap(f => constructDataFilters(f)) + .reduceOption((skip1, skip2) => DataSkippingPredicate( + skip1.expr && skip2.expr, skip1.referencedStats ++ skip2.referencedStats)) + } + + /** + * Conflict-detection helper (reader-side data skipping): returns the subset of `files` whose + * statistics do NOT prove they fail `dataFilters` -- i.e. the files that could still match and + * therefore must be treated as conflicts. Files with missing/insufficient stats (or when skipping + * is unavailable) are kept. + * + * `dataFilters` must come from a single logical read: they are AND-combined via + * [[buildDataSkippingPredicate]]. This is the one-read case of + * [[filterFilesMatchingAnyReadPredicate]]; callers with multiple independent reads should use + * that method so all reads are evaluated in a single Spark job. + */ + private[delta] def filterFilesByDataSkipping( + files: Seq[AddFile], + dataFilters: Seq[Expression]): Seq[AddFile] = + filterFilesMatchingAnyReadPredicate(files, Seq(dataFilters)) + + /** + * Conflict-detection helper (reader-side data skipping) over several INDEPENDENT reads: returns + * the subset of `files` that could still match ANY read and therefore must be treated as + * conflicts. Each inner `Seq[Expression]` is one logical read's data filters (AND-combined via + * [[buildDataSkippingPredicate]]); the reads are OR-combined, matching read semantics -- a file + * is a candidate if it could match any one of them. + * + * All reads are evaluated in a SINGLE Spark job: the per-read skipping predicates are OR-ed + * together into one `where` clause, rather than filtering per read and unioning the survivors + * (which launched one job per read). Note we cannot instead flatten every read's filters into one + * predicate -- that would AND them (read1 AND read2), the opposite of the OR we need. + * + * One-way safe: each read contributes `expr || !verifyStatsForFilter(...)` exactly as + * [[getDataSkippedFiles]], so a file is dropped only when its stats *prove* it fails EVERY read + * -- a real conflict is never a false negative. Files with missing/insufficient stats, a + * read with no usable skipping predicate (empty / ineligible filters -> matches everything), or a + * table without stats are all kept. Returns the original [[AddFile]]s (matched by path). + * + * Fail-safe: skipping here is a pure optimization over correct (conservative) conflict detection, + * so if building or evaluating the predicate throws we fall back to the default behavior of + * keeping all `files` as conflict candidates rather than failing the commit. + */ + private[delta] def filterFilesMatchingAnyReadPredicate( + files: Seq[AddFile], + dataFiltersPerRead: Seq[Seq[Expression]]): Seq[AddFile] = { + import org.apache.spark.sql.delta.implicits._ + if (files.isEmpty || dataFiltersPerRead.isEmpty || schema.isEmpty) return files + try { + // One skipping predicate per read. A read with no usable predicate (empty or ineligible + // filters) matches every file -> nothing can be skipped, so keep all files without a job. + val perReadPredicates = dataFiltersPerRead.map { dataFilters => + if (dataFilters.isEmpty) None else buildDataSkippingPredicate(dataFilters) + } + if (perReadPredicates.exists(_.isEmpty)) return files + // Survive if the file could match ANY read. Per read, `expr || !verifyStatsForFilter(...)` + // keeps any file whose referenced stats are missing/NULL (mirrors getDataSkippedFiles): only + // skip when stats prove no match. OR the reads so a match against any one keeps the file. + val survivorCondition = perReadPredicates.flatten + .map(pred => pred.expr || !verifyStatsForFilter(pred.referencedStats)) + .reduce(_ || _) + val survivingPaths = recordFrameProfile( + "Delta", "DataSkippingReader.filterFilesMatchingAnyReadPredicate") { + files.toDF(spark) + .withColumn("stats", parseAndDecodeStats(col("stats"))) + .where(survivorCondition) + .select("path") + .collect() + .map(_.getString(0)) + .toSet + } + files.filter(f => survivingPaths.contains(f.path)) + } catch { + case NonFatal(e) => + // Optimization only: never let a skipping failure abort a commit. Fall back to the default + // (feature-off) behavior of treating every added file as a conflict candidate. + logWarning(log"Conflict-time data skipping failed to evaluate; falling back to treating " + + log"all added files as conflict candidates", e) + files + } + } +} diff --git a/spark/src/main/scala/org/apache/spark/sql/delta/stats/DataSkippingReader.scala b/spark/src/main/scala/org/apache/spark/sql/delta/stats/DataSkippingReader.scala index 6c9673c55a4..73999236511 100644 --- a/spark/src/main/scala/org/apache/spark/sql/delta/stats/DataSkippingReader.scala +++ b/spark/src/main/scala/org/apache/spark/sql/delta/stats/DataSkippingReader.scala @@ -20,7 +20,6 @@ package org.apache.spark.sql.delta.stats import java.io.Closeable import scala.collection.mutable.ArrayBuffer -import scala.util.control.NonFatal import org.apache.spark.sql.delta.skipping.clustering.{ClusteredTableUtils, ClusteringColumnInfo} import org.apache.spark.sql.delta.ClassicColumnConversions._ @@ -265,6 +264,7 @@ trait DataSkippingReaderBase with StatisticsCollection with ReadsMetadataFields with StateCache + with ConflictDataSkippingReader with DeltaLogging { import DataSkippingReader._ @@ -287,29 +287,22 @@ trait DataSkippingReaderBase DeltaSQLConf.DELTA_DATASKIPPING_PARTITION_LIKE_FILTERS_ADDITIONAL_SUPPORTED_EXPRESSIONS) .toSet.flatMap((exprs: String) => exprs.split(",")) - /** - * Parses (and, when the schema contains VariantType, Z85-decodes) the JSON `statsCol` into the - * `statsSchema` struct. Shared by [[withStatsInternal0]] and conflict-detection data skipping - * ([[filterFilesByDataSkipping]]) so both parse stats the same way. - */ - private def parseAndDecodeStats(statsCol: Column): Column = { - val parsedStats = from_json(statsCol, statsSchema) + /** Returns a DataFrame expression to obtain a list of files with parsed statistics. */ + private def withStatsInternal0: DataFrame = { + val parsedStats = from_json(col("stats"), statsSchema) // Only use DecodeNestedZ85EncodedVariant if the schema contains VariantType. // This avoids performance overhead for tables without variant columns. // `DecodeNestedZ85EncodedVariant` is a temporary workaround since the Spark 4.1 from_json // expression has no way to decode a VariantVal from an encoded Z85 string. // TODO: Add Z85 decoding to Variant in Spark 4.2 and use that from_json option here. - if (SchemaUtils.checkForVariantTypeColumnsRecursively(statsSchema)) { + val decodedStats = if (SchemaUtils.checkForVariantTypeColumnsRecursively(statsSchema)) { Column(DecodeNestedZ85EncodedVariant(parsedStats.expr)) } else { parsedStats } + allFiles.withColumn("stats", decodedStats) } - /** Returns a DataFrame expression to obtain a list of files with parsed statistics. */ - private def withStatsInternal0: DataFrame = - allFiles.withColumn("stats", parseAndDecodeStats(col("stats"))) - private lazy val withStatsCache = cacheDS(withStatsInternal0, s"Delta Table State with Stats #$version - $redactedPath") @@ -649,112 +642,6 @@ trait DataSkippingReaderBase files.toSeq -> Seq(DataSize(totalSize), DataSize(partitionSize), DataSize(scanSize)) } - /** - * Builds a single [[DataSkippingPredicate]] (skipping expression + the stats it references) from - * `dataFilters`, mirroring [[filesForScan]]'s eligibility filtering, per-filter construction and - * conjunction fold. Returns None when stats skipping is unavailable or no eligible filter yields - * a predicate. The filters are AND-combined, so callers must pass filters from a single logical - * read (predicates from independent reads have OR, not AND, semantics). - */ - private[delta] def buildDataSkippingPredicate( - dataFilters: Seq[Expression]): Option[DataSkippingPredicate] = { - import DeltaTableUtils._ - if (!useStats) return None - // Mirror filesForScan eligibility: drop subquery / non-deterministic / metadata filters, so we - // never build a skipping predicate that could wrongly exclude a matching file. - val eligibleFilters = dataFilters.filterNot { f => - containsSubquery(f) || !f.deterministic || f.exists { - case MetadataAttribute(_) => true - case _ => false - } - } - val constructDataFilters = new DataFiltersBuilder( - spark = spark, - dataSkippingType = DeltaDataSkippingType.dataSkippingOnlyV1, - getStatsColumnOpt = (s: StatsColumn) => getStatsColumnOpt(s)) - eligibleFilters - .flatMap(f => constructDataFilters(f)) - .reduceOption((skip1, skip2) => DataSkippingPredicate( - skip1.expr && skip2.expr, skip1.referencedStats ++ skip2.referencedStats)) - } - - /** - * Conflict-detection helper (reader-side data skipping): returns the subset of `files` whose - * statistics do NOT prove they fail `dataFilters` -- i.e. the files that could still match and - * therefore must be treated as conflicts. Files with missing/insufficient stats (or when skipping - * is unavailable) are kept. - * - * `dataFilters` must come from a single logical read: they are AND-combined via - * [[buildDataSkippingPredicate]]. This is the one-read case of - * [[filterFilesMatchingAnyReadPredicate]]; callers with multiple independent reads should use - * that method so all reads are evaluated in a single Spark job. - */ - private[delta] def filterFilesByDataSkipping( - files: Seq[AddFile], - dataFilters: Seq[Expression]): Seq[AddFile] = - filterFilesMatchingAnyReadPredicate(files, Seq(dataFilters)) - - /** - * Conflict-detection helper (reader-side data skipping) over several INDEPENDENT reads: returns - * the subset of `files` that could still match ANY read and therefore must be treated as - * conflicts. Each inner `Seq[Expression]` is one logical read's data filters (AND-combined via - * [[buildDataSkippingPredicate]]); the reads are OR-combined, matching read semantics -- a file - * is a candidate if it could match any one of them. - * - * All reads are evaluated in a SINGLE Spark job: the per-read skipping predicates are OR-ed - * together into one `where` clause, rather than filtering per read and unioning the survivors - * (which launched one job per read). Note we cannot instead flatten every read's filters into one - * predicate -- that would AND them (read1 AND read2), the opposite of the OR we need. - * - * One-way safe: each read contributes `expr || !verifyStatsForFilter(...)` exactly as - * [[getDataSkippedFiles]], so a file is dropped only when its stats *prove* it fails EVERY read - * -- a real conflict is never a false negative. Files with missing/insufficient stats, a - * read with no usable skipping predicate (empty / ineligible filters -> matches everything), or a - * table without stats are all kept. Returns the original [[AddFile]]s (matched by path). - * - * Fail-safe: skipping here is a pure optimization over correct (conservative) conflict detection, - * so if building or evaluating the predicate throws we fall back to the default behavior of - * keeping all `files` as conflict candidates rather than failing the commit. - */ - private[delta] def filterFilesMatchingAnyReadPredicate( - files: Seq[AddFile], - dataFiltersPerRead: Seq[Seq[Expression]]): Seq[AddFile] = { - import org.apache.spark.sql.delta.implicits._ - if (files.isEmpty || dataFiltersPerRead.isEmpty || schema.isEmpty) return files - try { - // One skipping predicate per read. A read with no usable predicate (empty or ineligible - // filters) matches every file -> nothing can be skipped, so keep all files without a job. - val perReadPredicates = dataFiltersPerRead.map { dataFilters => - if (dataFilters.isEmpty) None else buildDataSkippingPredicate(dataFilters) - } - if (perReadPredicates.exists(_.isEmpty)) return files - // Survive if the file could match ANY read. Per read, `expr || !verifyStatsForFilter(...)` - // keeps any file whose referenced stats are missing/NULL (mirrors getDataSkippedFiles): only - // skip when stats prove no match. OR the reads so a match against any one keeps the file. - val survivorCondition = perReadPredicates.flatten - .map(pred => pred.expr || !verifyStatsForFilter(pred.referencedStats)) - .reduce(_ || _) - val survivingPaths = recordFrameProfile( - "Delta", "DataSkippingReader.filterFilesMatchingAnyReadPredicate") { - files.toDF(spark) - .withColumn("stats", parseAndDecodeStats(col("stats"))) - .where(survivorCondition) - .select("path") - .collect() - .map(_.getString(0)) - .toSet - } - files.filter(f => survivingPaths.contains(f.path)) - } catch { - case NonFatal(e) => - // Optimization only: never let a skipping failure abort a commit. Fall back to the default - // (feature-off) behavior of treating every added file as a conflict candidate. - logWarning(log"Conflict-time data skipping failed to evaluate; falling back to treating " + - log"all added files as conflict candidates", e) - files - } - } - private def getCorrectDataSkippingType( dataSkippingType: DeltaDataSkippingType): DeltaDataSkippingType = { dataSkippingType