From f5dd903ae4ac04c7bd43c0a564d0bcafa4d49abc Mon Sep 17 00:00:00 2001 From: yangjie01 Date: Thu, 27 Aug 2026 01:42:02 +0800 Subject: [PATCH 1/7] [SPARK-57205][SQL] Let the built-in file sources take part in DSv2 scan merging FileTable declares the SCAN_MERGING table capability, so PlanMerger can fuse two scans of the same file table that differ only in their projected columns and/or pushed filters. Spark rebuilds the merged scan itself; the file sources supply no merge logic. A file source's partition filters are the strictly enforced ones and its data filters are best-effort. So equal-filter/different-column shapes merge under the default configuration, differing data filters need dsv2SymmetricFilterPropagation, and differing partition filters are still declined -- V1 merges those, which is the residual gap. New suite FileSourceV2PlanMergingSuite; regression run over planmerging, Explain, FileBasedDataSource, FileTable, V2 schema-pruning, V2 filter, V2 aggregate-pushdown, DataSourceV2, SameResult, Subquery and AvroV2 suites. --- docs/sql-performance-tuning.md | 2 +- .../execution/datasources/v2/FileTable.scala | 6 +- .../FileSourceV2PlanMergingSuite.scala | 482 ++++++++++++++++++ 3 files changed, 488 insertions(+), 2 deletions(-) create mode 100644 sql/core/src/test/scala/org/apache/spark/sql/execution/planmerging/FileSourceV2PlanMergingSuite.scala diff --git a/docs/sql-performance-tuning.md b/docs/sql-performance-tuning.md index fee8801086c4c..bbb44d3305e89 100644 --- a/docs/sql-performance-tuning.md +++ b/docs/sql-performance-tuning.md @@ -384,7 +384,7 @@ In TPC-DS benchmark runs, enabling symmetric filter propagation made `q9` and `q spark.sql.optimizer.mergeSubplans.filterPropagation.dsv2SymmetricFilterPropagation.enabled false - When true, two DataSource V2 scans that pushed the same strictly enforced filters but carry different best-effort (post-scan) filters can be merged even when spark.sql.optimizer.mergeSubplans.filterPropagation.symmetricFilterPropagation.enabled is false. In this case widening cannot change the set of rows the scan is required to return, as the strict filters are re-pushed unchanged and the enclosing Filter re-checks the rest above the scan. This applies only to V2 sources that opt in to scan merging with the SCAN_MERGING table capability; no built-in source does. + When true, two DataSource V2 scans that pushed the same strictly enforced filters but carry different best-effort (post-scan) filters can be merged even when spark.sql.optimizer.mergeSubplans.filterPropagation.symmetricFilterPropagation.enabled is false. In this case widening cannot change the set of rows the scan is required to return, as the strict filters are re-pushed unchanged and the enclosing Filter re-checks the rest above the scan. This applies only to V2 sources that opt in to scan merging with the SCAN_MERGING table capability. The built-in file formats opt in on their V2 read path, which a format reaches only when it is removed from spark.sql.sources.useV1SourceList; there the strictly enforced filters are the partition filters, so this configuration lets two scans over the same partitions but with different data filters merge. 4.3.0 diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/FileTable.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/FileTable.scala index 8941a4c8d8c7d..0dd5ed2a958df 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/FileTable.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/FileTable.scala @@ -181,5 +181,9 @@ abstract class FileTable( } object FileTable { - private val CAPABILITIES = util.EnumSet.of(BATCH_READ, BATCH_WRITE) + // File tables satisfy the determinism contract SCAN_MERGING requires: `newScanBuilder` returns a + // fresh builder over the table's `fileIndex` and `mergedOptions(options)`, keeping no state from + // earlier calls, so what a scan reads is decided only by the filters pushed and the columns + // pruned on its own builder. + private val CAPABILITIES = util.EnumSet.of(BATCH_READ, BATCH_WRITE, SCAN_MERGING) } diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/planmerging/FileSourceV2PlanMergingSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/planmerging/FileSourceV2PlanMergingSuite.scala new file mode 100644 index 0000000000000..b7ac09e48c4af --- /dev/null +++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/planmerging/FileSourceV2PlanMergingSuite.scala @@ -0,0 +1,482 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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.execution.planmerging + +import org.apache.spark.sql.{DataFrame, QueryTest, Row} +import org.apache.spark.sql.connector.catalog.TableCapability +import org.apache.spark.sql.execution.{ReusedSubqueryExec, SubqueryExec} +import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanHelper +import org.apache.spark.sql.execution.datasources.LogicalRelation +import org.apache.spark.sql.execution.datasources.v2.{DataSourceV2Relation, DataSourceV2ScanRelation, FileTable} +import org.apache.spark.sql.internal.SQLConf +import org.apache.spark.sql.test.SharedSparkSession + +/** + * Scan merging for the built-in file sources on their DSv2 read path (SPARK-57205). + * + * [[FileTable]] declares the `SCAN_MERGING` table capability, so [[PlanMerger]] may fuse two file + * scans of the same table that differ only in their projected columns and/or pushed filters. For a + * file source the strictly enforced filters are the partition filters and the best-effort ones are + * the data filters, which is what decides the shapes that merge here. + * + * The read entry point matters: SQL-on-file and catalog tables resolve to the V1 `FileFormat` + * regardless of `spark.sql.sources.useV1SourceList`, so every test goes through `DataFrameReader` + * and asserts the plan really is V2 before asserting anything about merging. + */ +class FileSourceV2PlanMergingSuite extends QueryTest with SharedSparkSession + with AdaptiveSparkPlanHelper { + import testImplicits._ + + // The built-in formats living in sql/core. Avro is in connector/avro, covered by that module. + private val multiColumnFormats = Seq("parquet", "orc", "json", "csv") + + private val flatSchema = "a long, b long, c long, d long" + + private def writeFlat(format: String, path: String): Unit = + spark.range(0, 20) + .selectExpr("id AS a", "id * 2 AS b", "id % 3 AS c", "id * 3 AS d") + .write.format(format).save(path) + + private def writePartitioned(path: String): Unit = + spark.range(0, 20) + .selectExpr("id AS a", "id * 2 AS b", "id % 3 AS c", "id % 4 AS p") + .write.partitionBy("p").format("parquet").save(path) + + /** + * Registers `path` as a temp view read through the V2 path, or the V1 path if `useV1`. Not named + * `withView`: that name is taken by a varargs helper in `QueryCleanupHelper`, which a call with + * only positional String arguments would silently bind to instead. + */ + private def withFileView[T]( + format: String, + path: String, + useV1: Boolean = false, + schema: Option[String] = None, + options: Map[String, String] = Map.empty, + viewName: String = "t")(f: => T): T = { + withSQLConf(SQLConf.USE_V1_SOURCE_LIST.key -> (if (useV1) format else "")) { + val base = spark.read.format(format).options(options) + val reader = schema.map(s => base.schema(s)).getOrElse(base) + reader.load(path).createOrReplaceTempView(viewName) + try f finally spark.catalog.dropTempView(viewName) + } + } + + private def v2Scans(df: DataFrame): Seq[DataSourceV2ScanRelation] = + df.queryExecution.optimizedPlan.collectWithSubqueries { + case s: DataSourceV2ScanRelation => s + } + + /** + * A merged subquery is referenced once per original subquery, so the logical plan duplicates it + * (physical planning reuses it). Dedupe by canonical form: one distinct scan means the merge + * happened, two means it was declined. + */ + private def distinctScans(df: DataFrame): Int = v2Scans(df).map(_.canonicalized).distinct.length + + private def assertUsesFileSourceV2(df: DataFrame): Unit = { + val plan = df.queryExecution.optimizedPlan + assert(plan.collectWithSubqueries { case r: LogicalRelation => r }.isEmpty, + s"expected the V2 file source path, but the plan has a V1 relation:\n$plan") + val scans = v2Scans(df) + assert(scans.nonEmpty, s"expected a DSv2 file scan:\n$plan") + scans.foreach { s => + assert(s.relation.table.isInstanceOf[FileTable], + s"expected a FileTable, got ${s.relation.table.getClass.getSimpleName}") + } + } + + // A successful merge builds the scan and leaves no bare DataSourceV2Relation behind; a leaked + // deferred scan would show up as an unbuilt placeholder the read path cannot plan. + private def assertNoPlaceholderRelation(df: DataFrame): Unit = + assert( + df.queryExecution.optimizedPlan.collectWithSubqueries { + case r: DataSourceV2Relation => r + }.isEmpty, + s"unbuilt placeholder DataSourceV2Relation left in plan:\n${df.queryExecution.optimizedPlan}") + + /** `(SubqueryExec, ReusedSubqueryExec)` counts, the same measure `PlanMergingSuite` uses. */ + private def subqueryCounts(df: DataFrame): (Int, Int) = { + val plan = df.queryExecution.executedPlan + val subqueries = collectWithSubqueries(plan) { case s: SubqueryExec => s.id } + val reused = collectWithSubqueries(plan) { case rs: ReusedSubqueryExec => rs.child.id } + (subqueries.size, reused.size) + } + + /** + * Runs `query` over the parquet data at `path` on the V1 or V2 read path with both symmetric + * filter propagation configurations on, checks the rows and returns the subquery counts. + */ + private def mergedCounts( + path: String, + query: String, + expected: Row, + useV1: Boolean, + enableAQE: Boolean): (Int, Int) = { + withFileView("parquet", path, useV1 = useV1) { + withSQLConf( + SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> enableAQE.toString, + SQLConf.MERGE_SUBPLANS_SYMMETRIC_FILTER_PROPAGATION_ENABLED.key -> "true", + SQLConf.MERGE_SUBPLANS_DSV2_SYMMETRIC_FILTER_PROPAGATION_ENABLED.key -> "true") { + val df = sql(query) + checkAnswer(df, expected) + subqueryCounts(df) + } + } + } + + test("SPARK-57205: every built-in file table declares SCAN_MERGING") { + Seq("parquet", "orc", "json", "csv", "text").foreach { format => + withClue(s"format=$format: ") { + withTempPath { dir => + val path = dir.getCanonicalPath + spark.range(0, 5).selectExpr("cast(id AS string) AS value") + .write.format(format).save(path) + withSQLConf(SQLConf.USE_V1_SOURCE_LIST.key -> "") { + val relations = spark.read.format(format).load(path) + .queryExecution.analyzed.collect { case r: DataSourceV2Relation => r } + assert(relations.size == 1, s"expected a single DSv2 relation, got $relations") + val table = relations.head.table + assert(table.isInstanceOf[FileTable], s"expected a FileTable, got $table") + assert(table.capabilities().contains(TableCapability.SCAN_MERGING), + s"${table.getClass.getSimpleName} should declare SCAN_MERGING") + } + } + } + } + } + + test("SPARK-57205: merge two file scans that differ only in their projected columns") { + multiColumnFormats.foreach { format => + withClue(s"format=$format: ") { + withTempPath { dir => + val path = dir.getCanonicalPath + writeFlat(format, path) + withFileView(format, path, schema = Some(flatSchema)) { + val df = sql( + """ + |SELECT + | (SELECT sum(a) FROM t WHERE c = 1), + | (SELECT sum(b) FROM t WHERE c = 1) + |""".stripMargin) + + // c is id % 3, so c = 1 selects ids 1, 4, 7, 10, 13, 16 and 19. + checkAnswer(df, Row(70, 140)) + assertUsesFileSourceV2(df) + assert(distinctScans(df) == 1, + s"the two scans should be fused into one:\n${df.queryExecution.optimizedPlan}") + // Both sides carry the same data filter, so no widening is needed and this merges + // under the default configuration. c is read because the filter stays above the scan. + assert(v2Scans(df).head.output.map(_.name).toSet == Set("a", "b", "c"), + s"the merged scan should read the union of both columns; " + + s"got ${v2Scans(df).head.output}") + assertNoPlaceholderRelation(df) + } + } + } + } + } + + test("SPARK-57205: merge two file scans over the same partition filter") { + withTempPath { dir => + val path = dir.getCanonicalPath + writePartitioned(path) + withFileView("parquet", path) { + val df = sql( + """ + |SELECT + | (SELECT sum(a) FROM t WHERE p = 1), + | (SELECT sum(b) FROM t WHERE p = 1) + |""".stripMargin) + + // p is id % 4, so p = 1 selects ids 1, 5, 9, 13 and 17. + checkAnswer(df, Row(45, 90)) + assertUsesFileSourceV2(df) + assert(distinctScans(df) == 1, + s"the two scans should be fused into one:\n${df.queryExecution.optimizedPlan}") + val scan = v2Scans(df).head + // A partition filter is fully enforced by the scan and nothing above it re-checks, so p is + // not read; the rebuilt scan has to push the filter again or it would read all partitions. + assert(scan.output.map(_.name).toSet == Set("a", "b"), + s"the merged scan should read the union of both columns; got ${scan.output}") + assert(scan.pushedFilters.exists(_.references.exists(_.name == "p")), + s"the partition filter should be re-pushed strict onto the merged scan; " + + s"got pushedFilters=${scan.pushedFilters.mkString("[", ", ", "]")}") + assertNoPlaceholderRelation(df) + } + } + } + + test("SPARK-57205: merge three file scans into one") { + withTempPath { dir => + val path = dir.getCanonicalPath + writeFlat("parquet", path) + withFileView("parquet", path) { + val df = sql( + """ + |SELECT + | (SELECT sum(a) FROM t WHERE c = 1), + | (SELECT sum(b) FROM t WHERE c = 1), + | (SELECT sum(d) FROM t WHERE c = 1) + |""".stripMargin) + + checkAnswer(df, Row(70, 140, 210)) + assertUsesFileSourceV2(df) + assert(distinctScans(df) == 1, + s"the three scans should be fused into one:\n${df.queryExecution.optimizedPlan}") + assert(v2Scans(df).head.output.map(_.name).toSet == Set("a", "b", "c", "d"), + s"the merged scan should read the union of all three; got ${v2Scans(df).head.output}") + assertNoPlaceholderRelation(df) + } + } + } + + test("SPARK-57205: merge file scans with differing data filters only when dsv2 symmetric " + + "filter propagation is on") { + Seq(true, false).foreach { dsv2Symmetric => + withClue(s"dsv2SymmetricFilterPropagation=$dsv2Symmetric: ") { + withTempPath { dir => + val path = dir.getCanonicalPath + writeFlat("parquet", path) + withFileView("parquet", path) { + withSQLConf(SQLConf.MERGE_SUBPLANS_DSV2_SYMMETRIC_FILTER_PROPAGATION_ENABLED.key -> + dsv2Symmetric.toString) { + val df = sql( + """ + |SELECT + | (SELECT sum(a) FROM t WHERE a > 10), + | (SELECT sum(b) FROM t WHERE b > 10) + |""".stripMargin) + + // a > 10 selects ids 11 to 19; b is 2 * a, so b > 10 selects ids 6 to 19. + checkAnswer(df, Row(135, 350)) + assertUsesFileSourceV2(df) + // a and b are data columns, so neither scan pushes a strict filter: the strict sets + // are equal and only the OR-widening of the differing best-effort filters gates the + // merge. The enclosing Filter keeps each aggregate exact either way. + assert(distinctScans(df) == (if (dsv2Symmetric) 1 else 2), + s"unexpected scan count:\n${df.queryExecution.optimizedPlan}") + assertNoPlaceholderRelation(df) + } + } + } + } + } + } + + test("SPARK-57205: do not merge file scans with different partition filters") { + withTempPath { dir => + val path = dir.getCanonicalPath + writePartitioned(path) + withFileView("parquet", path) { + // Known gap against V1, which merges this shape: a partition filter is strictly enforced + // by the scan, so widening it to OR would make the merged scan return rows nothing above + // it filters out. Both propagation configs are on to show the merge is declined regardless. + withSQLConf( + SQLConf.MERGE_SUBPLANS_SYMMETRIC_FILTER_PROPAGATION_ENABLED.key -> "true", + SQLConf.MERGE_SUBPLANS_DSV2_SYMMETRIC_FILTER_PROPAGATION_ENABLED.key -> "true") { + val df = sql( + """ + |SELECT + | (SELECT sum(a) FROM t WHERE p = 1), + | (SELECT sum(b) FROM t WHERE p = 2) + |""".stripMargin) + + // p = 1 selects ids 1, 5, 9, 13, 17; p = 2 selects ids 2, 6, 10, 14, 18. + checkAnswer(df, Row(45, 100)) + assertUsesFileSourceV2(df) + assert(distinctScans(df) == 2, + s"scans with different partition filters must not be fused:\n" + + df.queryExecution.optimizedPlan) + } + } + } + } + + test("SPARK-57205: do not merge file scans that read different nested fields") { + withTempPath { dir => + val path = dir.getCanonicalPath + spark.range(0, 20).selectExpr("id AS a", "named_struct('x', id, 'y', id * 2) AS s") + .write.format("parquet").save(path) + Seq(true, false).foreach { nestedPruning => + withClue(s"nestedSchemaPruning=$nestedPruning: ") { + withFileView("parquet", path) { + withSQLConf(SQLConf.NESTED_SCHEMA_PRUNING_ENABLED.key -> nestedPruning.toString) { + val df = sql( + """ + |SELECT + | (SELECT sum(s.x) FROM t), + | (SELECT sum(s.y) FROM t) + |""".stripMargin) + + checkAnswer(df, Row(190, 380)) + assertUsesFileSourceV2(df) + // Nested pruning narrows s to the one field each side reads, so the read column is + // no longer a same-type subset of the relation's s and the merge is declined -- the + // field ordinals in the extractors above the scan are resolved against the narrowed + // type. Without pruning both scans read the whole struct and are simply identical. + assert(distinctScans(df) == (if (nestedPruning) 2 else 1), + s"unexpected scan count:\n${df.queryExecution.optimizedPlan}") + } + } + } + } + } + } + + test("SPARK-57205: do not merge file scans that carry a pushed aggregate") { + withTempPath { dir => + val path = dir.getCanonicalPath + writeFlat("parquet", path) + withFileView("parquet", path) { + withSQLConf(SQLConf.PARQUET_AGGREGATE_PUSHDOWN_ENABLED.key -> "true") { + val df = sql( + """ + |SELECT + | (SELECT max(a) FROM t), + | (SELECT max(b) FROM t) + |""".stripMargin) + + checkAnswer(df, Row(19, 38)) + assertUsesFileSourceV2(df) + // A pushed aggregate is built on a branch of V2ScanRelationPushDown that never marks the + // scan mergeable, so the merge is declined before the capability is consulted. + assert(distinctScans(df) == 2, + s"scans with a pushed aggregate must not be fused:\n" + + df.queryExecution.optimizedPlan) + } + } + } + } + + test("SPARK-57205: do not merge file scans of different tables") { + withTempPath { dir1 => + withTempPath { dir2 => + writeFlat("parquet", dir1.getCanonicalPath) + writeFlat("parquet", dir2.getCanonicalPath) + withFileView("parquet", dir1.getCanonicalPath, viewName = "t1") { + withFileView("parquet", dir2.getCanonicalPath, viewName = "t2") { + val df = sql( + """ + |SELECT + | (SELECT sum(a) FROM t1 WHERE c = 1), + | (SELECT sum(b) FROM t2 WHERE c = 1) + |""".stripMargin) + + checkAnswer(df, Row(70, 140)) + assertUsesFileSourceV2(df) + assert(distinctScans(df) == 2, + s"scans of different tables must remain separate:\n" + + df.queryExecution.optimizedPlan) + } + } + } + } + } + + test("SPARK-57205: V1 and V2 file sources merge the same subquery shapes") { + val shapes = Seq( + ("differing projected columns", + """ + |SELECT + | (SELECT sum(a) FROM t WHERE c = 1), + | (SELECT sum(b) FROM t WHERE c = 1) + |""".stripMargin, + Row(70, 140)), + ("differing data filters", + """ + |SELECT + | (SELECT sum(a) FROM t WHERE a > 10), + | (SELECT sum(b) FROM t WHERE b > 10) + |""".stripMargin, + Row(135, 350)), + ("same partition filter, differing data filters", + """ + |SELECT + | (SELECT sum(a) FROM t WHERE p = 1 AND a > 4), + | (SELECT sum(b) FROM t WHERE p = 1 AND b > 20) + |""".stripMargin, + Row(44, 60))) + + withTempPath { dir => + val path = dir.getCanonicalPath + writePartitioned(path) + shapes.foreach { case (shape, query, expected) => + Seq(false, true).foreach { enableAQE => + withClue(s"$shape, AQE=$enableAQE: ") { + val v1 = mergedCounts(path, query, expected, useV1 = true, enableAQE) + val v2 = mergedCounts(path, query, expected, useV1 = false, enableAQE) + assert(v1 == v2, s"V1 and V2 should merge alike; V1 got $v1, V2 got $v2") + assert(v1 == ((1, 1)), s"both paths should merge into a single subquery; got $v1") + } + } + } + } + } + + test("SPARK-57205: V1 merges differing partition filters, V2 does not") { + withTempPath { dir => + val path = dir.getCanonicalPath + writePartitioned(path) + val query = + """ + |SELECT + | (SELECT sum(a) FROM t WHERE p = 1), + | (SELECT sum(b) FROM t WHERE p = 2) + |""".stripMargin + Seq(false, true).foreach { enableAQE => + withClue(s"AQE=$enableAQE: ") { + // The one shape where the two paths still differ. V1 keeps the partition filter in a + // Filter node until physical planning, so symmetric propagation can widen it; on the V2 + // path V2ScanRelationPushDown has already absorbed it into the scan as a strict filter by + // the time MergeSubplans runs, and strict filters have to be equal to merge. Same rows. + assert(mergedCounts(path, query, Row(45, 100), useV1 = true, enableAQE) == ((1, 1))) + assert(mergedCounts(path, query, Row(45, 100), useV1 = false, enableAQE) == ((2, 0))) + } + } + } + } + + test("SPARK-57205: merging widens the columns a CSV scan parses, as it already does on V1") { + withTempPath { dir => + val path = dir.getCanonicalPath + Seq("0,0", "1,10", "2,BAD", "3,30", "4,40").toDS().write.text(path) + val query = + """ + |SELECT + | (SELECT sum(a) FROM t), + | (SELECT sum(b) FROM t) + |""".stripMargin + + def rows(useV1: Boolean): Seq[Row] = + withFileView("csv", path, useV1 = useV1, schema = Some("a long, b long"), + options = Map("mode" -> "DROPMALFORMED")) { + sql(query).collect().toSeq + } + + // The CSV parser only sees the columns the scan requests + // (spark.sql.csv.parser.columnPruning.enabled), so which columns a scan reads decides which + // malformed values it notices. Merging makes one scan parse both a and b, and the record + // malformed in b is then dropped for both aggregates: sum(a) is 8, not the 10 two separate + // scans would produce. V1 already read the union after merging, so this is the V2 path + // catching up rather than a new meaning for a merge. + assert(rows(useV1 = false) == Seq(Row(8, 80))) + assert(rows(useV1 = true) == rows(useV1 = false)) + } + } +} From 87e94eebcdcaceffa4a07aebaedac1bad2d39e6a Mon Sep 17 00:00:00 2001 From: yangjie01 Date: Thu, 27 Aug 2026 15:10:01 +0800 Subject: [PATCH 2/7] Address review: migration guide entry, JSON parse-mode coverage, tighter comments Add a migration-guide entry: on the V2 read path a format reaches after being removed from useV1SourceList, merging can change results for CSV and JSON under mode=DROPMALFORMED, since the merged scan parses the union of both scans' columns. Extend the parse-mode test from CSV to CSV and JSON. JSON drops the record too, so enablePartialResults does not change the outcome; previously this was only inferred. Tighten the comments the change adds. Two carried claims the code does not support: FileTable's said what a scan reads is decided "only" by the pushed filters and pruned columns, dropping the options-constant condition the SCAN_MERGING contract states; and the format-list comment claimed text is not in sql/core and that connector/avro covers scan merging, which it does not. The rest were universal claims, sentence fragments and filler. --- docs/sql-migration-guide.md | 1 + .../execution/datasources/v2/FileTable.scala | 8 +- .../FileSourceV2PlanMergingSuite.scala | 82 +++++++++++-------- 3 files changed, 54 insertions(+), 37 deletions(-) diff --git a/docs/sql-migration-guide.md b/docs/sql-migration-guide.md index 40c9784b986f6..bba82d3c2c7a4 100644 --- a/docs/sql-migration-guide.md +++ b/docs/sql-migration-guide.md @@ -25,6 +25,7 @@ license: | ## Upgrading from Spark SQL 4.3 to 4.4 - Since Spark 4.4, for storage-partitioned joins, `spark.sql.requireAllClusterKeysForCoPartition` requires every join key to be covered by some partition key instead of matching the partition keys positionally. As a result, a join-key column partitioned by more than one transform no longer prevents shuffle elimination, and `spark.sql.sources.v2.bucketing.allowKeysSubsetOfPartitionKeys.enabled` no longer additionally requires `spark.sql.requireAllClusterKeysForCoPartition` to be `false` when the join keys are a subset of the partition keys. As before, when the partition keys cover only part of the join keys, eliminating the shuffle still requires `spark.sql.requireAllClusterKeysForCoPartition` to be `false`. +- Since Spark 4.4, the built-in file formats declare the `SCAN_MERGING` table capability on their DataSource V2 read path, so two scans of the same file table that differ only in their projected columns can be merged into a single scan reading the union of those columns. A format takes that path only when it is removed from `spark.sql.sources.useV1SourceList`, and merging there now matches what the V1 path already did. For CSV and JSON this also changes which records count as malformed, because the parser is handed only the columns the scan reads: with `mode` set to `DROPMALFORMED`, a record malformed only in the columns the other scan reads is now dropped for both. To restore the previous behavior, add the format back to `spark.sql.sources.useV1SourceList`, or disable subplan merging entirely by adding `org.apache.spark.sql.execution.planmerging.MergeSubplans` to `spark.sql.optimizer.excludedRules`. ## Upgrading from Spark SQL 4.2 to 4.3 diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/FileTable.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/FileTable.scala index 0dd5ed2a958df..e3fe9a0f56d3b 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/FileTable.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/FileTable.scala @@ -181,9 +181,9 @@ abstract class FileTable( } object FileTable { - // File tables satisfy the determinism contract SCAN_MERGING requires: `newScanBuilder` returns a - // fresh builder over the table's `fileIndex` and `mergedOptions(options)`, keeping no state from - // earlier calls, so what a scan reads is decided only by the filters pushed and the columns - // pruned on its own builder. + // A file table meets the determinism contract SCAN_MERGING requires: `fileIndex` is a lazy val, + // so every scan built from this table lists the same files, and `newScanBuilder` returns a fresh + // builder over `mergedOptions(options)`. The same options, pushed filters and pruned columns + // therefore rebuild an equivalent scan. private val CAPABILITIES = util.EnumSet.of(BATCH_READ, BATCH_WRITE, SCAN_MERGING) } diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/planmerging/FileSourceV2PlanMergingSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/planmerging/FileSourceV2PlanMergingSuite.scala index b7ac09e48c4af..5c2e94d15b460 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/execution/planmerging/FileSourceV2PlanMergingSuite.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/planmerging/FileSourceV2PlanMergingSuite.scala @@ -32,17 +32,17 @@ import org.apache.spark.sql.test.SharedSparkSession * [[FileTable]] declares the `SCAN_MERGING` table capability, so [[PlanMerger]] may fuse two file * scans of the same table that differ only in their projected columns and/or pushed filters. For a * file source the strictly enforced filters are the partition filters and the best-effort ones are - * the data filters, which is what decides the shapes that merge here. + * the data filters. * - * The read entry point matters: SQL-on-file and catalog tables resolve to the V1 `FileFormat` - * regardless of `spark.sql.sources.useV1SourceList`, so every test goes through `DataFrameReader` - * and asserts the plan really is V2 before asserting anything about merging. + * SQL-on-file and catalog tables resolve to the V1 `FileFormat` regardless of + * `spark.sql.sources.useV1SourceList`, so every test goes through `DataFrameReader` and asserts the + * plan is V2 before asserting anything about merging. */ class FileSourceV2PlanMergingSuite extends QueryTest with SharedSparkSession with AdaptiveSparkPlanHelper { import testImplicits._ - // The built-in formats living in sql/core. Avro is in connector/avro, covered by that module. + // The multi-column formats in sql/core; text is single-column and Avro lives in connector/avro. private val multiColumnFormats = Seq("parquet", "orc", "json", "csv") private val flatSchema = "a long, b long, c long, d long" @@ -329,7 +329,8 @@ class FileSourceV2PlanMergingSuite extends QueryTest with SharedSparkSession // Nested pruning narrows s to the one field each side reads, so the read column is // no longer a same-type subset of the relation's s and the merge is declined -- the // field ordinals in the extractors above the scan are resolved against the narrowed - // type. Without pruning both scans read the whole struct and are simply identical. + // type. Without pruning both scans read the whole struct and are canonically equal, + // so they merge on PlanMerger's identical-plan path, which needs no capability. assert(distinctScans(df) == (if (nestedPruning) 2 else 1), s"unexpected scan count:\n${df.queryExecution.optimizedPlan}") } @@ -441,10 +442,10 @@ class FileSourceV2PlanMergingSuite extends QueryTest with SharedSparkSession |""".stripMargin Seq(false, true).foreach { enableAQE => withClue(s"AQE=$enableAQE: ") { - // The one shape where the two paths still differ. V1 keeps the partition filter in a - // Filter node until physical planning, so symmetric propagation can widen it; on the V2 - // path V2ScanRelationPushDown has already absorbed it into the scan as a strict filter by - // the time MergeSubplans runs, and strict filters have to be equal to merge. Same rows. + // V1 keeps the partition filter in a Filter node until physical planning, so symmetric + // propagation can widen it; on the V2 path V2ScanRelationPushDown has already pushed it + // into the scan as a strict filter by the time MergeSubplans runs, and strict filters + // have to be equal to merge. Both paths return the same rows. assert(mergedCounts(path, query, Row(45, 100), useV1 = true, enableAQE) == ((1, 1))) assert(mergedCounts(path, query, Row(45, 100), useV1 = false, enableAQE) == ((2, 0))) } @@ -452,31 +453,46 @@ class FileSourceV2PlanMergingSuite extends QueryTest with SharedSparkSession } } - test("SPARK-57205: merging widens the columns a CSV scan parses, as it already does on V1") { - withTempPath { dir => - val path = dir.getCanonicalPath - Seq("0,0", "1,10", "2,BAD", "3,30", "4,40").toDS().write.text(path) - val query = - """ - |SELECT - | (SELECT sum(a) FROM t), - | (SELECT sum(b) FROM t) - |""".stripMargin + test("SPARK-57205: merging widens the columns a text-based scan parses, as it does on V1") { + // A record malformed only in the column the other scan reads. The parsers are handed just the + // requested columns (spark.sql.csv.parser.columnPruning.enabled for CSV), so which columns a + // scan reads decides which malformed values it notices. + val cases = Seq( + "csv" -> Seq("0,0", "1,10", "2,BAD", "3,30", "4,40"), + "json" -> Seq( + """{"a":0,"b":0}""", + """{"a":1,"b":10}""", + """{"a":2,"b":"BAD"}""", + """{"a":3,"b":30}""", + """{"a":4,"b":40}""")) + val query = + """ + |SELECT + | (SELECT sum(a) FROM t), + | (SELECT sum(b) FROM t) + |""".stripMargin + + cases.foreach { case (format, lines) => + withClue(s"format=$format: ") { + withTempPath { dir => + val path = dir.getCanonicalPath + lines.toDS().write.text(path) - def rows(useV1: Boolean): Seq[Row] = - withFileView("csv", path, useV1 = useV1, schema = Some("a long, b long"), - options = Map("mode" -> "DROPMALFORMED")) { - sql(query).collect().toSeq - } + def rows(useV1: Boolean): Seq[Row] = + withFileView(format, path, useV1 = useV1, schema = Some("a long, b long"), + options = Map("mode" -> "DROPMALFORMED")) { + sql(query).collect().toSeq + } - // The CSV parser only sees the columns the scan requests - // (spark.sql.csv.parser.columnPruning.enabled), so which columns a scan reads decides which - // malformed values it notices. Merging makes one scan parse both a and b, and the record - // malformed in b is then dropped for both aggregates: sum(a) is 8, not the 10 two separate - // scans would produce. V1 already read the union after merging, so this is the V2 path - // catching up rather than a new meaning for a merge. - assert(rows(useV1 = false) == Seq(Row(8, 80))) - assert(rows(useV1 = true) == rows(useV1 = false)) + // Merging makes one scan parse both a and b, so the record malformed in b is dropped for + // both aggregates and sum(a) is 8, not the 10 two separate scans produce. The merged scan + // therefore reads fewer rows than the a-only scan did, which SCAN_MERGING's "superset of + // their rows" premise does not allow. V1 already read the union after merging, so the V2 + // path now matches it. + assert(rows(useV1 = false) == Seq(Row(8, 80))) + assert(rows(useV1 = true) == rows(useV1 = false)) + } + } } } } From 70204f964761017d829ca81b3b9e0b8c0e47415c Mon Sep 17 00:00:00 2001 From: yangjie01 Date: Fri, 28 Aug 2026 13:32:47 +0800 Subject: [PATCH 3/7] Address PR review: non-vacuous assertions, shared helpers, Avro coverage The nested-fields test asserted distinctScans on both arms, but with nested pruning off the two scans read the identical whole struct, so they canonicalize equal whether or not the merge happened and the assertion could not fail. That arm now asserts subqueryCounts, (1, 1) merged against (2, 0) declined; distinctScans stays on the pruning-on arm, where the two readSchemas differ. Drop assertNoPlaceholderRelation and its four call sites. DataSourceV2Strategy's only batch-read case matches DataSourceV2ScanRelation, so a leaked bare DataSourceV2Relation fails planning inside the preceding checkAnswer and the helper can never fire. Move v2Scans and distinctScans into V2ScanMergingTestHelper, shared with DSv2PlanMergingSuite, which held a copy of v2Scans. Stop re-running queries: bind the parse-mode result once instead of calling the helper in both asserts, and write the data outside the flag loop in the differing-data-filters test. Cover Avro in AvroV2Suite: AvroTable inherits the capability through FileTable, and connector/avro had no scan-merging test. Give FileTable a class-level scaladoc stating the contract subclasses inherit with SCAN_MERGING, since the base class cannot enforce what newScanBuilder does. Measure the other two CSV/JSON parse modes rather than assuming. PERMISSIVE returns the same rows merged and unmerged, because the parser keeps the fields it did parse, and FAILFAST throws either way, because the merged scan reads the union of both column sets and the scan reading the malformed column already throws alone. The migration guide records that neither changes, and both are now assertions. --- .../org/apache/spark/sql/avro/AvroSuite.scala | 35 +++++++- docs/sql-migration-guide.md | 2 +- .../execution/datasources/v2/FileTable.scala | 15 +++- .../planmerging/DSv2PlanMergingSuite.scala | 9 +- .../FileSourceV2PlanMergingSuite.scala | 87 +++++++++---------- .../planmerging/V2ScanMergingTestHelper.scala | 40 +++++++++ 6 files changed, 129 insertions(+), 59 deletions(-) create mode 100644 sql/core/src/test/scala/org/apache/spark/sql/execution/planmerging/V2ScanMergingTestHelper.scala diff --git a/connector/avro/src/test/scala/org/apache/spark/sql/avro/AvroSuite.scala b/connector/avro/src/test/scala/org/apache/spark/sql/avro/AvroSuite.scala index 8134017738346..26bd271bb83f9 100644 --- a/connector/avro/src/test/scala/org/apache/spark/sql/avro/AvroSuite.scala +++ b/connector/avro/src/test/scala/org/apache/spark/sql/avro/AvroSuite.scala @@ -42,9 +42,10 @@ import org.apache.spark.sql.catalyst.expressions.AttributeReference import org.apache.spark.sql.catalyst.plans.logical.Filter import org.apache.spark.sql.catalyst.util.DateTimeTestUtils import org.apache.spark.sql.catalyst.util.DateTimeTestUtils.{withDefaultTimeZone, LA, UTC} +import org.apache.spark.sql.connector.catalog.TableCapability import org.apache.spark.sql.execution.{FormattedMode, SparkPlan} import org.apache.spark.sql.execution.datasources.{CommonFileDataSourceSuite, DataSource, FilePartition} -import org.apache.spark.sql.execution.datasources.v2.{BatchScanExec, FileDataSourceV2, FileTable} +import org.apache.spark.sql.execution.datasources.v2.{BatchScanExec, DataSourceV2ScanRelation, FileDataSourceV2, FileTable} import org.apache.spark.sql.functions._ import org.apache.spark.sql.internal.LegacyBehaviorPolicy import org.apache.spark.sql.internal.LegacyBehaviorPolicy._ @@ -3937,6 +3938,38 @@ class AvroV2Suite extends AvroSuite with ExplainSuiteHelper { s"V2 formatName '${v2Table.formatName}' != V1 toString '${v1Format.toString}'") } + test("SPARK-57205: Avro V2 inherits SCAN_MERGING and merges scans differing only in columns") { + val v2Provider = DataSource.lookupDataSourceV2("avro", spark.sessionState.conf) + assert(v2Provider.isDefined) + val v2Table = v2Provider.get.asInstanceOf[FileDataSourceV2].getTable( + new StructType(), Array.empty, JCollections.emptyMap[String, String]()) + assert(v2Table.capabilities().contains(TableCapability.SCAN_MERGING)) + + withTempPath { dir => + val path = dir.getCanonicalPath + spark.range(0, 20).selectExpr("id AS a", "id * 2 AS b", "id % 3 AS c") + .write.format("avro").save(path) + withTempView("avro_scan_merging") { + spark.read.format("avro").load(path).createOrReplaceTempView("avro_scan_merging") + val df = sql( + """ + |SELECT + | (SELECT sum(a) FROM avro_scan_merging WHERE c = 1), + | (SELECT sum(b) FROM avro_scan_merging WHERE c = 1) + |""".stripMargin) + checkAnswer(df, Row(70, 140)) + val scans = df.queryExecution.optimizedPlan.collectWithSubqueries { + case s: DataSourceV2ScanRelation => s + } + assert(scans.map(_.canonicalized).distinct.length == 1, + s"the two Avro scans should be fused into one:\n${df.queryExecution.optimizedPlan}") + // c is read because the filter stays above the merged scan. + assert(scans.head.output.map(_.name).toSet == Set("a", "b", "c"), + s"the merged scan should read the union of both columns; got ${scans.head.output}") + } + } + } + test("Geospatial types are not supported in Avro") { withTempDir { dir => // Temporary directory for writing the test data. diff --git a/docs/sql-migration-guide.md b/docs/sql-migration-guide.md index bba82d3c2c7a4..d953fa826ef3b 100644 --- a/docs/sql-migration-guide.md +++ b/docs/sql-migration-guide.md @@ -25,7 +25,7 @@ license: | ## Upgrading from Spark SQL 4.3 to 4.4 - Since Spark 4.4, for storage-partitioned joins, `spark.sql.requireAllClusterKeysForCoPartition` requires every join key to be covered by some partition key instead of matching the partition keys positionally. As a result, a join-key column partitioned by more than one transform no longer prevents shuffle elimination, and `spark.sql.sources.v2.bucketing.allowKeysSubsetOfPartitionKeys.enabled` no longer additionally requires `spark.sql.requireAllClusterKeysForCoPartition` to be `false` when the join keys are a subset of the partition keys. As before, when the partition keys cover only part of the join keys, eliminating the shuffle still requires `spark.sql.requireAllClusterKeysForCoPartition` to be `false`. -- Since Spark 4.4, the built-in file formats declare the `SCAN_MERGING` table capability on their DataSource V2 read path, so two scans of the same file table that differ only in their projected columns can be merged into a single scan reading the union of those columns. A format takes that path only when it is removed from `spark.sql.sources.useV1SourceList`, and merging there now matches what the V1 path already did. For CSV and JSON this also changes which records count as malformed, because the parser is handed only the columns the scan reads: with `mode` set to `DROPMALFORMED`, a record malformed only in the columns the other scan reads is now dropped for both. To restore the previous behavior, add the format back to `spark.sql.sources.useV1SourceList`, or disable subplan merging entirely by adding `org.apache.spark.sql.execution.planmerging.MergeSubplans` to `spark.sql.optimizer.excludedRules`. +- Since Spark 4.4, the built-in file formats declare the `SCAN_MERGING` table capability on their DataSource V2 read path, so two scans of the same file table that differ only in their projected columns can be merged into a single scan reading the union of those columns. A format takes that path only when it is removed from `spark.sql.sources.useV1SourceList`, and for these shapes merging there now matches what the V1 path already did. For CSV and JSON this also changes which records count as malformed, because the parser is handed only the columns the scan reads: with `mode` set to `DROPMALFORMED`, a record malformed only in the columns the other scan reads is now dropped for both. `PERMISSIVE` keeps the fields it did parse and `FAILFAST` rejects such a record either way, so neither of those modes changes. To restore the previous behavior, add the format back to `spark.sql.sources.useV1SourceList`, or disable subplan merging entirely by adding `org.apache.spark.sql.execution.planmerging.MergeSubplans` to `spark.sql.optimizer.excludedRules`. ## Upgrading from Spark SQL 4.2 to 4.3 diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/FileTable.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/FileTable.scala index e3fe9a0f56d3b..b8c8c344abe10 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/FileTable.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/FileTable.scala @@ -37,6 +37,14 @@ import org.apache.spark.sql.util.CaseInsensitiveStringMap import org.apache.spark.sql.util.SchemaUtils import org.apache.spark.util.ArrayImplicits._ +/** + * A [[Table]] backed by files. + * + * Subclasses inherit the `SCAN_MERGING` capability, which holds them to this: with the scan options + * held constant, what a scan reads is determined by the filters pushed and the columns pruned on + * its builder. A subclass whose `newScanBuilder` does not meet that has to override + * [[capabilities]] to drop `SCAN_MERGING`, or Spark may fuse two of its scans into one. + */ abstract class FileTable( sparkSession: SparkSession, options: CaseInsensitiveStringMap, @@ -181,9 +189,8 @@ abstract class FileTable( } object FileTable { - // A file table meets the determinism contract SCAN_MERGING requires: `fileIndex` is a lazy val, - // so every scan built from this table lists the same files, and `newScanBuilder` returns a fresh - // builder over `mergedOptions(options)`. The same options, pushed filters and pruned columns - // therefore rebuild an equivalent scan. + // The built-in file tables meet the SCAN_MERGING contract documented on FileTable: `fileIndex` is + // a lazy val, so every scan built from one table lists the same files, and `newScanBuilder` + // returns a fresh builder over `mergedOptions(options)`. private val CAPABILITIES = util.EnumSet.of(BATCH_READ, BATCH_WRITE, SCAN_MERGING) } diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/planmerging/DSv2PlanMergingSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/planmerging/DSv2PlanMergingSuite.scala index 86bbe951ab9b1..b426b27acac18 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/execution/planmerging/DSv2PlanMergingSuite.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/planmerging/DSv2PlanMergingSuite.scala @@ -23,7 +23,7 @@ import org.apache.spark.sql.{DataFrame, QueryTest, Row} import org.apache.spark.sql.catalyst.plans.physical.KeyedPartitioning import org.apache.spark.sql.connector.FakeV2ProviderWithCustomSchema import org.apache.spark.sql.connector.catalog.{InMemoryScanMergingPartitionFilterCatalog, InMemoryScanMergingReportingCatalog} -import org.apache.spark.sql.execution.datasources.v2.{BatchScanExec, DataSourceV2Relation, DataSourceV2ScanRelation} +import org.apache.spark.sql.execution.datasources.v2.{BatchScanExec, DataSourceV2Relation} import org.apache.spark.sql.internal.SQLConf import org.apache.spark.sql.test.SharedSparkSession @@ -37,7 +37,7 @@ import org.apache.spark.sql.test.SharedSparkSession * mis-classify it as non-strict and decline the merge (leaving two scans). */ class DSv2PlanMergingSuite extends QueryTest with SharedSparkSession - with BeforeAndAfter { + with BeforeAndAfter with V2ScanMergingTestHelper { private val v2Source = classOf[FakeV2ProviderWithCustomSchema].getName private val tbl = "scanmerge.t" @@ -56,11 +56,6 @@ class DSv2PlanMergingSuite extends QueryTest with SharedSparkSession spark.conf.unset("spark.sql.catalog.scanmergereport") } - private def v2Scans(df: DataFrame): Seq[DataSourceV2ScanRelation] = - df.queryExecution.optimizedPlan.collectWithSubqueries { - case s: DataSourceV2ScanRelation => s - } - // A successful DSv2 merge builds the scan and leaves NO bare DataSourceV2Relation in the plan. // A leaked deferred scan (e.g. if a future recursion arm forwarded `deferredScan` without // building it) would surface as an unbuilt placeholder relation the read path cannot plan -- diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/planmerging/FileSourceV2PlanMergingSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/planmerging/FileSourceV2PlanMergingSuite.scala index 5c2e94d15b460..9d7352aa7780b 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/execution/planmerging/FileSourceV2PlanMergingSuite.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/planmerging/FileSourceV2PlanMergingSuite.scala @@ -17,12 +17,13 @@ package org.apache.spark.sql.execution.planmerging +import org.apache.spark.SparkException import org.apache.spark.sql.{DataFrame, QueryTest, Row} import org.apache.spark.sql.connector.catalog.TableCapability import org.apache.spark.sql.execution.{ReusedSubqueryExec, SubqueryExec} import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanHelper import org.apache.spark.sql.execution.datasources.LogicalRelation -import org.apache.spark.sql.execution.datasources.v2.{DataSourceV2Relation, DataSourceV2ScanRelation, FileTable} +import org.apache.spark.sql.execution.datasources.v2.{DataSourceV2Relation, FileTable} import org.apache.spark.sql.internal.SQLConf import org.apache.spark.sql.test.SharedSparkSession @@ -39,7 +40,7 @@ import org.apache.spark.sql.test.SharedSparkSession * plan is V2 before asserting anything about merging. */ class FileSourceV2PlanMergingSuite extends QueryTest with SharedSparkSession - with AdaptiveSparkPlanHelper { + with AdaptiveSparkPlanHelper with V2ScanMergingTestHelper { import testImplicits._ // The multi-column formats in sql/core; text is single-column and Avro lives in connector/avro. @@ -77,18 +78,6 @@ class FileSourceV2PlanMergingSuite extends QueryTest with SharedSparkSession } } - private def v2Scans(df: DataFrame): Seq[DataSourceV2ScanRelation] = - df.queryExecution.optimizedPlan.collectWithSubqueries { - case s: DataSourceV2ScanRelation => s - } - - /** - * A merged subquery is referenced once per original subquery, so the logical plan duplicates it - * (physical planning reuses it). Dedupe by canonical form: one distinct scan means the merge - * happened, two means it was declined. - */ - private def distinctScans(df: DataFrame): Int = v2Scans(df).map(_.canonicalized).distinct.length - private def assertUsesFileSourceV2(df: DataFrame): Unit = { val plan = df.queryExecution.optimizedPlan assert(plan.collectWithSubqueries { case r: LogicalRelation => r }.isEmpty, @@ -101,15 +90,6 @@ class FileSourceV2PlanMergingSuite extends QueryTest with SharedSparkSession } } - // A successful merge builds the scan and leaves no bare DataSourceV2Relation behind; a leaked - // deferred scan would show up as an unbuilt placeholder the read path cannot plan. - private def assertNoPlaceholderRelation(df: DataFrame): Unit = - assert( - df.queryExecution.optimizedPlan.collectWithSubqueries { - case r: DataSourceV2Relation => r - }.isEmpty, - s"unbuilt placeholder DataSourceV2Relation left in plan:\n${df.queryExecution.optimizedPlan}") - /** `(SubqueryExec, ReusedSubqueryExec)` counts, the same measure `PlanMergingSuite` uses. */ private def subqueryCounts(df: DataFrame): (Int, Int) = { val plan = df.queryExecution.executedPlan @@ -185,7 +165,6 @@ class FileSourceV2PlanMergingSuite extends QueryTest with SharedSparkSession assert(v2Scans(df).head.output.map(_.name).toSet == Set("a", "b", "c"), s"the merged scan should read the union of both columns; " + s"got ${v2Scans(df).head.output}") - assertNoPlaceholderRelation(df) } } } @@ -217,7 +196,6 @@ class FileSourceV2PlanMergingSuite extends QueryTest with SharedSparkSession assert(scan.pushedFilters.exists(_.references.exists(_.name == "p")), s"the partition filter should be re-pushed strict onto the merged scan; " + s"got pushedFilters=${scan.pushedFilters.mkString("[", ", ", "]")}") - assertNoPlaceholderRelation(df) } } } @@ -241,18 +219,17 @@ class FileSourceV2PlanMergingSuite extends QueryTest with SharedSparkSession s"the three scans should be fused into one:\n${df.queryExecution.optimizedPlan}") assert(v2Scans(df).head.output.map(_.name).toSet == Set("a", "b", "c", "d"), s"the merged scan should read the union of all three; got ${v2Scans(df).head.output}") - assertNoPlaceholderRelation(df) } } } test("SPARK-57205: merge file scans with differing data filters only when dsv2 symmetric " + "filter propagation is on") { - Seq(true, false).foreach { dsv2Symmetric => - withClue(s"dsv2SymmetricFilterPropagation=$dsv2Symmetric: ") { - withTempPath { dir => - val path = dir.getCanonicalPath - writeFlat("parquet", path) + withTempPath { dir => + val path = dir.getCanonicalPath + writeFlat("parquet", path) + Seq(true, false).foreach { dsv2Symmetric => + withClue(s"dsv2SymmetricFilterPropagation=$dsv2Symmetric: ") { withFileView("parquet", path) { withSQLConf(SQLConf.MERGE_SUBPLANS_DSV2_SYMMETRIC_FILTER_PROPAGATION_ENABLED.key -> dsv2Symmetric.toString) { @@ -271,7 +248,6 @@ class FileSourceV2PlanMergingSuite extends QueryTest with SharedSparkSession // merge. The enclosing Filter keeps each aggregate exact either way. assert(distinctScans(df) == (if (dsv2Symmetric) 1 else 2), s"unexpected scan count:\n${df.queryExecution.optimizedPlan}") - assertNoPlaceholderRelation(df) } } } @@ -328,11 +304,17 @@ class FileSourceV2PlanMergingSuite extends QueryTest with SharedSparkSession assertUsesFileSourceV2(df) // Nested pruning narrows s to the one field each side reads, so the read column is // no longer a same-type subset of the relation's s and the merge is declined -- the - // field ordinals in the extractors above the scan are resolved against the narrowed - // type. Without pruning both scans read the whole struct and are canonically equal, - // so they merge on PlanMerger's identical-plan path, which needs no capability. - assert(distinctScans(df) == (if (nestedPruning) 2 else 1), - s"unexpected scan count:\n${df.queryExecution.optimizedPlan}") + // field ordinals in the extractors above the scan resolve against the narrowed type. + // Without pruning both scans read the whole struct and merge on PlanMerger's + // identical-plan path, which needs no capability. Two whole-struct scans canonicalize + // equal, so distinctScans cannot tell that merge from a decline; subqueryCounts can. + val expectedCounts = if (nestedPruning) (2, 0) else (1, 1) + assert(subqueryCounts(df) == expectedCounts, + s"unexpected subquery counts:\n${df.queryExecution.executedPlan}") + if (nestedPruning) { + assert(distinctScans(df) == 2, + s"the pruned scans should stay separate:\n${df.queryExecution.optimizedPlan}") + } } } } @@ -478,19 +460,32 @@ class FileSourceV2PlanMergingSuite extends QueryTest with SharedSparkSession val path = dir.getCanonicalPath lines.toDS().write.text(path) - def rows(useV1: Boolean): Seq[Row] = + def rows(mode: String, useV1: Boolean): Seq[Row] = withFileView(format, path, useV1 = useV1, schema = Some("a long, b long"), - options = Map("mode" -> "DROPMALFORMED")) { + options = Map("mode" -> mode)) { sql(query).collect().toSeq } - // Merging makes one scan parse both a and b, so the record malformed in b is dropped for - // both aggregates and sum(a) is 8, not the 10 two separate scans produce. The merged scan - // therefore reads fewer rows than the a-only scan did, which SCAN_MERGING's "superset of - // their rows" premise does not allow. V1 already read the union after merging, so the V2 - // path now matches it. - assert(rows(useV1 = false) == Seq(Row(8, 80))) - assert(rows(useV1 = true) == rows(useV1 = false)) + // Merging makes one scan parse both a and b, so DROPMALFORMED drops the record malformed + // in b for both aggregates and sum(a) is 8, not the 10 two separate scans produce. The + // merged scan therefore reads fewer rows than the a-only scan did, which SCAN_MERGING's + // "superset of their rows" premise does not allow. V1 already read the union after + // merging, so the V2 path now matches it. + val droppedV2 = rows("DROPMALFORMED", useV1 = false) + assert(droppedV2 == Seq(Row(8, 80))) + assert(rows("DROPMALFORMED", useV1 = true) == droppedV2) + + // PERMISSIVE keeps the fields it did parse, so widening the parsed columns leaves both + // aggregates on all five records. + val permissiveV2 = rows("PERMISSIVE", useV1 = false) + assert(permissiveV2 == Seq(Row(10, 80))) + assert(rows("PERMISSIVE", useV1 = true) == permissiveV2) + + // FAILFAST throws whether or not the scans merged: the merged scan reads the union of the + // two column sets, so the scan that reads b already throws on its own. + Seq(true, false).foreach { useV1 => + intercept[SparkException](rows("FAILFAST", useV1 = useV1)) + } } } } diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/planmerging/V2ScanMergingTestHelper.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/planmerging/V2ScanMergingTestHelper.scala new file mode 100644 index 0000000000000..a6b1154ae45bd --- /dev/null +++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/planmerging/V2ScanMergingTestHelper.scala @@ -0,0 +1,40 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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.execution.planmerging + +import org.apache.spark.sql.DataFrame +import org.apache.spark.sql.execution.datasources.v2.DataSourceV2ScanRelation + +/** + * Collects the DSv2 scans of a plan for the scan-merging suites in this package. Shared so that a + * change to how merged scans are collected reaches every suite that measures merging. + */ +private[planmerging] trait V2ScanMergingTestHelper { + + protected def v2Scans(df: DataFrame): Seq[DataSourceV2ScanRelation] = + df.queryExecution.optimizedPlan.collectWithSubqueries { + case s: DataSourceV2ScanRelation => s + } + + /** + * A merged subquery is referenced once per original subquery, so the logical plan duplicates it + * (physical planning reuses it). Dedupe by canonical form: one distinct scan means the merge + * happened, two means it was declined. + */ + protected def distinctScans(df: DataFrame): Int = v2Scans(df).map(_.canonicalized).distinct.length +} From 7dd18334f987e5c5bd30519149d5356c0ec20395 Mon Sep 17 00:00:00 2001 From: yangjie01 Date: Fri, 28 Aug 2026 23:02:23 +0800 Subject: [PATCH 4/7] Declare SCAN_MERGING per format and leave CSV and JSON out CSV and JSON break the contract SCAN_MERGING states. Their parsers are handed the columns the scan asked for and decide from that set what counts as a malformed record, so a merged scan reading the union does not read a superset of the rows either input read. Measured, on V1 and V2 alike, csv and json alike: with mode=DROPMALFORMED and a record malformed only in the other subquery's column, sum(a) is 8 where two separate scans give 10; with PERMISSIVE, the default, and _corrupt_record in the schema, the column is populated for a row the narrow scan counted as clean; with FAILFAST and a CSV row carrying fewer tokens than the schema has columns, the merged scan throws where the unmerged one returned rows. So FileTable no longer declares the capability for every subclass. A supportsScanMerging seam picks between two capability sets, and ParquetTable, OrcTable, TextTable and AvroTable override it. It defaults to false because the two directions fail differently: a format that does not merge misses an optimization, while a format that merges when its parser is projection-sensitive returns wrong rows. The criterion for the four that do declare is that reading more columns can only surface an error, never silently change which rows come back. A corrupt column chunk or datetimeRebaseModeInRead=EXCEPTION can make any format throw on a column the narrow scan pruned, so "can throw" would empty the list; what separates CSV and JSON is the silent, unrecoverable change to row membership and row content. The migration-guide entry is gone with the behaviour change it described. Tests: the capability test now covers both sides; the projection-only merge test narrows to parquet and orc; a new test pins that CSV and JSON decline the same shape; and the parse-mode test asserts the three measured shapes above on both read paths, with subquery counts pinning that V1 merged and V2 declined, and CSV column pruning pinned rather than assumed. mergedCounts and the parse-mode helper now assert which read path they ran on, which the suite scaladoc already claimed of every test. --- .../apache/spark/sql/v2/avro/AvroTable.scala | 5 + .../org/apache/spark/sql/avro/AvroSuite.scala | 2 +- docs/sql-migration-guide.md | 1 - docs/sql-performance-tuning.md | 2 +- .../execution/datasources/v2/FileTable.scala | 30 ++- .../datasources/v2/orc/OrcTable.scala | 4 + .../datasources/v2/parquet/ParquetTable.scala | 4 + .../datasources/v2/text/TextTable.scala | 4 + .../FileSourceV2PlanMergingSuite.scala | 231 ++++++++++++------ .../planmerging/V2ScanMergingTestHelper.scala | 5 +- 10 files changed, 201 insertions(+), 87 deletions(-) diff --git a/connector/avro/src/main/scala/org/apache/spark/sql/v2/avro/AvroTable.scala b/connector/avro/src/main/scala/org/apache/spark/sql/v2/avro/AvroTable.scala index abcea9a2a238e..dc71c98b98ca6 100644 --- a/connector/avro/src/main/scala/org/apache/spark/sql/v2/avro/AvroTable.scala +++ b/connector/avro/src/main/scala/org/apache/spark/sql/v2/avro/AvroTable.scala @@ -52,4 +52,9 @@ case class AvroTable( override def supportsDataType(dataType: DataType): Boolean = AvroUtils.supportsDataType(dataType) override def formatName: String = "Avro" + + // Every record is decoded against the full schema before the projection is applied to the decoded + // record, so reading more columns can only surface an error, never silently change which rows + // come back. The `mode` option in AvroOptions is read by the from_avro expression, not this scan. + override def supportsScanMerging: Boolean = true } diff --git a/connector/avro/src/test/scala/org/apache/spark/sql/avro/AvroSuite.scala b/connector/avro/src/test/scala/org/apache/spark/sql/avro/AvroSuite.scala index 26bd271bb83f9..0a37d051fb7e1 100644 --- a/connector/avro/src/test/scala/org/apache/spark/sql/avro/AvroSuite.scala +++ b/connector/avro/src/test/scala/org/apache/spark/sql/avro/AvroSuite.scala @@ -3938,7 +3938,7 @@ class AvroV2Suite extends AvroSuite with ExplainSuiteHelper { s"V2 formatName '${v2Table.formatName}' != V1 toString '${v1Format.toString}'") } - test("SPARK-57205: Avro V2 inherits SCAN_MERGING and merges scans differing only in columns") { + test("SPARK-57205: Avro V2 declares SCAN_MERGING and merges scans differing only in columns") { val v2Provider = DataSource.lookupDataSourceV2("avro", spark.sessionState.conf) assert(v2Provider.isDefined) val v2Table = v2Provider.get.asInstanceOf[FileDataSourceV2].getTable( diff --git a/docs/sql-migration-guide.md b/docs/sql-migration-guide.md index d953fa826ef3b..40c9784b986f6 100644 --- a/docs/sql-migration-guide.md +++ b/docs/sql-migration-guide.md @@ -25,7 +25,6 @@ license: | ## Upgrading from Spark SQL 4.3 to 4.4 - Since Spark 4.4, for storage-partitioned joins, `spark.sql.requireAllClusterKeysForCoPartition` requires every join key to be covered by some partition key instead of matching the partition keys positionally. As a result, a join-key column partitioned by more than one transform no longer prevents shuffle elimination, and `spark.sql.sources.v2.bucketing.allowKeysSubsetOfPartitionKeys.enabled` no longer additionally requires `spark.sql.requireAllClusterKeysForCoPartition` to be `false` when the join keys are a subset of the partition keys. As before, when the partition keys cover only part of the join keys, eliminating the shuffle still requires `spark.sql.requireAllClusterKeysForCoPartition` to be `false`. -- Since Spark 4.4, the built-in file formats declare the `SCAN_MERGING` table capability on their DataSource V2 read path, so two scans of the same file table that differ only in their projected columns can be merged into a single scan reading the union of those columns. A format takes that path only when it is removed from `spark.sql.sources.useV1SourceList`, and for these shapes merging there now matches what the V1 path already did. For CSV and JSON this also changes which records count as malformed, because the parser is handed only the columns the scan reads: with `mode` set to `DROPMALFORMED`, a record malformed only in the columns the other scan reads is now dropped for both. `PERMISSIVE` keeps the fields it did parse and `FAILFAST` rejects such a record either way, so neither of those modes changes. To restore the previous behavior, add the format back to `spark.sql.sources.useV1SourceList`, or disable subplan merging entirely by adding `org.apache.spark.sql.execution.planmerging.MergeSubplans` to `spark.sql.optimizer.excludedRules`. ## Upgrading from Spark SQL 4.2 to 4.3 diff --git a/docs/sql-performance-tuning.md b/docs/sql-performance-tuning.md index bbb44d3305e89..8933897c52fef 100644 --- a/docs/sql-performance-tuning.md +++ b/docs/sql-performance-tuning.md @@ -384,7 +384,7 @@ In TPC-DS benchmark runs, enabling symmetric filter propagation made `q9` and `q spark.sql.optimizer.mergeSubplans.filterPropagation.dsv2SymmetricFilterPropagation.enabled false - When true, two DataSource V2 scans that pushed the same strictly enforced filters but carry different best-effort (post-scan) filters can be merged even when spark.sql.optimizer.mergeSubplans.filterPropagation.symmetricFilterPropagation.enabled is false. In this case widening cannot change the set of rows the scan is required to return, as the strict filters are re-pushed unchanged and the enclosing Filter re-checks the rest above the scan. This applies only to V2 sources that opt in to scan merging with the SCAN_MERGING table capability. The built-in file formats opt in on their V2 read path, which a format reaches only when it is removed from spark.sql.sources.useV1SourceList; there the strictly enforced filters are the partition filters, so this configuration lets two scans over the same partitions but with different data filters merge. + When true, two DataSource V2 scans that pushed the same strictly enforced filters but carry different best-effort (post-scan) filters can be merged even when spark.sql.optimizer.mergeSubplans.filterPropagation.symmetricFilterPropagation.enabled is false. In this case widening cannot change the set of rows the scan is required to return, as the strict filters are re-pushed unchanged and the enclosing Filter re-checks the rest above the scan. This applies only to V2 sources that opt in to scan merging with the SCAN_MERGING table capability. Among the built-in file formats, Parquet, ORC, text and Avro opt in on their V2 read path, which a format reaches only when it is removed from spark.sql.sources.useV1SourceList; there the strictly enforced filters are the partition filters, so this configuration lets two scans over the same partitions but with different data filters merge. 4.3.0 diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/FileTable.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/FileTable.scala index b8c8c344abe10..04bfd32b24034 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/FileTable.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/FileTable.scala @@ -40,10 +40,11 @@ import org.apache.spark.util.ArrayImplicits._ /** * A [[Table]] backed by files. * - * Subclasses inherit the `SCAN_MERGING` capability, which holds them to this: with the scan options - * held constant, what a scan reads is determined by the filters pushed and the columns pruned on - * its builder. A subclass whose `newScanBuilder` does not meet that has to override - * [[capabilities]] to drop `SCAN_MERGING`, or Spark may fuse two of its scans into one. + * A subclass opts in to the `SCAN_MERGING` capability by overriding [[supportsScanMerging]], which + * holds it to this: with the scan options held constant, the rows and columns a scan reads are + * determined by the filters pushed and the columns pruned on its builder. A format whose parser + * decides what counts as a malformed record from the set of columns it was asked for does not meet + * that, because widening the column set can drop or rewrite rows that the narrower scan returned. */ abstract class FileTable( sparkSession: SparkSession, @@ -119,7 +120,15 @@ abstract class FileTable( override def properties: util.Map[String, String] = options.asCaseSensitiveMap - override def capabilities: java.util.Set[TableCapability] = FileTable.CAPABILITIES + override def capabilities: java.util.Set[TableCapability] = + if (supportsScanMerging) FileTable.CAPABILITIES_WITH_SCAN_MERGING else FileTable.CAPABILITIES + + /** + * Whether this table meets the `SCAN_MERGING` contract described on this class. Defaults to + * false: a format that does not merge only misses an optimization, while a format that merges + * when its parser is projection-sensitive returns wrong rows. + */ + protected def supportsScanMerging: Boolean = false /** * When possible, this method should return the schema of the given `files`. When the format @@ -189,8 +198,11 @@ abstract class FileTable( } object FileTable { - // The built-in file tables meet the SCAN_MERGING contract documented on FileTable: `fileIndex` is - // a lazy val, so every scan built from one table lists the same files, and `newScanBuilder` - // returns a fresh builder over `mergedOptions(options)`. - private val CAPABILITIES = util.EnumSet.of(BATCH_READ, BATCH_WRITE, SCAN_MERGING) + private val CAPABILITIES = util.EnumSet.of(BATCH_READ, BATCH_WRITE) + + // For the formats that override supportsScanMerging. `fileIndex` is a lazy val, so every scan + // built from one table lists the same files, and `newScanBuilder` returns a fresh builder over + // `mergedOptions(options)`. + private val CAPABILITIES_WITH_SCAN_MERGING = + util.EnumSet.of(BATCH_READ, BATCH_WRITE, SCAN_MERGING) } diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/orc/OrcTable.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/orc/OrcTable.scala index 08cd89fdacc61..d2c1f9a7d6ed5 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/orc/OrcTable.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/orc/OrcTable.scala @@ -68,4 +68,8 @@ case class OrcTable( } override def formatName: String = "ORC" + + // A row is decoded from the columns the scan asked for, so reading more columns can only surface + // an error, never silently change which rows come back. + override def supportsScanMerging: Boolean = true } diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/parquet/ParquetTable.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/parquet/ParquetTable.scala index 67052c201a9df..b13edacc61a0c 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/parquet/ParquetTable.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/parquet/ParquetTable.scala @@ -70,4 +70,8 @@ case class ParquetTable( } override def formatName: String = "Parquet" + + // A row is decoded from the column chunks the scan asked for, so reading more columns can only + // surface an error, never silently change which rows come back. + override def supportsScanMerging: Boolean = true } diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/text/TextTable.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/text/TextTable.scala index d8880b84c6211..17c9315d9d251 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/text/TextTable.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/text/TextTable.scala @@ -49,4 +49,8 @@ case class TextTable( override def supportsDataType(dataType: DataType): Boolean = dataType == StringType override def formatName: String = "Text" + + // The schema is a single `value` column, and every line -- or every file under `wholetext` -- + // becomes a row, so there is no parse step whose outcome a wider column set could change. + override def supportsScanMerging: Boolean = true } diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/planmerging/FileSourceV2PlanMergingSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/planmerging/FileSourceV2PlanMergingSuite.scala index 9d7352aa7780b..97c57a6db3786 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/execution/planmerging/FileSourceV2PlanMergingSuite.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/planmerging/FileSourceV2PlanMergingSuite.scala @@ -30,21 +30,27 @@ import org.apache.spark.sql.test.SharedSparkSession /** * Scan merging for the built-in file sources on their DSv2 read path (SPARK-57205). * - * [[FileTable]] declares the `SCAN_MERGING` table capability, so [[PlanMerger]] may fuse two file - * scans of the same table that differ only in their projected columns and/or pushed filters. For a - * file source the strictly enforced filters are the partition filters and the best-effort ones are - * the data filters. + * Parquet, ORC, text and Avro override [[FileTable.supportsScanMerging]], so [[PlanMerger]] may + * fuse two of their scans of the same table that differ only in their projected columns and/or + * pushed filters. For a file source the strictly enforced filters are the partition filters and the + * best-effort ones are the data filters. CSV and JSON do not override it, because their parsers are + * handed the columns the scan asked for and decide from that set what counts as a malformed record. * * SQL-on-file and catalog tables resolve to the V1 `FileFormat` regardless of - * `spark.sql.sources.useV1SourceList`, so every test goes through `DataFrameReader` and asserts the - * plan is V2 before asserting anything about merging. + * `spark.sql.sources.useV1SourceList`, so every test goes through `DataFrameReader` and asserts + * which read path the plan took before asserting anything about merging. */ class FileSourceV2PlanMergingSuite extends QueryTest with SharedSparkSession with AdaptiveSparkPlanHelper with V2ScanMergingTestHelper { import testImplicits._ - // The multi-column formats in sql/core; text is single-column and Avro lives in connector/avro. - private val multiColumnFormats = Seq("parquet", "orc", "json", "csv") + // The formats in sql/core that declare SCAN_MERGING and have more than one column. Avro also + // declares it but lives in connector/avro; text declares it but has only `value`. + private val mergingFormats = Seq("parquet", "orc") + + // These do not declare it: the parser is handed the columns the scan asked for, so widening the + // column set changes which records it treats as malformed. + private val projectionSensitiveFormats = Seq("csv", "json") private val flatSchema = "a long, b long, c long, d long" @@ -90,6 +96,13 @@ class FileSourceV2PlanMergingSuite extends QueryTest with SharedSparkSession } } + private def assertUsesFileSourceV1(df: DataFrame): Unit = { + val plan = df.queryExecution.optimizedPlan + assert(plan.collectWithSubqueries { case r: LogicalRelation => r }.nonEmpty, + s"expected the V1 file source path, but the plan has no V1 relation:\n$plan") + assert(v2Scans(df).isEmpty, s"expected no DSv2 file scan on the V1 path:\n$plan") + } + /** `(SubqueryExec, ReusedSubqueryExec)` counts, the same measure `PlanMergingSuite` uses. */ private def subqueryCounts(df: DataFrame): (Int, Int) = { val plan = df.queryExecution.executedPlan @@ -115,34 +128,38 @@ class FileSourceV2PlanMergingSuite extends QueryTest with SharedSparkSession SQLConf.MERGE_SUBPLANS_DSV2_SYMMETRIC_FILTER_PROPAGATION_ENABLED.key -> "true") { val df = sql(query) checkAnswer(df, expected) + if (useV1) assertUsesFileSourceV1(df) else assertUsesFileSourceV2(df) subqueryCounts(df) } } } - test("SPARK-57205: every built-in file table declares SCAN_MERGING") { - Seq("parquet", "orc", "json", "csv", "text").foreach { format => - withClue(s"format=$format: ") { - withTempPath { dir => - val path = dir.getCanonicalPath - spark.range(0, 5).selectExpr("cast(id AS string) AS value") - .write.format(format).save(path) - withSQLConf(SQLConf.USE_V1_SOURCE_LIST.key -> "") { - val relations = spark.read.format(format).load(path) - .queryExecution.analyzed.collect { case r: DataSourceV2Relation => r } - assert(relations.size == 1, s"expected a single DSv2 relation, got $relations") - val table = relations.head.table - assert(table.isInstanceOf[FileTable], s"expected a FileTable, got $table") - assert(table.capabilities().contains(TableCapability.SCAN_MERGING), - s"${table.getClass.getSimpleName} should declare SCAN_MERGING") + test("SPARK-57205: which built-in file tables declare SCAN_MERGING") { + // Avro is covered in AvroV2Suite, the module that has AvroTable on the classpath. + Seq("parquet" -> true, "orc" -> true, "text" -> true, "csv" -> false, "json" -> false) + .foreach { case (format, declares) => + withClue(s"format=$format: ") { + withTempPath { dir => + val path = dir.getCanonicalPath + spark.range(0, 5).selectExpr("cast(id AS string) AS value") + .write.format(format).save(path) + withSQLConf(SQLConf.USE_V1_SOURCE_LIST.key -> "") { + val relations = spark.read.format(format).load(path) + .queryExecution.analyzed.collect { case r: DataSourceV2Relation => r } + assert(relations.size == 1, s"expected a single DSv2 relation, got $relations") + val table = relations.head.table + assert(table.isInstanceOf[FileTable], s"expected a FileTable, got $table") + assert(table.capabilities().contains(TableCapability.SCAN_MERGING) == declares, + s"${table.getClass.getSimpleName}.capabilities() should " + + s"${if (declares) "declare" else "not declare"} SCAN_MERGING") + } } } } - } } test("SPARK-57205: merge two file scans that differ only in their projected columns") { - multiColumnFormats.foreach { format => + mergingFormats.foreach { format => withClue(s"format=$format: ") { withTempPath { dir => val path = dir.getCanonicalPath @@ -162,9 +179,39 @@ class FileSourceV2PlanMergingSuite extends QueryTest with SharedSparkSession s"the two scans should be fused into one:\n${df.queryExecution.optimizedPlan}") // Both sides carry the same data filter, so no widening is needed and this merges // under the default configuration. c is read because the filter stays above the scan. - assert(v2Scans(df).head.output.map(_.name).toSet == Set("a", "b", "c"), - s"the merged scan should read the union of both columns; " + - s"got ${v2Scans(df).head.output}") + val mergedOutput = v2Scans(df).head.output + assert(mergedOutput.map(_.name).toSet == Set("a", "b", "c"), + s"the merged scan should read the union of both columns; got $mergedOutput") + } + } + } + } + } + + test("SPARK-57205: do not merge CSV or JSON scans that differ in their projected columns") { + projectionSensitiveFormats.foreach { format => + withClue(s"format=$format: ") { + withTempPath { dir => + val path = dir.getCanonicalPath + writeFlat(format, path) + withFileView(format, path, schema = Some(flatSchema)) { + val df = sql( + """ + |SELECT + | (SELECT sum(a) FROM t WHERE c = 1), + | (SELECT sum(b) FROM t WHERE c = 1) + |""".stripMargin) + + checkAnswer(df, Row(70, 140)) + assertUsesFileSourceV2(df) + // Same shape as the test above, which merges for parquet and orc. These two decline + // because neither table declares SCAN_MERGING, which is what keeps the union of the + // columns out of the parser. Both measures are meaningful here: the two scans read + // different columns, so they do not canonicalize equal either. + assert(distinctScans(df) == 2, + s"the two scans should stay separate:\n${df.queryExecution.optimizedPlan}") + assert(subqueryCounts(df) == ((2, 0)), + s"unexpected subquery counts:\n${df.queryExecution.executedPlan}") } } } @@ -217,8 +264,9 @@ class FileSourceV2PlanMergingSuite extends QueryTest with SharedSparkSession assertUsesFileSourceV2(df) assert(distinctScans(df) == 1, s"the three scans should be fused into one:\n${df.queryExecution.optimizedPlan}") - assert(v2Scans(df).head.output.map(_.name).toSet == Set("a", "b", "c", "d"), - s"the merged scan should read the union of all three; got ${v2Scans(df).head.output}") + val mergedOutput = v2Scans(df).head.output + assert(mergedOutput.map(_.name).toSet == Set("a", "b", "c", "d"), + s"the merged scan should read the union of all three; got $mergedOutput") } } } @@ -435,59 +483,96 @@ class FileSourceV2PlanMergingSuite extends QueryTest with SharedSparkSession } } - test("SPARK-57205: merging widens the columns a text-based scan parses, as it does on V1") { - // A record malformed only in the column the other scan reads. The parsers are handed just the - // requested columns (spark.sql.csv.parser.columnPruning.enabled for CSV), so which columns a - // scan reads decides which malformed values it notices. - val cases = Seq( - "csv" -> Seq("0,0", "1,10", "2,BAD", "3,30", "4,40"), - "json" -> Seq( - """{"a":0,"b":0}""", - """{"a":1,"b":10}""", - """{"a":2,"b":"BAD"}""", - """{"a":3,"b":30}""", - """{"a":4,"b":40}""")) - val query = + test("SPARK-57205: CSV and JSON decline to merge, so their parsing stays per subquery") { + // The parsers are handed just the columns the scan asked for (for CSV under + // spark.sql.csv.parser.columnPruning.enabled), so which columns a scan reads decides which + // records it treats as malformed. V1 merges these shapes and parses the union; V2 declines, + // because neither table declares SCAN_MERGING. Each shape below is a case where that decision + // is visible in the result, so adding the capability back to either table fails this test. + val typeErrorCsv = Seq("0,0", "1,10", "2,BAD", "3,30", "4,40") + val typeErrorJson = Seq( + """{"a":0,"b":0}""", + """{"a":1,"b":10}""", + """{"a":2,"b":"BAD"}""", + """{"a":3,"b":30}""", + """{"a":4,"b":40}""") + // One token where the schema has two columns. Neither narrow scan is malformed: with column + // pruning the parsed schema is the projection, so a one-column scan matches a one-token row. + val shortRowCsv = Seq("0,0", "1,10", "2", "3,30", "4,40") + val sumQuery = """ |SELECT | (SELECT sum(a) FROM t), | (SELECT sum(b) FROM t) |""".stripMargin + val corruptQuery = + """ + |SELECT + | (SELECT count(_corrupt_record) FROM t WHERE a >= 0), + | (SELECT sum(b) FROM t WHERE a >= 0) + |""".stripMargin - cases.foreach { case (format, lines) => - withClue(s"format=$format: ") { - withTempPath { dir => - val path = dir.getCanonicalPath - lines.toDS().write.text(path) + def withData(lines: Seq[String])(f: String => Unit): Unit = + withTempPath { dir => + val path = dir.getCanonicalPath + lines.toDS().write.text(path) + f(path) + } - def rows(mode: String, useV1: Boolean): Seq[Row] = - withFileView(format, path, useV1 = useV1, schema = Some("a long, b long"), - options = Map("mode" -> mode)) { - sql(query).collect().toSeq - } + def rows( + format: String, + path: String, + schema: String, + mode: String, + query: String, + useV1: Boolean): Seq[Row] = + // Pin CSV column pruning rather than rely on its default: with it off the parser is handed + // the full data schema and every expectation below changes. + withSQLConf(SQLConf.CSV_PARSER_COLUMN_PRUNING.key -> "true") { + withFileView(format, path, useV1 = useV1, schema = Some(schema), + options = Map("mode" -> mode, "columnNameOfCorruptRecord" -> "_corrupt_record")) { + val df = sql(query) + if (useV1) assertUsesFileSourceV1(df) else assertUsesFileSourceV2(df) + val result = df.collect().toSeq + // V1 merges the two subqueries into one; V2 declines. Asserted after collect() so that + // AQE has finalized and the reuse of the merged subquery is visible in the plan. Pinning + // this alongside the rows attributes the difference to the merge decision itself. + assert(subqueryCounts(df) == (if (useV1) ((1, 1)) else ((2, 0))), + s"unexpected subquery counts on ${if (useV1) "V1" else "V2"}:\n" + + df.queryExecution.executedPlan) + result + } + } - // Merging makes one scan parse both a and b, so DROPMALFORMED drops the record malformed - // in b for both aggregates and sum(a) is 8, not the 10 two separate scans produce. The - // merged scan therefore reads fewer rows than the a-only scan did, which SCAN_MERGING's - // "superset of their rows" premise does not allow. V1 already read the union after - // merging, so the V2 path now matches it. - val droppedV2 = rows("DROPMALFORMED", useV1 = false) - assert(droppedV2 == Seq(Row(8, 80))) - assert(rows("DROPMALFORMED", useV1 = true) == droppedV2) - - // PERMISSIVE keeps the fields it did parse, so widening the parsed columns leaves both - // aggregates on all five records. - val permissiveV2 = rows("PERMISSIVE", useV1 = false) - assert(permissiveV2 == Seq(Row(10, 80))) - assert(rows("PERMISSIVE", useV1 = true) == permissiveV2) - - // FAILFAST throws whether or not the scans merged: the merged scan reads the union of the - // two column sets, so the scan that reads b already throws on its own. - Seq(true, false).foreach { useV1 => - intercept[SparkException](rows("FAILFAST", useV1 = useV1)) - } + Seq("csv" -> typeErrorCsv, "json" -> typeErrorJson).foreach { case (format, lines) => + withClue(s"format=$format: ") { + withData(lines) { path => + // DROPMALFORMED. The a-only scan never parses b, so V2 keeps the record for sum(a). V1's + // merged scan parses the union and drops it for both, giving 8. + assert(rows(format, path, "a long, b long", "DROPMALFORMED", sumQuery, + useV1 = false) == Seq(Row(10, 80))) + assert(rows(format, path, "a long, b long", "DROPMALFORMED", sumQuery, + useV1 = true) == Seq(Row(8, 80))) + + // PERMISSIVE, the default mode, with the corrupt-record column in the schema. V2 does not + // flag the record for the subquery that reads a and the corrupt column; V1's merged scan + // parses b, so the column is populated for a row the first subquery counted as clean. + assert(rows(format, path, "a long, b long, _corrupt_record string", "PERMISSIVE", + corruptQuery, useV1 = false) == Seq(Row(0, 80))) + assert(rows(format, path, "a long, b long, _corrupt_record string", "PERMISSIVE", + corruptQuery, useV1 = true) == Seq(Row(1, 80))) } } } + + // FAILFAST, CSV only: JSON has no arity check, so a missing field is null, not malformed. V2 + // returns rows; V1's merged scan parses two columns against a one-token row and throws, which + // is a working query turning into an error. + withData(shortRowCsv) { path => + assert(rows("csv", path, "a long, b long", "FAILFAST", sumQuery, + useV1 = false) == Seq(Row(10, 80))) + intercept[SparkException]( + rows("csv", path, "a long, b long", "FAILFAST", sumQuery, useV1 = true)) + } } } diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/planmerging/V2ScanMergingTestHelper.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/planmerging/V2ScanMergingTestHelper.scala index a6b1154ae45bd..91ef2a36d4f2f 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/execution/planmerging/V2ScanMergingTestHelper.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/planmerging/V2ScanMergingTestHelper.scala @@ -33,8 +33,9 @@ private[planmerging] trait V2ScanMergingTestHelper { /** * A merged subquery is referenced once per original subquery, so the logical plan duplicates it - * (physical planning reuses it). Dedupe by canonical form: one distinct scan means the merge - * happened, two means it was declined. + * (physical planning reuses it). Dedupe by canonical form: one distinct scan is consistent with a + * merge and two means it was declined. Two scans that read the same columns canonicalize equal + * either way, so use subquery counts instead when the two column sets match. */ protected def distinctScans(df: DataFrame): Int = v2Scans(df).map(_.canonicalized).distinct.length } From 60961fe4e10ae0de75301da85cbd7ec3659ccd95 Mon Sep 17 00:00:00 2001 From: yangjie01 Date: Sat, 29 Aug 2026 06:55:34 +0800 Subject: [PATCH 5/7] Withhold SCAN_MERGING when the read is not strict, and gate Avro on positional matching Under spark.sql.files.ignoreCorruptFiles a read failure in a column that only the sibling subquery projects is swallowed, and the rest of that file's rows go with it, so the merged scan returns fewer rows than the narrow one did. Measured on parquet, V1 and V2 alike: with b written as a string and read as a long, SELECT (SELECT sum(a) FROM t), (SELECT count(b) FROM t) returns [45, 0] with MergeSubplans excluded and [null, 0] with it on. That is the same silent change to row membership that keeps CSV and JSON out, reached by a format that declares the capability, so FileTable now ANDs a hasStrictFileReads gate into capabilities, matching FileScanRDD.hasStrictFileReads on the physical side. The gate is evaluated per call, not cached, so a table built before the configuration was set still answers for the read that is running. AvroTable withholds the capability under positionalFieldMatching. AvroPartitionReaderFactory builds the deserializer from the pruned read schema while the Avro side stays the full Avro schema, so under that option catalyst field i of the projection takes Avro field i of that schema and widening the projection changes the values a column comes back with. The option is read off the options map rather than through AvroOptions, whose constructor resolves avroSchemaUrl and would do I/O on every capabilities() call, and read leniently so a malformed value still fails where Avro reports it. The four overrides are protected, matching the seam. The class scaladoc's contract is restated as invariance under widening rather than determinism. The old wording, "the rows and values a scan reads are determined by the filters pushed and the columns pruned", licenses dependence on the pruned column set, which is exactly what merging must not have: a DROPMALFORMED CSV scan's rows are a deterministic function of (filters, columns) and so satisfied the old clause literally while the next sentence asserted it did not. Tests: a new test pins that a non-strict read withholds the capability, on both strictness configurations and with the table and view built outside the configuration scope so a cached gate would fail it; the Avro test pins that positionalFieldMatching withholds it; a text merge test covers the one shape a single-column table can differ in, with a note that both aggregates have to be hash-aggregatable or PlanMerger declines above the scans; the parity helper and the parse-mode helper assert which read path they ran on; the merged scan is checked to still carry the partition filter and the OR-widened data filter; the different-tables test writes different rows to the second table so a cross-table merge would change the answer and not just the plan; and the confs the merge decision depends on are pinned in the suite rather than inherited. docs/sql-performance-tuning.md records both withholding rules. --- .../apache/spark/sql/v2/avro/AvroTable.scala | 18 +- .../org/apache/spark/sql/avro/AvroSuite.scala | 62 +++--- docs/sql-performance-tuning.md | 2 +- .../execution/datasources/v2/FileTable.scala | 30 ++- .../datasources/v2/orc/OrcTable.scala | 8 +- .../datasources/v2/parquet/ParquetTable.scala | 8 +- .../datasources/v2/text/TextTable.scala | 2 +- .../FileSourceV2PlanMergingSuite.scala | 191 +++++++++++++++--- .../planmerging/V2ScanMergingTestHelper.scala | 4 +- 9 files changed, 247 insertions(+), 78 deletions(-) diff --git a/connector/avro/src/main/scala/org/apache/spark/sql/v2/avro/AvroTable.scala b/connector/avro/src/main/scala/org/apache/spark/sql/v2/avro/AvroTable.scala index dc71c98b98ca6..7b6db3c16c5b7 100644 --- a/connector/avro/src/main/scala/org/apache/spark/sql/v2/avro/AvroTable.scala +++ b/connector/avro/src/main/scala/org/apache/spark/sql/v2/avro/AvroTable.scala @@ -21,7 +21,7 @@ import scala.jdk.CollectionConverters._ import org.apache.hadoop.fs.FileStatus import org.apache.spark.sql.SparkSession -import org.apache.spark.sql.avro.AvroUtils +import org.apache.spark.sql.avro.{AvroOptions, AvroUtils} import org.apache.spark.sql.connector.write.{LogicalWriteInfo, Write, WriteBuilder} import org.apache.spark.sql.execution.datasources.FileFormat import org.apache.spark.sql.execution.datasources.v2.FileTable @@ -53,8 +53,16 @@ case class AvroTable( override def formatName: String = "Avro" - // Every record is decoded against the full schema before the projection is applied to the decoded - // record, so reading more columns can only surface an error, never silently change which rows - // come back. The `mode` option in AvroOptions is read by the from_avro expression, not this scan. - override def supportsScanMerging: Boolean = true + // Avro has no record-level parse verdict: a record is either decodable or the read fails, and + // there is no mode that drops or rewrites a record based on the columns asked for. The `mode` + // option in AvroOptions is read by from_avro and schema_of_avro, not by this scan. + // + // `positionalFieldMatching` is the exception. AvroPartitionReaderFactory builds the deserializer + // from the pruned read schema while the Avro side stays the full Avro schema, so under that + // option catalyst field i of the projection takes Avro field i of that schema, and widening the + // projection changes the values a column comes back with. Read the option off the map rather than + // through AvroOptions, whose constructor resolves `avroSchemaUrl` and would do I/O here, and read + // it leniently so a malformed value still fails where Avro reports it rather than here. + override protected def supportsScanMerging: Boolean = + !"true".equalsIgnoreCase(options.get(AvroOptions.POSITIONAL_FIELD_MATCHING)) } diff --git a/connector/avro/src/test/scala/org/apache/spark/sql/avro/AvroSuite.scala b/connector/avro/src/test/scala/org/apache/spark/sql/avro/AvroSuite.scala index 0a37d051fb7e1..f4e27dd7f0381 100644 --- a/connector/avro/src/test/scala/org/apache/spark/sql/avro/AvroSuite.scala +++ b/connector/avro/src/test/scala/org/apache/spark/sql/avro/AvroSuite.scala @@ -3939,33 +3939,45 @@ class AvroV2Suite extends AvroSuite with ExplainSuiteHelper { } test("SPARK-57205: Avro V2 declares SCAN_MERGING and merges scans differing only in columns") { - val v2Provider = DataSource.lookupDataSourceV2("avro", spark.sessionState.conf) - assert(v2Provider.isDefined) - val v2Table = v2Provider.get.asInstanceOf[FileDataSourceV2].getTable( - new StructType(), Array.empty, JCollections.emptyMap[String, String]()) - assert(v2Table.capabilities().contains(TableCapability.SCAN_MERGING)) + // AvroTable withholds the capability under positionalFieldMatching, because the deserializer is + // built from the pruned read schema while the Avro side stays unpruned, so catalyst field i of + // the projection takes Avro field i of that schema and widening the projection shifts values. + // FileTable also withholds it when the reads are not strict, so pin that rather than inherit. + withSQLConf( + SQLConf.IGNORE_CORRUPT_FILES.key -> "false", + SQLConf.IGNORE_MISSING_FILES.key -> "false") { + val v2Provider = DataSource.lookupDataSourceV2("avro", spark.sessionState.conf) + assert(v2Provider.isDefined) + val dsV2 = v2Provider.get.asInstanceOf[FileDataSourceV2] + val v2Table = dsV2.getTable( + new StructType(), Array.empty, JCollections.emptyMap[String, String]()) + assert(v2Table.capabilities().contains(TableCapability.SCAN_MERGING)) + val positional = dsV2.getTable(new StructType(), Array.empty, + JCollections.singletonMap("positionalFieldMatching", "true")) + assert(!positional.capabilities().contains(TableCapability.SCAN_MERGING)) - withTempPath { dir => - val path = dir.getCanonicalPath - spark.range(0, 20).selectExpr("id AS a", "id * 2 AS b", "id % 3 AS c") - .write.format("avro").save(path) - withTempView("avro_scan_merging") { - spark.read.format("avro").load(path).createOrReplaceTempView("avro_scan_merging") - val df = sql( - """ - |SELECT - | (SELECT sum(a) FROM avro_scan_merging WHERE c = 1), - | (SELECT sum(b) FROM avro_scan_merging WHERE c = 1) - |""".stripMargin) - checkAnswer(df, Row(70, 140)) - val scans = df.queryExecution.optimizedPlan.collectWithSubqueries { - case s: DataSourceV2ScanRelation => s + withTempPath { dir => + val path = dir.getCanonicalPath + spark.range(0, 20).selectExpr("id AS a", "id * 2 AS b", "id % 3 AS c") + .write.format("avro").save(path) + withTempView("avro_scan_merging") { + spark.read.format("avro").load(path).createOrReplaceTempView("avro_scan_merging") + val df = sql( + """ + |SELECT + | (SELECT sum(a) FROM avro_scan_merging WHERE c = 1), + | (SELECT sum(b) FROM avro_scan_merging WHERE c = 1) + |""".stripMargin) + checkAnswer(df, Row(70, 140)) + val scans = df.queryExecution.optimizedPlan.collectWithSubqueries { + case s: DataSourceV2ScanRelation => s + } + assert(scans.map(_.canonicalized).distinct.length == 1, + s"the two Avro scans should be fused into one:\n${df.queryExecution.optimizedPlan}") + // c is read because the filter stays above the merged scan. + assert(scans.head.output.map(_.name).toSet == Set("a", "b", "c"), + s"the merged scan should read the union of both columns; got ${scans.head.output}") } - assert(scans.map(_.canonicalized).distinct.length == 1, - s"the two Avro scans should be fused into one:\n${df.queryExecution.optimizedPlan}") - // c is read because the filter stays above the merged scan. - assert(scans.head.output.map(_.name).toSet == Set("a", "b", "c"), - s"the merged scan should read the union of both columns; got ${scans.head.output}") } } } diff --git a/docs/sql-performance-tuning.md b/docs/sql-performance-tuning.md index 8933897c52fef..7553861f9f66f 100644 --- a/docs/sql-performance-tuning.md +++ b/docs/sql-performance-tuning.md @@ -384,7 +384,7 @@ In TPC-DS benchmark runs, enabling symmetric filter propagation made `q9` and `q spark.sql.optimizer.mergeSubplans.filterPropagation.dsv2SymmetricFilterPropagation.enabled false - When true, two DataSource V2 scans that pushed the same strictly enforced filters but carry different best-effort (post-scan) filters can be merged even when spark.sql.optimizer.mergeSubplans.filterPropagation.symmetricFilterPropagation.enabled is false. In this case widening cannot change the set of rows the scan is required to return, as the strict filters are re-pushed unchanged and the enclosing Filter re-checks the rest above the scan. This applies only to V2 sources that opt in to scan merging with the SCAN_MERGING table capability. Among the built-in file formats, Parquet, ORC, text and Avro opt in on their V2 read path, which a format reaches only when it is removed from spark.sql.sources.useV1SourceList; there the strictly enforced filters are the partition filters, so this configuration lets two scans over the same partitions but with different data filters merge. + When true, two DataSource V2 scans that pushed the same strictly enforced filters but carry different best-effort (post-scan) filters can be merged even when spark.sql.optimizer.mergeSubplans.filterPropagation.symmetricFilterPropagation.enabled is false. In this case widening cannot change the set of rows the scan is required to return, as the strict filters are re-pushed unchanged and the enclosing Filter re-checks the rest above the scan. This applies only to V2 sources that opt in to scan merging with the SCAN_MERGING table capability. Among the built-in file formats, Parquet, ORC, text and Avro opt in on their V2 read path, which a format reaches only when it is removed from spark.sql.sources.useV1SourceList; there the strictly enforced filters are the partition filters, so this configuration lets two scans over the same partitions but with different data filters merge. A file table withholds the capability when spark.sql.files.ignoreCorruptFiles or spark.sql.files.ignoreMissingFiles is true, because a read failure in a column that only the other scan projects would then be swallowed along with the rest of that file's rows, and Avro withholds it under positionalFieldMatching, which resolves a column by its position in the projection. 4.3.0 diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/FileTable.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/FileTable.scala index 04bfd32b24034..41b4a6196d5f1 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/FileTable.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/FileTable.scala @@ -41,10 +41,13 @@ import org.apache.spark.util.ArrayImplicits._ * A [[Table]] backed by files. * * A subclass opts in to the `SCAN_MERGING` capability by overriding [[supportsScanMerging]], which - * holds it to this: with the scan options held constant, the rows and columns a scan reads are - * determined by the filters pushed and the columns pruned on its builder. A format whose parser - * decides what counts as a malformed record from the set of columns it was asked for does not meet - * that, because widening the column set can drop or rewrite rows that the narrower scan returned. + * holds it to this: with the scan options and the pushed filters held constant, widening the set of + * columns pruned on its builder must not change which rows the scan returns, nor the values it + * returns for the columns it was already asked for. It may at most surface a read error. A format + * whose parser decides what counts as a malformed record from the set of columns it was asked for + * does not meet that, and neither does one that resolves a column by its position in the + * projection. The capability is also withheld from a table whose reads are not strict, see + * `hasStrictFileReads`. */ abstract class FileTable( sparkSession: SparkSession, @@ -121,7 +124,11 @@ abstract class FileTable( override def properties: util.Map[String, String] = options.asCaseSensitiveMap override def capabilities: java.util.Set[TableCapability] = - if (supportsScanMerging) FileTable.CAPABILITIES_WITH_SCAN_MERGING else FileTable.CAPABILITIES + if (supportsScanMerging && hasStrictFileReads) { + FileTable.CAPABILITIES_WITH_SCAN_MERGING + } else { + FileTable.CAPABILITIES + } /** * Whether this table meets the `SCAN_MERGING` contract described on this class. Defaults to @@ -130,6 +137,19 @@ abstract class FileTable( */ protected def supportsScanMerging: Boolean = false + /** + * Whether a read of this table is strict. Under `ignoreCorruptFiles`, a read failure in a column + * that only the other scan projects is swallowed and the remaining rows of that file are dropped, + * so the merged scan would not read a superset of either input's rows. `ignoreMissingFiles` drops + * the same rows whatever is projected, and is included to match `FileScanRDD.hasStrictFileReads`, + * the same predicate on the physical side. Evaluated per call rather than cached, so a table + * built before either configuration was set still answers for the read that is running. + */ + private def hasStrictFileReads: Boolean = { + val fileSourceOptions = new FileSourceOptions(options.asCaseSensitiveMap.asScala.toMap) + !fileSourceOptions.ignoreCorruptFiles && !fileSourceOptions.ignoreMissingFiles + } + /** * When possible, this method should return the schema of the given `files`. When the format * does not support inference, or no valid files are given should return None. In these cases diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/orc/OrcTable.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/orc/OrcTable.scala index d2c1f9a7d6ed5..b3aaf10ccc2bc 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/orc/OrcTable.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/orc/OrcTable.scala @@ -69,7 +69,9 @@ case class OrcTable( override def formatName: String = "ORC" - // A row is decoded from the columns the scan asked for, so reading more columns can only surface - // an error, never silently change which rows come back. - override def supportsScanMerging: Boolean = true + // A row is decoded from the columns the scan asked for, so under strict file reads reading more + // columns can only surface an error, never silently change which rows come back. When the read is + // not strict that error is swallowed and the rest of the file's rows go with it, which is why + // FileTable withholds the capability there. + override protected def supportsScanMerging: Boolean = true } diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/parquet/ParquetTable.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/parquet/ParquetTable.scala index b13edacc61a0c..768e1ce329f59 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/parquet/ParquetTable.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/parquet/ParquetTable.scala @@ -71,7 +71,9 @@ case class ParquetTable( override def formatName: String = "Parquet" - // A row is decoded from the column chunks the scan asked for, so reading more columns can only - // surface an error, never silently change which rows come back. - override def supportsScanMerging: Boolean = true + // A row is decoded from the column chunks the scan asked for, so under strict file reads reading + // more columns can only surface an error, never silently change which rows come back. When the + // read is not strict that error is swallowed and the rest of the file's rows go with it, which is + // why FileTable withholds the capability there. + override protected def supportsScanMerging: Boolean = true } diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/text/TextTable.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/text/TextTable.scala index 17c9315d9d251..440b2013ce1ac 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/text/TextTable.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/text/TextTable.scala @@ -52,5 +52,5 @@ case class TextTable( // The schema is a single `value` column, and every line -- or every file under `wholetext` -- // becomes a row, so there is no parse step whose outcome a wider column set could change. - override def supportsScanMerging: Boolean = true + override protected def supportsScanMerging: Boolean = true } diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/planmerging/FileSourceV2PlanMergingSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/planmerging/FileSourceV2PlanMergingSuite.scala index 97c57a6db3786..ba329d3e1afb7 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/execution/planmerging/FileSourceV2PlanMergingSuite.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/planmerging/FileSourceV2PlanMergingSuite.scala @@ -17,13 +17,14 @@ package org.apache.spark.sql.execution.planmerging -import org.apache.spark.SparkException +import org.apache.spark.{SparkConf, SparkException} import org.apache.spark.sql.{DataFrame, QueryTest, Row} import org.apache.spark.sql.connector.catalog.TableCapability import org.apache.spark.sql.execution.{ReusedSubqueryExec, SubqueryExec} import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanHelper import org.apache.spark.sql.execution.datasources.LogicalRelation -import org.apache.spark.sql.execution.datasources.v2.{DataSourceV2Relation, FileTable} +import org.apache.spark.sql.execution.datasources.v2.{DataSourceV2Relation, FileScan, FileTable} +import org.apache.spark.sql.execution.datasources.v2.parquet.ParquetScan import org.apache.spark.sql.internal.SQLConf import org.apache.spark.sql.test.SharedSparkSession @@ -31,10 +32,15 @@ import org.apache.spark.sql.test.SharedSparkSession * Scan merging for the built-in file sources on their DSv2 read path (SPARK-57205). * * Parquet, ORC, text and Avro override [[FileTable.supportsScanMerging]], so [[PlanMerger]] may - * fuse two of their scans of the same table that differ only in their projected columns and/or - * pushed filters. For a file source the strictly enforced filters are the partition filters and the - * best-effort ones are the data filters. CSV and JSON do not override it, because their parsers are - * handed the columns the scan asked for and decide from that set what counts as a malformed record. + * fuse two of their scans of the same table. Scans that differ only in their projected columns + * merge under the default configuration; scans whose data filters differ need one of the symmetric + * filter propagation configurations, and scans whose partition filters differ never merge, because + * for a file source the partition filters are the strictly enforced ones and the data filters are + * best-effort. The capability is also withheld from a table whose reads are not strict. + * + * CSV and JSON do not override it. Their parsers are handed the columns the scan asked for, at + * least while `spark.sql.csv.parser.columnPruning.enabled` is on for CSV, and decide from that set + * what counts as a malformed record. * * SQL-on-file and catalog tables resolve to the V1 `FileFormat` regardless of * `spark.sql.sources.useV1SourceList`, so every test goes through `DataFrameReader` and asserts @@ -44,6 +50,15 @@ class FileSourceV2PlanMergingSuite extends QueryTest with SharedSparkSession with AdaptiveSparkPlanHelper with V2ScanMergingTestHelper { import testImplicits._ + // Pin what the merge decision now depends on, and what the subquery-count measure depends on, so + // a changed default fails in one legible place rather than inverting every assertion below. Tests + // that vary one of these set it themselves. + override protected def sparkConf: SparkConf = super.sparkConf + .set(SQLConf.IGNORE_CORRUPT_FILES, false) + .set(SQLConf.IGNORE_MISSING_FILES, false) + .set(SQLConf.MERGE_SUBPLANS_FILTER_PROPAGATION_ENABLED, true) + .set(SQLConf.SUBQUERY_REUSE_ENABLED, true) + // The formats in sql/core that declare SCAN_MERGING and have more than one column. Avro also // declares it but lives in connector/avro; text declares it but has only `value`. private val mergingFormats = Seq("parquet", "orc") @@ -54,8 +69,8 @@ class FileSourceV2PlanMergingSuite extends QueryTest with SharedSparkSession private val flatSchema = "a long, b long, c long, d long" - private def writeFlat(format: String, path: String): Unit = - spark.range(0, 20) + private def writeFlat(format: String, path: String, start: Long = 0): Unit = + spark.range(start, start + 20) .selectExpr("id AS a", "id * 2 AS b", "id % 3 AS c", "id * 3 AS d") .write.format(format).save(path) @@ -65,9 +80,12 @@ class FileSourceV2PlanMergingSuite extends QueryTest with SharedSparkSession .write.partitionBy("p").format("parquet").save(path) /** - * Registers `path` as a temp view read through the V2 path, or the V1 path if `useV1`. Not named - * `withView`: that name is taken by a varargs helper in `QueryCleanupHelper`, which a call with - * only positional String arguments would silently bind to instead. + * Registers `path` as a temp view read through the V2 path, or the V1 path if `useV1`. The view + * is created inside the `USE_V1_SOURCE_LIST` scope on purpose: a temp view stores its analyzed + * plan, so which read path it takes is fixed when the view is created, not when it is queried. + * + * Not named `withView`: that name is taken by a varargs helper in `QueryCleanupHelper`, which a + * call with only positional String arguments would silently bind to instead. */ private def withFileView[T]( format: String, @@ -112,8 +130,9 @@ class FileSourceV2PlanMergingSuite extends QueryTest with SharedSparkSession } /** - * Runs `query` over the parquet data at `path` on the V1 or V2 read path with both symmetric - * filter propagation configurations on, checks the rows and returns the subquery counts. + * Runs `query` over the parquet data at `path` on the V1 or V2 read path, with AQE as given and + * both symmetric filter propagation configurations on, checks the rows and returns the subquery + * counts. */ private def mergedCounts( path: String, @@ -158,6 +177,47 @@ class FileSourceV2PlanMergingSuite extends QueryTest with SharedSparkSession } } + test("SPARK-57205: withhold SCAN_MERGING from a table whose reads are not strict") { + withTempPath { dir => + val path = dir.getCanonicalPath + // b is written as a string and read as a long, so the reader fails only once it reads b. + spark.range(0, 10).selectExpr("id AS a", "cast(id AS string) AS b").write.parquet(path) + // The table is built outside the strictness scope on purpose: the gate is evaluated per call, + // so it has to answer for the read that is running rather than for the read that built it. + withSQLConf(SQLConf.USE_V1_SOURCE_LIST.key -> "") { + val relations = spark.read.schema("a long, b long").parquet(path) + .queryExecution.analyzed.collect { case r: DataSourceV2Relation => r } + assert(relations.size == 1, s"expected a single DSv2 relation, got $relations") + val table = relations.head.table + assert(table.capabilities().contains(TableCapability.SCAN_MERGING), + "a strict read should declare SCAN_MERGING") + // Both halves of the strictness predicate, on the table that was built while both were off. + Seq(SQLConf.IGNORE_CORRUPT_FILES.key, SQLConf.IGNORE_MISSING_FILES.key).foreach { conf => + withClue(s"$conf=true: ") { + withSQLConf(conf -> "true") { + assert(!table.capabilities().contains(TableCapability.SCAN_MERGING), + s"$conf should withhold SCAN_MERGING") + } + } + } + } + // The scans stay separate, so the a-only scan never reads b and sum(a) is still exact. If + // they merged, reading b would fail, ignoreCorruptFiles would swallow it and drop the rest of + // the file, and sum(a) would come back null over rows nothing above the scan removed. The + // view is registered before the conf is set, so a cached gate would answer from the strict + // read. + withFileView("parquet", path, schema = Some("a long, b long")) { + withSQLConf(SQLConf.IGNORE_CORRUPT_FILES.key -> "true") { + val df = sql("SELECT (SELECT sum(a) FROM t), (SELECT count(b) FROM t)") + checkAnswer(df, Row(45, 0)) + assertUsesFileSourceV2(df) + assert(distinctScans(df) == 2, + s"the two scans should stay separate:\n${df.queryExecution.optimizedPlan}") + } + } + } + } + test("SPARK-57205: merge two file scans that differ only in their projected columns") { mergingFormats.foreach { format => withClue(s"format=$format: ") { @@ -188,6 +248,27 @@ class FileSourceV2PlanMergingSuite extends QueryTest with SharedSparkSession } } + test("SPARK-57205: merge two text scans that differ only in their projected columns") { + withTempPath { dir => + val path = dir.getCanonicalPath + Seq("a", "bb", "ccc").toDS().write.text(path) + withFileView("text", path) { + // A text table has the single column `value`, so the only projection difference reachable + // is an empty read set against `[value]`. Both aggregates have to be hash-aggregatable or + // PlanMerger's supportedAggregateMerge declines above the scans, before the capability is + // reached: max(value) over a string is neither hash nor object-hash, sum(length(value)) is. + val df = sql("SELECT (SELECT count(*) FROM t), (SELECT sum(length(value)) FROM t)") + checkAnswer(df, Row(3, 6)) + assertUsesFileSourceV2(df) + assert(distinctScans(df) == 1, + s"the two scans should be fused into one:\n${df.queryExecution.optimizedPlan}") + val mergedOutput = v2Scans(df).head.output + assert(mergedOutput.map(_.name) == Seq("value"), + s"the merged scan should read value; got $mergedOutput") + } + } + } + test("SPARK-57205: do not merge CSV or JSON scans that differ in their projected columns") { projectionSensitiveFormats.foreach { format => withClue(s"format=$format: ") { @@ -237,12 +318,17 @@ class FileSourceV2PlanMergingSuite extends QueryTest with SharedSparkSession s"the two scans should be fused into one:\n${df.queryExecution.optimizedPlan}") val scan = v2Scans(df).head // A partition filter is fully enforced by the scan and nothing above it re-checks, so p is - // not read; the rebuilt scan has to push the filter again or it would read all partitions. + // not read. assert(scan.output.map(_.name).toSet == Set("a", "b"), s"the merged scan should read the union of both columns; got ${scan.output}") - assert(scan.pushedFilters.exists(_.references.exists(_.name == "p")), - s"the partition filter should be re-pushed strict onto the merged scan; " + - s"got pushedFilters=${scan.pushedFilters.mkString("[", ", ", "]")}") + // The rebuilt scan has to carry the filter or it would read all four partitions. Read it + // off the FileScan rather than off `pushedFilters`, which each unmerged scan records too. + val partitionFilters = scan.scan match { + case f: FileScan => f.partitionFilters + case other => fail(s"expected a FileScan, got ${other.getClass.getSimpleName}") + } + assert(partitionFilters.exists(_.references.exists(_.name == "p")), + s"the merged scan should still enforce the partition filter; got $partitionFilters") } } } @@ -276,11 +362,15 @@ class FileSourceV2PlanMergingSuite extends QueryTest with SharedSparkSession withTempPath { dir => val path = dir.getCanonicalPath writeFlat("parquet", path) - Seq(true, false).foreach { dsv2Symmetric => - withClue(s"dsv2SymmetricFilterPropagation=$dsv2Symmetric: ") { - withFileView("parquet", path) { - withSQLConf(SQLConf.MERGE_SUBPLANS_DSV2_SYMMETRIC_FILTER_PROPAGATION_ENABLED.key -> - dsv2Symmetric.toString) { + withFileView("parquet", path) { + Seq(true, false).foreach { dsv2Symmetric => + withClue(s"dsv2SymmetricFilterPropagation=$dsv2Symmetric: ") { + // The generic symmetric propagation would enable this merge on its own, so pin it off: + // the point of the test is that the dsv2 configuration alone decides. + withSQLConf( + SQLConf.MERGE_SUBPLANS_SYMMETRIC_FILTER_PROPAGATION_ENABLED.key -> "false", + SQLConf.MERGE_SUBPLANS_DSV2_SYMMETRIC_FILTER_PROPAGATION_ENABLED.key -> + dsv2Symmetric.toString) { val df = sql( """ |SELECT @@ -296,6 +386,18 @@ class FileSourceV2PlanMergingSuite extends QueryTest with SharedSparkSession // merge. The enclosing Filter keeps each aggregate exact either way. assert(distinctScans(df) == (if (dsv2Symmetric) 1 else 2), s"unexpected scan count:\n${df.queryExecution.optimizedPlan}") + if (dsv2Symmetric) { + // The widened predicate has to reach the rebuilt scan, or the merge would keep the + // answer right through the enclosing Filter while losing the row-group pruning that + // is the whole point. checkAnswer and the scan count both stay green in that case. + val dataFilters = v2Scans(df).head.scan match { + case f: FileScan => f.dataFilters + case other => fail(s"expected a FileScan, got ${other.getClass.getSimpleName}") + } + val referenced = dataFilters.flatMap(_.references.map(_.name)).toSet + assert(referenced == Set("a", "b"), + s"the merged scan should carry the widened predicate; got $dataFilters") + } } } } @@ -337,9 +439,9 @@ class FileSourceV2PlanMergingSuite extends QueryTest with SharedSparkSession val path = dir.getCanonicalPath spark.range(0, 20).selectExpr("id AS a", "named_struct('x', id, 'y', id * 2) AS s") .write.format("parquet").save(path) - Seq(true, false).foreach { nestedPruning => - withClue(s"nestedSchemaPruning=$nestedPruning: ") { - withFileView("parquet", path) { + withFileView("parquet", path) { + Seq(true, false).foreach { nestedPruning => + withClue(s"nestedSchemaPruning=$nestedPruning: ") { withSQLConf(SQLConf.NESTED_SCHEMA_PRUNING_ENABLED.key -> nestedPruning.toString) { val df = sql( """ @@ -385,6 +487,14 @@ class FileSourceV2PlanMergingSuite extends QueryTest with SharedSparkSession checkAnswer(df, Row(19, 38)) assertUsesFileSourceV2(df) + // Observe the aggregate itself, not just `!mergeableScan`, which several other pushdowns + // also clear: a decline for one of those reasons must not pass as this one. + assert(v2Scans(df).forall(_.scan match { + case p: ParquetScan => p.pushedAggregate.isDefined + case _ => false + }), + s"the aggregate should have been pushed into both scans:\n" + + df.queryExecution.optimizedPlan) // A pushed aggregate is built on a branch of V2ScanRelationPushDown that never marks the // scan mergeable, so the merge is declined before the capability is consulted. assert(distinctScans(df) == 2, @@ -399,7 +509,9 @@ class FileSourceV2PlanMergingSuite extends QueryTest with SharedSparkSession withTempPath { dir1 => withTempPath { dir2 => writeFlat("parquet", dir1.getCanonicalPath) - writeFlat("parquet", dir2.getCanonicalPath) + // Different rows in the second table, so a merge across the two would change the answer and + // not just the plan shape. + writeFlat("parquet", dir2.getCanonicalPath, start = 100) withFileView("parquet", dir1.getCanonicalPath, viewName = "t1") { withFileView("parquet", dir2.getCanonicalPath, viewName = "t2") { val df = sql( @@ -409,7 +521,8 @@ class FileSourceV2PlanMergingSuite extends QueryTest with SharedSparkSession | (SELECT sum(b) FROM t2 WHERE c = 1) |""".stripMargin) - checkAnswer(df, Row(70, 140)) + // c = 1 selects ids 1, 4, ..., 19 in t1 and 100, 103, ..., 118 in t2. + checkAnswer(df, Row(70, 1526)) assertUsesFileSourceV2(df) assert(distinctScans(df) == 2, s"scans of different tables must remain separate:\n" + @@ -476,8 +589,10 @@ class FileSourceV2PlanMergingSuite extends QueryTest with SharedSparkSession // propagation can widen it; on the V2 path V2ScanRelationPushDown has already pushed it // into the scan as a strict filter by the time MergeSubplans runs, and strict filters // have to be equal to merge. Both paths return the same rows. - assert(mergedCounts(path, query, Row(45, 100), useV1 = true, enableAQE) == ((1, 1))) - assert(mergedCounts(path, query, Row(45, 100), useV1 = false, enableAQE) == ((2, 0))) + assert(mergedCounts(path, query, Row(45, 100), useV1 = true, enableAQE) == ((1, 1)), + "V1 should merge the two partition filters into one subquery") + assert(mergedCounts(path, query, Row(45, 100), useV1 = false, enableAQE) == ((2, 0)), + "V2 should leave the two differing partition filters unmerged") } } } @@ -526,9 +641,12 @@ class FileSourceV2PlanMergingSuite extends QueryTest with SharedSparkSession mode: String, query: String, useV1: Boolean): Seq[Row] = - // Pin CSV column pruning rather than rely on its default: with it off the parser is handed - // the full data schema and every expectation below changes. - withSQLConf(SQLConf.CSV_PARSER_COLUMN_PRUNING.key -> "true") { + // Pin what the expectations below depend on rather than rely on the defaults: with CSV column + // pruning off the parser is handed the full data schema, and with JSON partial results off + // the malformed record yields an all-null row, which the WHERE then drops. + withSQLConf( + SQLConf.CSV_PARSER_COLUMN_PRUNING.key -> "true", + SQLConf.JSON_ENABLE_PARTIAL_RESULTS.key -> "true") { withFileView(format, path, useV1 = useV1, schema = Some(schema), options = Map("mode" -> mode, "columnNameOfCorruptRecord" -> "_corrupt_record")) { val df = sql(query) @@ -571,8 +689,15 @@ class FileSourceV2PlanMergingSuite extends QueryTest with SharedSparkSession withData(shortRowCsv) { path => assert(rows("csv", path, "a long, b long", "FAILFAST", sumQuery, useV1 = false) == Seq(Row(10, 80))) - intercept[SparkException]( + // Pin why V1 threw, not just that it did: a bare intercept would keep passing if this read + // started failing for an unrelated reason, and the subquery-count assertion inside `rows` + // sits after collect(), so it never runs on the throwing path. + val thrown = intercept[SparkException]( rows("csv", path, "a long, b long", "FAILFAST", sumQuery, useV1 = true)) + val chain = + Iterator.iterate[Throwable](thrown)(_.getCause).takeWhile(_ != null).toList + assert(chain.exists(t => Option(t.getMessage).exists(_.contains("MALFORMED_RECORD"))), + s"expected a malformed-record parse failure, got:\n$thrown") } } } diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/planmerging/V2ScanMergingTestHelper.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/planmerging/V2ScanMergingTestHelper.scala index 91ef2a36d4f2f..b822d65ac0ad6 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/execution/planmerging/V2ScanMergingTestHelper.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/planmerging/V2ScanMergingTestHelper.scala @@ -34,8 +34,8 @@ private[planmerging] trait V2ScanMergingTestHelper { /** * A merged subquery is referenced once per original subquery, so the logical plan duplicates it * (physical planning reuses it). Dedupe by canonical form: one distinct scan is consistent with a - * merge and two means it was declined. Two scans that read the same columns canonicalize equal - * either way, so use subquery counts instead when the two column sets match. + * merge and more than one means it was declined. Scans that read the same columns and carry the + * same filters canonicalize equal either way, so use subquery counts for those. */ protected def distinctScans(df: DataFrame): Int = v2Scans(df).map(_.canonicalized).distinct.length } From b6c4b7b748bb6046fd887011a36a8a3c800dfca2 Mon Sep 17 00:00:00 2001 From: yangjie01 Date: Mon, 31 Aug 2026 03:07:10 +0800 Subject: [PATCH 6/7] Drop the V1 arms, split the strictness test, tighten the javadoc --- .../connector/catalog/TableCapability.java | 6 ++ .../FileSourceV2PlanMergingSuite.scala | 78 +++++++++---------- 2 files changed, 42 insertions(+), 42 deletions(-) diff --git a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/TableCapability.java b/sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/TableCapability.java index 599e801f2a07f..e98cf7c11a052 100644 --- a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/TableCapability.java +++ b/sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/TableCapability.java @@ -139,6 +139,12 @@ public enum TableCapability { * obtaining a fresh {@link org.apache.spark.sql.connector.read.ScanBuilder} with the same options * and re-applying the same pushed filters and pruned columns yields an equivalent scan. *

+ * Determinism alone is not enough: widening the set of pruned columns, with the options and + * pushed filters held constant, must not change which rows the scan returns nor the values it + * returns for the columns already asked for. It may at most surface a read error. A source whose + * parser decides what counts as a malformed record from the set of columns it was asked for does + * not meet this, and neither does one that resolves a column by its position in the projection. + *

* Given that contract, Spark builds the merged scan itself: it prunes a fresh ScanBuilder to the * union of both read schemas, re-pushes the (possibly OR-widened) filters, and builds. The merged * scan reads the union of the two scans' columns and a superset of their rows; each original diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/planmerging/FileSourceV2PlanMergingSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/planmerging/FileSourceV2PlanMergingSuite.scala index ba329d3e1afb7..51fb31be8ee3b 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/execution/planmerging/FileSourceV2PlanMergingSuite.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/planmerging/FileSourceV2PlanMergingSuite.scala @@ -17,7 +17,7 @@ package org.apache.spark.sql.execution.planmerging -import org.apache.spark.{SparkConf, SparkException} +import org.apache.spark.SparkConf import org.apache.spark.sql.{DataFrame, QueryTest, Row} import org.apache.spark.sql.connector.catalog.TableCapability import org.apache.spark.sql.execution.{ReusedSubqueryExec, SubqueryExec} @@ -180,7 +180,6 @@ class FileSourceV2PlanMergingSuite extends QueryTest with SharedSparkSession test("SPARK-57205: withhold SCAN_MERGING from a table whose reads are not strict") { withTempPath { dir => val path = dir.getCanonicalPath - // b is written as a string and read as a long, so the reader fails only once it reads b. spark.range(0, 10).selectExpr("id AS a", "cast(id AS string) AS b").write.parquet(path) // The table is built outside the strictness scope on purpose: the gate is evaluated per call, // so it has to answer for the read that is running rather than for the read that built it. @@ -201,6 +200,16 @@ class FileSourceV2PlanMergingSuite extends QueryTest with SharedSparkSession } } } + } + } + + // Separate from the capability test above so that a change breaking the gate names which half it + // broke: an assertion that aborts on the capability never reports on the rows. + test("SPARK-57205: a non-strict read keeps its scans separate") { + withTempPath { dir => + val path = dir.getCanonicalPath + // b is written as a string and read as a long, so the reader fails only once it reads b. + spark.range(0, 10).selectExpr("id AS a", "cast(id AS string) AS b").write.parquet(path) // The scans stay separate, so the a-only scan never reads b and sum(a) is still exact. If // they merged, reading b would fail, ignoreCorruptFiles would swallow it and drop the rest of // the file, and sum(a) would come back null over rows nothing above the scan removed. The @@ -601,9 +610,10 @@ class FileSourceV2PlanMergingSuite extends QueryTest with SharedSparkSession test("SPARK-57205: CSV and JSON decline to merge, so their parsing stays per subquery") { // The parsers are handed just the columns the scan asked for (for CSV under // spark.sql.csv.parser.columnPruning.enabled), so which columns a scan reads decides which - // records it treats as malformed. V1 merges these shapes and parses the union; V2 declines, - // because neither table declares SCAN_MERGING. Each shape below is a case where that decision - // is visible in the result, so adding the capability back to either table fails this test. + // records it treats as malformed. Neither table declares SCAN_MERGING, so each subquery keeps + // its own scan. Every shape below is one where a merged scan would return something else, so + // adding the capability back to either table fails this test. The V1 path does merge them + // today, and does return something else, which is SPARK-59107. val typeErrorCsv = Seq("0,0", "1,10", "2,BAD", "3,30", "4,40") val typeErrorJson = Seq( """{"a":0,"b":0}""", @@ -639,25 +649,23 @@ class FileSourceV2PlanMergingSuite extends QueryTest with SharedSparkSession path: String, schema: String, mode: String, - query: String, - useV1: Boolean): Seq[Row] = + query: String): Seq[Row] = // Pin what the expectations below depend on rather than rely on the defaults: with CSV column // pruning off the parser is handed the full data schema, and with JSON partial results off // the malformed record yields an all-null row, which the WHERE then drops. withSQLConf( SQLConf.CSV_PARSER_COLUMN_PRUNING.key -> "true", SQLConf.JSON_ENABLE_PARTIAL_RESULTS.key -> "true") { - withFileView(format, path, useV1 = useV1, schema = Some(schema), + withFileView(format, path, schema = Some(schema), options = Map("mode" -> mode, "columnNameOfCorruptRecord" -> "_corrupt_record")) { val df = sql(query) - if (useV1) assertUsesFileSourceV1(df) else assertUsesFileSourceV2(df) + assertUsesFileSourceV2(df) val result = df.collect().toSeq - // V1 merges the two subqueries into one; V2 declines. Asserted after collect() so that - // AQE has finalized and the reuse of the merged subquery is visible in the plan. Pinning - // this alongside the rows attributes the difference to the merge decision itself. - assert(subqueryCounts(df) == (if (useV1) ((1, 1)) else ((2, 0))), - s"unexpected subquery counts on ${if (useV1) "V1" else "V2"}:\n" + - df.queryExecution.executedPlan) + // Asserted after collect() so that AQE has finalized and a merged subquery's reuse would + // be visible in the plan. Pinning this alongside the rows attributes them to the merge + // decision itself. + assert(subqueryCounts(df) == ((2, 0)), + s"the two subqueries should keep their own scans:\n${df.queryExecution.executedPlan}") result } } @@ -665,39 +673,25 @@ class FileSourceV2PlanMergingSuite extends QueryTest with SharedSparkSession Seq("csv" -> typeErrorCsv, "json" -> typeErrorJson).foreach { case (format, lines) => withClue(s"format=$format: ") { withData(lines) { path => - // DROPMALFORMED. The a-only scan never parses b, so V2 keeps the record for sum(a). V1's - // merged scan parses the union and drops it for both, giving 8. - assert(rows(format, path, "a long, b long", "DROPMALFORMED", sumQuery, - useV1 = false) == Seq(Row(10, 80))) - assert(rows(format, path, "a long, b long", "DROPMALFORMED", sumQuery, - useV1 = true) == Seq(Row(8, 80))) - - // PERMISSIVE, the default mode, with the corrupt-record column in the schema. V2 does not - // flag the record for the subquery that reads a and the corrupt column; V1's merged scan - // parses b, so the column is populated for a row the first subquery counted as clean. - assert(rows(format, path, "a long, b long, _corrupt_record string", "PERMISSIVE", - corruptQuery, useV1 = false) == Seq(Row(0, 80))) + // DROPMALFORMED. The a-only scan never parses b, so the record survives for sum(a). A + // merged scan would parse the union and drop it for both, giving 8. + assert(rows(format, path, "a long, b long", "DROPMALFORMED", sumQuery) == + Seq(Row(10, 80))) + + // PERMISSIVE, the default mode, with the corrupt-record column in the schema. The + // subquery that reads a and the corrupt column does not flag the record; a merged scan + // would parse b and populate the column for a row that subquery counted as clean. assert(rows(format, path, "a long, b long, _corrupt_record string", "PERMISSIVE", - corruptQuery, useV1 = true) == Seq(Row(1, 80))) + corruptQuery) == Seq(Row(0, 80))) } } } - // FAILFAST, CSV only: JSON has no arity check, so a missing field is null, not malformed. V2 - // returns rows; V1's merged scan parses two columns against a one-token row and throws, which - // is a working query turning into an error. + // FAILFAST, CSV only: JSON has no arity check, so a missing field is null, not malformed. Each + // narrow scan matches the short row, so the query returns rows; a merged scan would parse two + // columns against a one-token row and throw, turning a working query into an error. withData(shortRowCsv) { path => - assert(rows("csv", path, "a long, b long", "FAILFAST", sumQuery, - useV1 = false) == Seq(Row(10, 80))) - // Pin why V1 threw, not just that it did: a bare intercept would keep passing if this read - // started failing for an unrelated reason, and the subquery-count assertion inside `rows` - // sits after collect(), so it never runs on the throwing path. - val thrown = intercept[SparkException]( - rows("csv", path, "a long, b long", "FAILFAST", sumQuery, useV1 = true)) - val chain = - Iterator.iterate[Throwable](thrown)(_.getCause).takeWhile(_ != null).toList - assert(chain.exists(t => Option(t.getMessage).exists(_.contains("MALFORMED_RECORD"))), - s"expected a malformed-record parse failure, got:\n$thrown") + assert(rows("csv", path, "a long, b long", "FAILFAST", sumQuery) == Seq(Row(10, 80))) } } } From 019cda6c095cc99274ca173ddef56b2c4b5e332a Mon Sep 17 00:00:00 2001 From: yangjie01 Date: Mon, 31 Aug 2026 16:59:37 +0800 Subject: [PATCH 7/7] Move the capability's documentation into the prose --- docs/sql-performance-tuning.md | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/docs/sql-performance-tuning.md b/docs/sql-performance-tuning.md index 7553861f9f66f..eff53e60d8c14 100644 --- a/docs/sql-performance-tuning.md +++ b/docs/sql-performance-tuning.md @@ -342,6 +342,8 @@ They are merged into one aggregate that computes `min` and `max` together, so `s Two subplans are merged when their plans match node by node: `Project` lists are unioned, `Aggregate`s must have the same grouping and use the same aggregation implementation (so a `min` is not merged with a `collect_list`), `Filter`s must have the same condition, `Join`s must have the same type, condition and hints, and the leaves must read the same input. Subplans that differ only in their `WHERE` conditions can be merged as well, by turning each side's condition into a boolean column and giving each side's aggregate expressions a `FILTER (WHERE ...)` clause. That is controlled by the configurations below. Queries that still contain a `WITH` clause when this rule runs (one that was not inlined) are skipped. +On the DataSource V2 read path the requirement that the leaves read the same input is relaxed for a source that declares the `SCAN_MERGING` table capability: two leaves that differ only in their projected columns merge into a single scan reading the union of those columns. Among the built-in file formats Parquet, ORC, text and Avro declare it; a format reaches its V2 read path only when it is removed from `spark.sql.sources.useV1SourceList`. A file table withholds the capability when `spark.sql.files.ignoreCorruptFiles` is true, because a read failure in a column that only the other subplan projects would then be swallowed along with the rest of that file's rows, and when `spark.sql.files.ignoreMissingFiles` is true, to match the strictness predicate the file reader uses. Avro withholds it under `positionalFieldMatching`, which resolves a column by its position in the projection. + When only one of the two subplans has a filter, merging is always beneficial, because the unfiltered side reads all the data anyway. This case is on by default, unless the filter has to cross a `Join` to reach the aggregate, which needs the through-join configuration below. When both sides have a filter (the symmetric case), the merged scan filter becomes `OR(f1, f2)`, which is less selective than either original filter and can therefore read more data - for example when the filters prune partitions or Parquet row groups. That is why the symmetric case is disabled by default. Still, it is worth considering on queries that compute several differently filtered aggregates over the same table, which is a common analytical shape: @@ -384,7 +386,7 @@ In TPC-DS benchmark runs, enabling symmetric filter propagation made `q9` and `q spark.sql.optimizer.mergeSubplans.filterPropagation.dsv2SymmetricFilterPropagation.enabled false - When true, two DataSource V2 scans that pushed the same strictly enforced filters but carry different best-effort (post-scan) filters can be merged even when spark.sql.optimizer.mergeSubplans.filterPropagation.symmetricFilterPropagation.enabled is false. In this case widening cannot change the set of rows the scan is required to return, as the strict filters are re-pushed unchanged and the enclosing Filter re-checks the rest above the scan. This applies only to V2 sources that opt in to scan merging with the SCAN_MERGING table capability. Among the built-in file formats, Parquet, ORC, text and Avro opt in on their V2 read path, which a format reaches only when it is removed from spark.sql.sources.useV1SourceList; there the strictly enforced filters are the partition filters, so this configuration lets two scans over the same partitions but with different data filters merge. A file table withholds the capability when spark.sql.files.ignoreCorruptFiles or spark.sql.files.ignoreMissingFiles is true, because a read failure in a column that only the other scan projects would then be swallowed along with the rest of that file's rows, and Avro withholds it under positionalFieldMatching, which resolves a column by its position in the projection. + When true, two DataSource V2 scans that pushed the same strictly enforced filters but carry different best-effort (post-scan) filters can be merged even when spark.sql.optimizer.mergeSubplans.filterPropagation.symmetricFilterPropagation.enabled is false. In this case widening cannot change the set of rows the scan is required to return, as the strict filters are re-pushed unchanged and the enclosing Filter re-checks the rest above the scan. This applies only to V2 sources that opt in to scan merging with the SCAN_MERGING table capability. For a file source the strictly enforced filters are the partition filters, so this configuration is what lets two scans over the same partitions but with different data filters merge. 4.3.0