dongjoon-hyun commented on code in PR #58340:
URL: https://github.com/apache/spark/pull/58340#discussion_r3876055606
##########
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)
+ // 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)
Review Comment:
**[correctness / CONFIRMED]** CSV/JSON in non-PERMISSIVE parse modes violate
`SCAN_MERGING`'s documented contract, yet the capability is declared
unconditionally. `TableCapability.java:144` promises the merged scan reads "a
superset of their rows", but with `DROPMALFORMED` the merged scan parses the
union of columns and drops a record malformed only in the other subquery's
column, returning *fewer* rows than the original scan — the new suite's own
comment concedes this, and pins `sum(a)` flipping from 10 to 8 depending on
whether an unrelated sibling subquery triggers a merge. Today's only consumer
(`PlanMerger`) survives because each side re-filters above the scan, but a
future `SCAN_MERGING` consumer relying on the superset premise (e.g. reusing a
wider scan for a narrower query without re-checking) would silently drop rows
for CSV/JSON. `CSVTable`/`JsonTable` have the parse mode in hand, so the
capability could be withheld when `mode` is not `PERMISSIVE` — or at minimum the
exception should be carved out in the capability javadoc and this comment
rather than living only in a test comment and a migration-guide sentence.
##########
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)
+ // A file table meets the determinism contract SCAN_MERGING requires:
`fileIndex` is a lazy val,
Review Comment:
**[altitude / CONFIRMED]** Adding the capability to the shared
`FileTable.CAPABILITIES` opts in every subclass at once — including out-of-tree
file connectors (the class is public, `capabilities()` is non-final, and
cross-module extension is real: `connector/avro`'s `AvroTable`). But the
determinism this comment cites is a property of the *subclasses*
(`newScanBuilder` is abstract), which the base class cannot enforce. A
third-party `FileTable` whose `ScanBuilder` is not deterministic per (options,
filters, columns) silently inherits the contract on upgrade and gets its scans
fused, with no opt-out other than discovering it must override
`capabilities()`. Mitigating: `org.apache.spark.sql.execution` is nominally
internal (blanket MiMa exclusion), so this is unsupported-but-common extension
rather than stable API. Per-format declaration in the six built-in tables — or
at least documenting the inherited contract in `FileTable`'s scaladoc for
subclass authors — would avoid the
silent opt-in.
##########
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`.
Review Comment:
**[documentation / CONFIRMED + PLAUSIBLE]** Two issues in this bullet:
1. It spells out the behavior change only for `DROPMALFORMED`, but the same
widened-parsed-columns mechanism changes results in the other two modes as well
(`FailureSafeParser.scala:41-84`): in `PERMISSIVE` mode, a record malformed
only in the other subquery's column flips `columnNameOfCorruptRecord` from null
to populated (and nulls that row's other fields); in `FAILFAST` mode, a query
that previously succeeded now throws
`malformedRecordsDetectedInRecordParsingError` — the harshest flip, with no
explicit warning here. Consider enumerating all three modes or saying "with any
parse `mode`".
2. The clause "merging there now matches what the V1 path already did"
mildly overclaims: the PR's own test "V1 merges differing partition filters, V2
does not" pins a shape the suite itself labels "Known gap against V1". A
qualifier like "for these shapes" would keep a skimming reader from
over-generalizing.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/planmerging/FileSourceV2PlanMergingSuite.scala:
##########
@@ -0,0 +1,498 @@
+/*
+ * 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.
+ *
+ * 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 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"
+
+ 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 =
Review Comment:
**[simplification / CONFIRMED]** This assert can never fire. All 4 call
sites run after `checkAnswer`, which has already physically planned and
executed every subquery — and a leaked bare `DataSourceV2Relation` has no
batch-read physical strategy (`DataSourceV2Strategy` plans only
`DataSourceV2ScanRelation` for reads), so planning would already have failed
inside `checkAnswer` with `QueryPlanner`'s "No plan for" assertion before this
helper's friendlier message could be reached. It was copied from
`DSv2PlanMergingSuite`, where the placement is equally post-`checkAnswer` and
equally dead. Deleting the helper and its 4 calls loses nothing.
##########
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)
+ // 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
Review Comment:
**[correctness / PLAUSIBLE]** The determinism justification here holds only
per table *instance*: `PlanMerger`'s gate is canonicalized-plan equality
(`PlanMerger.scala:680`), the table case classes exclude the lazy `fileIndex`
from equality, and `InMemoryFileIndex.equals` compares root paths only — so
scans from two table instances whose listings were taken at different times
(df1 read, files land, df2 read) are canonically equal, and the rebuilt merged
scan reads one side's snapshot for both. This is largely an extension of
long-standing rootPaths-only equality semantics (identical-column scans were
already collapsed by the capability-free identical-plan path and
`ReuseSubquery`/`ReuseExchange`, and V1 merges these shapes the same way), so
no action beyond wording may be needed — but the comment overclaims as written;
stating it as an instance-level property (or noting the cross-instance caveat)
would keep it accurate.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/planmerging/FileSourceV2PlanMergingSuite.scala:
##########
@@ -0,0 +1,498 @@
+/*
+ * 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.
+ *
+ * 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 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"
+
+ 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 canonically equal,
+ // so they merge on PlanMerger's identical-plan path, which
needs no capability.
+ assert(distinctScans(df) == (if (nestedPruning) 2 else 1),
Review Comment:
**[test-coverage / CONFIRMED]** The `nestedPruning=false` branch of this
assertion is vacuous. With pruning off, both subqueries' scans read the
identical whole struct `s`, and `FileScan.equals` (`FileScan.scala:104-111`)
compares only fileIndex, readSchema and normalized filters — so even if the
identical-plan merge were declined and two separate scan relations remained,
they would canonicalize equal and `distinctScans` would still return 1. The
comment says this branch verifies merging "on PlanMerger's identical-plan
path", but a regression that stops that merge would go undetected.
`subqueryCounts` (already defined in this suite) would actually distinguish:
(1, 1) merged vs (2, 0) unmerged. The `nestedPruning=true` branch is fine —
there the readSchemas differ, so `== 2` is meaningful.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/planmerging/FileSourceV2PlanMergingSuite.scala:
##########
@@ -0,0 +1,498 @@
+/*
+ * 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.
+ *
+ * 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 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"
+
+ 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 =>
Review Comment:
**[test-coverage / CONFIRMED]** `AvroTable` inherits `SCAN_MERGING` through
`FileTable`, but this test enumerates only parquet/orc/json/csv/text and
nothing in `connector/avro` asserts the capability or exercises merging (grep
for `SCAN_MERGING`/`PlanMerging`/`MergeSubplans` there returns zero hits). An
Avro-specific merge defect — e.g. interaction of the union-column rebuild with
`positionalFieldMatching` or Avro's filter pushdown — would ship
enabled-by-default with zero coverage, while the migration guide tells users
"the built-in file formats" are included. A small merging test in
`connector/avro` (or at least a capability assertion there) would close the gap.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/planmerging/FileSourceV2PlanMergingSuite.scala:
##########
@@ -0,0 +1,498 @@
+/*
+ * 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.
+ *
+ * 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 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"
+
+ 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] =
Review Comment:
**[reuse / CONFIRMED]** `v2Scans` is a character-for-character copy of the
private helper in `DSv2PlanMergingSuite.scala:59-62`, and
`assertNoPlaceholderRelation` below duplicates its lines 69-75 — same test
package. A future change to how merged V2 scans are collected (e.g. a new
wrapper node) must land in both suites or they silently measure different
things. A small shared package-private trait would collapse the duplication,
and is also a natural home for `distinctScans`/`subqueryCounts` (the latter's
measure is inlined eleven times in `PlanMergingSuite`).
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/planmerging/FileSourceV2PlanMergingSuite.scala:
##########
@@ -0,0 +1,498 @@
+/*
+ * 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.
+ *
+ * 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 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"
+
+ 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 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}")
+ }
+ }
+ }
+ }
+ }
+ }
+
+ 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: ") {
+ // 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)))
+ }
+ }
+ }
+ }
+
+ 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(format, path, useV1 = useV1, schema = Some("a long, b
long"),
+ options = Map("mode" -> "DROPMALFORMED")) {
+ 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))
Review Comment:
**[efficiency / CONFIRMED]** `rows(useV1 = false)` is called in both
asserts, and each call re-registers the temp view and re-executes the whole
two-subquery query via `collect()` — one full redundant execution per format in
CI. Binding `val v2Rows = rows(useV1 = false)` once and reusing it in both
asserts eliminates the waste. Same pattern class elsewhere:
`withTempPath`/`writeFlat` sit inside the `dsv2Symmetric` flag loop in the
differing-data-filters test, writing identical data twice, where the
nested-fields test already hoists the write outside its flag loop.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
