You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "LuciferYang (via GitHub)" <gi...@apache.org> on 2026/08/27 13:26:56 UTC
[PR] [SPARK-57205][SQL] Declare SCAN_MERGING on the built-in file sources [spark]
LuciferYang opened a new pull request, #58340:
URL: https://github.com/apache/spark/pull/58340
### What changes were proposed in this pull request?
`FileTable` declares `TableCapability.SCAN_MERGING`, which lets the built-in file formats (Parquet, ORC, CSV, JSON, text, Avro) take part in the DSv2 scan merging added by #57360 (SPARK-40259). That PR put the whole mechanism on Spark's side: `PlanMerger` drives `V2ScanRelationPushDown.rebuildScan` to rebuild the merged scan, and a source supplies no merge logic of its own. It only declares that what its scans read is determined by the filters pushed and the columns pruned. The description of #57360 noted that no built-in source declared the capability yet; this is that follow-up.
The production change is one line plus a comment recording why the contract holds for a file table: `fileIndex` is a table-level lazy val, so every scan built from the table lists the same files, and `newScanBuilder` returns a fresh builder over `mergedOptions(options)`.
The rest is a new test suite and two documentation updates.
### Why are the changes needed?
Two scans of the same file table that differ only in their projected columns cannot be reused today. A file source folds its data filters into the `FileScan` object, where they are used to list files and prune row groups, and `FileScan.equals` compares them. Two subquery scans over the same path are therefore not canonically identical, and `PlanMerger`'s identical-plan fast path does not fire. On the V1 path those filters sit in a `Filter` above an identical `LogicalRelation`, so the fast path does fire. The V2 file source path merges strictly less than V1 does, and declaring the capability closes most of that gap.
### Does this PR introduce _any_ user-facing change?
Yes, on the V2 file source read path, which a format reaches only when it is removed from `spark.sql.sources.useV1SourceList`.
Plan shape: subqueries over the same file table that differ only in their projected columns now collapse into a single scan reading the union of those columns. This needs no new configuration and is on by default.
Results: for CSV and JSON the merged scan parses the union of both scans' columns, so with `mode` set to `DROPMALFORMED` a record malformed only in the columns the other scan reads is now dropped for both. V1 already read the union after merging, so this makes the two paths agree. The migration guide records it.
`docs/sql-performance-tuning.md`: the `dsv2SymmetricFilterPropagation` entry ended with "no built-in source does", which this PR falsifies, so that sentence is rewritten.
Three gaps are left in place:
1. Two scans with *different* partition filters do not merge, while V1 merges them. A partition filter is fully enforced by the V2 scan and reported as strict, so widening it to `OR` would leave the merged scan returning rows nothing above it filters out. V1 keeps the filter in a `Filter` node until physical planning, so symmetric propagation can widen it there. This is the differing-strict-filter follow-up #57360 lists. `FileSourceV2PlanMergingSuite` pins the current numbers so the gap stays visible.
2. Parquet and ORC nested columns do not merge while nested schema pruning is on. Each side narrows the struct to the field it reads, so the read column is no longer a same-type subset of the relation's column, and the field ordinals in the extractors above the scan would be resolved against the wrong type. #57360 lists widening to the union of nested fields as a follow-up.
3. Differing data filters need `spark.sql.optimizer.mergeSubplans.filterPropagation.dsv2SymmetricFilterPropagation.enabled`, which defaults to false. On the V2 path that config alone is enough, where V1 needs the broader `symmetricFilterPropagation`, because the strict filters are equal and widening the best-effort ones cannot change the row set the scan is required to return.
I have not measured TPC-DS yet, so this PR makes no performance claim.
### How was this patch tested?
New suite `FileSourceV2PlanMergingSuite`, 12 tests. Every built-in file table declares the capability; scans differing only in projected columns merge for parquet, orc, json and csv; scans over the same partition filter merge with that filter re-pushed strict onto the rebuilt scan; three scans merge into one; differing data filters merge only with the dsv2 config on. Declines are covered too: differing partition filters, nested-pruned columns, a pushed aggregate, and two different tables. Then V1/V2 parity on three shapes, the partition-filter gap pinned as the one shape where they disagree, and the CSV and JSON `DROPMALFORMED` behavior.
Every test asserts the plan is on the V2 path before asserting anything about merging. SQL-on-file and catalog tables resolve to the V1 `FileFormat` regardless of `spark.sql.sources.useV1SourceList`, so a suite driven by that config alone would quietly run V1 on both sides.
Removing `SCAN_MERGING` from `FileTable.CAPABILITIES` fails 7 of the 12 tests. Of the 5 that still pass, 4 assert a decline and one merges through the identical-plan fast path, which needs no capability.
Regression: the `planmerging` suites, `ExplainSuite` and `ExplainSuiteAE`, `FileBasedDataSourceSuite`, `FileTableSuite`, `OrcV2SchemaPruningSuite`, `ParquetV2SchemaPruningSuite`, `ParquetV2FilterSuite`, `SubquerySuite`, `SameResultSuite`, `ParquetV2AggregatePushDownSuite`, `OrcV2AggregatePushDownSuite`, `DataSourceV2Suite` and `AvroV2Suite`.
No golden file or `PlanStabilitySuite` plan needed regenerating. Those build their tables with `CREATE TABLE ... USING <format>`, which resolves to the V1 `FileFormat`, so none of them reaches a V2 file scan.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57205][SQL] Declare SCAN_MERGING on the built-in file sources [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on code in PR #58340:
URL: https://github.com/apache/spark/pull/58340#discussion_r3879451300
##########
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 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.planme
rging.MergeSubplans` to `spark.sql.optimizer.excludedRules`.
Review Comment:
**Finding 1.** The first remedy does not restore the previous behaviour.
Both effects this bullet describes are already what V1 does. V1 merges two scans of the same relation that differ only in their projected columns, and the merged required schema is the union. So moving the format back into `spark.sql.sources.useV1SourceList` hands the user the *new* rows, not the old ones.
Measured on this PR's head, with the suite's own data (`Seq("0,0", "1,10", "2,BAD", "3,30", "4,40")`, schema `a long, b long`, `mode=DROPMALFORMED`) and the suite's two-subquery query:
| path | default | `MergeSubplans` in `excludedRules` |
|---|---|---|
| V2 | `[8, 80]` | `[10, 80]` |
| V1 | `[8, 80]` | `[10, 80]` |
The V1 column is already pinned two files over, at `FileSourceV2PlanMergingSuite.scala:476` (`assert(rows("DROPMALFORMED", useV1 = true) == droppedV2)`), so this sentence contradicts the test that ships with it.
Only the `excludedRules` remedy works. Suggest ending the bullet with just that, e.g. "To restore the previous behavior, disable subplan merging by adding `org.apache.spark.sql.execution.planmerging.MergeSubplans` to `spark.sql.optimizer.excludedRules`; falling back to `spark.sql.sources.useV1SourceList` does not help, because the V1 path merges these shapes the same way."
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/planmerging/FileSourceV2PlanMergingSuite.scala:
##########
@@ -0,0 +1,493 @@
+/*
+ * 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.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.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 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")
+
+ 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 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}")
+ }
+ }
+
+ /** `(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) {
Review Comment:
**Finding 4.** This helper never checks which read path it ran on, so the parity claim is asserted but not measured.
The suite scaladoc says "every test goes through `DataFrameReader` and asserts the plan is V2 before asserting anything about merging". `mergedCounts` and the `rows` helper in the parse-mode test are the two that do not. In "V1 and V2 file sources merge the same subquery shapes" the only assertions are `v1 == v2` and `v1 == ((1, 1))`; if `USE_V1_SOURCE_LIST` ever stopped taking effect for a format, both arms would run V2, both would return `(1, 1)`, and the test would pass while measuring nothing about V1.
"V1 merges differing partition filters, V2 does not" is safe already, because its two arms assert different values. Only this one needs the check:
```scala
val df = sql(query)
checkAnswer(df, expected)
if (useV1) assertUsesFileSourceV1(df) else assertUsesFileSourceV2(df)
subqueryCounts(df)
```
with `assertUsesFileSourceV1` the one-line mirror of the existing helper (`collectWithSubqueries { case r: LogicalRelation => r }.nonEmpty`, and no `DataSourceV2ScanRelation`). Worth calling `assertUsesFileSourceV2` in the V2 arm of the parse-mode `rows` too.
##########
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.
Review Comment:
**Finding 7.** "two means it was declined" is not true in the direction that matters.
A decline between two scans that read the same columns still canonicalizes to one distinct scan, which is exactly the trap the nested-fields test documents at its `nestedPruning=false` arm. Reading it as stated is how that arm became vacuous in the first place. Suggest: "one distinct scan is consistent with a merge, two means it was declined; two identical scans canonicalize equal either way, so use `subqueryCounts` when the columns match."
##########
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.
Review Comment:
**Finding 5.** The opt-out this paragraph prescribes cannot reuse the base set.
`FileTable.CAPABILITIES` is `private` to the companion, so a subclass following this advice has to write `util.EnumSet.of(BATCH_READ, BATCH_WRITE)` from scratch. It then silently drops whatever the base set gains later: the next capability added to `FileTable` would reach every built-in format but not that subclass, and nothing would fail.
A seam on the class makes the opt-out one line, keeps the base set in one place, and is testable:
```scala
/** Whether this table meets the `SCAN_MERGING` contract described above. */
protected def supportsScanMerging: Boolean = true
override def capabilities: java.util.Set[TableCapability] =
if (supportsScanMerging) FileTable.CAPABILITIES_WITH_SCAN_MERGING else FileTable.CAPABILITIES
```
This is also the seam finding 6 would need, if you take that route.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57205][SQL] Declare SCAN_MERGING on the built-in file sources [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on PR #58340:
URL: https://github.com/apache/spark/pull/58340#issuecomment-5476282301
Thanks @LuciferYang, nothing blocking from my end.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57205][SQL] Declare SCAN_MERGING on the built-in file sources [spark]
Posted by "LuciferYang (via GitHub)" <gi...@apache.org>.
LuciferYang commented on PR #58340:
URL: https://github.com/apache/spark/pull/58340#issuecomment-5454199423
Thanks, this is all taken. The change is now what you proposed: `FileTable.CAPABILITIES` no longer carries `SCAN_MERGING`, a `CAPABILITIES_WITH_SCAN_MERGING` sits beside it, and a `protected def supportsScanMerging: Boolean = false` seam picks between them. `ParquetTable`, `OrcTable`, `TextTable` and `AvroTable` override it; `CSVTable` and `JsonTable` do not. Pushed as 7dd18334f98.
1. The migration-guide entry is gone rather than corrected. With CSV and JSON out there is no user-visible result change left to document.
2. Conceded above, and now pinned rather than argued. The parse-mode test asserts all three of your shapes on both read paths: `DROPMALFORMED` `[10, 80]` on V2 against `[8, 80]` on V1, `PERMISSIVE` with `_corrupt_record` `[0, 80]` against `[1, 80]`, and `FAILFAST` with the short CSV row returning rows on V2 and throwing on V1.
3. The description is rewritten against the current diff, including the helper, the `DSv2PlanMergingSuite` change and the Avro test. Your correction to the count stands: rerun on the current suite it is 6 of 13 plus the Avro test, and of the 7 that still pass, 4 assert a decline, one asserts different values on the two paths, and two are the new CSV and JSON tests.
4. Taken. `assertUsesFileSourceV1` is the mirror you sketched, and both `mergedCounts` and the parse-mode helper call one or the other. I also took the `subqueryCounts` suggestion for `rows`, which earned itself: asserting before `collect()` reported `(2, 0)` on V1, because AQE had not finalized and the reuse was not yet in the plan, so it moved after.
5. Taken as written, including the name.
6. Taken. One line added to the criterion, because the `FAILFAST` measurement taken alone proves more than it should: a corrupt column chunk, or `datetimeRebaseModeInRead=EXCEPTION` meeting an ancient value, makes any format throw on a column the narrower scan had pruned, so "merging can surface a new error" would empty the list. What separates CSV and JSON is the silent and unrecoverable change to row membership and row content, and `FAILFAST` is that same projection dependence surfacing as an error. The four override comments say it that way, so a later reader does not ask why Parquet is not out too.
7. Taken, with your wording.
That leaves the V1 side, which I will file separately rather than sequence this behind. Fixing it needs a signal on `FileFormat` plus plumbing so the decline can see whether the projections differ, which the leaf arm cannot: only `MergeContext` threads down and both leaves carry the full schema. It also changes behaviour users have today. Your reading that V1 is the bug rather than the target is what the fourth gap in the description now says.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57205][SQL] Declare SCAN_MERGING on the built-in file sources [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on code in PR #58340:
URL: https://github.com/apache/spark/pull/58340#discussion_r3879475838
##########
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:
**Finding 2.** Adding measurements to your first point, @dongjoon-hyun - both mode claims hold. The rebuttal above missed them because its data only exercises a type error inside one column.
**`PERMISSIVE`, with `columnNameOfCorruptRecord` in the schema.** Schema `a long, b long, _corrupt_record string`, `mode=PERMISSIVE`, query `SELECT (SELECT count(_corrupt_record) FROM t WHERE a >= 0), (SELECT sum(b) FROM t WHERE a >= 0)`:
| format | default | `MergeSubplans` excluded |
|---|---|---|
| csv, row `2,BAD` | `[1, 80]` | `[0, 80]` |
| json, row `{"a":2,"b":"BAD"}` | `[1, 80]` | `[0, 80]` |
The narrow scan parses only `a`, so the record is not corrupt for it. The merged scan parses `b` as well, so `_corrupt_record` is populated for a row the first subquery had counted as clean. `FailureSafeParser.toResultRow` sets the corrupt column whenever the raw parser threw, and which columns get parsed is exactly what the merge widens. Same on V1.
**`FAILFAST`, with a short CSV row.** Data `Seq("0,0", "1,10", "2", "3,30", "4,40")`, schema `a long, b long`, the suite's `sum(a)`/`sum(b)` query:
| path | default | `MergeSubplans` excluded |
|---|---|---|
| V2 | throws | `[10, 80]` |
| V1 | throws | `[10, 80]` |
Neither narrow scan is malformed here. With column pruning on, `UnivocityParser.parsedSchema` is the *pruned* schema and univocity is handed `selectIndexes`, so a one-token row matches a one-column parsed schema. The merged scan parses two columns, the `tokens.length != parsedSchema.length` branch fires (`UnivocityParser.scala:406`), and the query throws where it previously returned rows. That is the success-into-failure flip you described.
@LuciferYang, two consequences. The guide's "so neither of those modes changes" is not accurate as written, and the suite's `intercept[SparkException](rows("FAILFAST", ...))` is two-sided only because `2,BAD` makes the merged and the unmerged scan both throw - a short row would make it discriminating.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57205][SQL] Declare SCAN_MERGING on the built-in file sources [spark]
Posted by "LuciferYang (via GitHub)" <gi...@apache.org>.
LuciferYang commented on code in PR #58340:
URL: https://github.com/apache/spark/pull/58340#discussion_r3878158743
##########
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:
Thanks for tracing this through. The mechanism is right: canonical equality does not distinguish two table instances whose listings were taken at different times, because the table case classes exclude the lazy `fileIndex` and `InMemoryFileIndex.equals` compares root paths only.
On the wording, I read the comment as already staying inside instance scope: `fileIndex` is a lazy val "so every scan built from this table lists the same files", and the sentence after it is about rebuilding a scan from the same builder. I would rather not add a cross-instance caveat, for the reason you gave: two scans over the same paths reading the same columns were already collapsed before this change, by the identical-plan path and by `ReuseSubquery`/`ReuseExchange`, and V1 collapses them too. A caveat here would document `InMemoryFileIndex`'s listing semantics rather than anything the capability introduces, and it would be easier to find next to that equality.
One note: the comment has moved since you read it. The contract now lives in a class-level scaladoc on `FileTable` addressed to subclass authors, and what is left at `CAPABILITIES` is the two facts that make it hold for the built-in tables. If the new text still reads as overclaiming, point at the clause and I will fix it.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57205][SQL] Declare SCAN_MERGING on the built-in file sources [spark]
Posted by "LuciferYang (via GitHub)" <gi...@apache.org>.
LuciferYang commented on PR #58340:
URL: https://github.com/apache/spark/pull/58340#issuecomment-5476462224
Thanks for your thorough and detailed review. @peter-toth
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57205][SQL] Declare SCAN_MERGING on the built-in file sources [spark]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on PR #58340:
URL: https://github.com/apache/spark/pull/58340#issuecomment-5459202499
To @peter-toth , if you don't mind, could you ask your agent to exclude the pre-existing comments or avoid to mention the other GitHub Account IDs more than once. It's because I received too many pings from your comments. Since it's a little disruptive to me (or other GitHub Account).
<img width="933" height="1129" alt="Screenshot 2026-08-28 at 17 28 32" src="https://github.com/user-attachments/assets/69ca7683-c598-4703-a562-2b8e4673667d" />
<img width="530" height="132" alt="Screenshot 2026-08-28 at 17 31 27" src="https://github.com/user-attachments/assets/c9881033-f470-4661-a773-da6aecff2600" />
<img width="460" height="76" alt="Screenshot 2026-08-28 at 17 31 36" src="https://github.com/user-attachments/assets/e08fd91a-7e36-43c6-9615-7830cc630ab0" />
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57205][SQL] Declare SCAN_MERGING on the built-in file sources [spark]
Posted by "LuciferYang (via GitHub)" <gi...@apache.org>.
LuciferYang commented on code in PR #58340:
URL: https://github.com/apache/spark/pull/58340#discussion_r3878157694
##########
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:
Thanks, this is a real gap, and you have put it more precisely than the suite's own comment does. Agreed on the mechanism: with `DROPMALFORMED` the merged scan parses the union of the columns, so a record malformed only in the other subquery's column is dropped for both, and the merged scan returns fewer rows than the narrower scan did. That is not a superset.
On withholding the capability when `mode` is not `PERMISSIVE`, I measured what that would buy first. Same query as the suite's, CSV and JSON, all three modes, with `MergeSubplans` on and excluded:
| mode | merged | not merged |
|---|---|---|
| `PERMISSIVE` | `[10, 80]` | `[10, 80]` |
| `DROPMALFORMED` | `[8, 80]` | `[10, 80]` |
| `FAILFAST` | throws | throws |
The V1 path also returns `[8, 80]` under `DROPMALFORMED`; the suite asserts it, and it is why the migration guide says the V2 path now matches V1. Gating on the parse mode would leave the same query returning 10 on V2 and 8 on V1, so it trades a violated javadoc premise for a result that depends on `spark.sql.sources.useV1SourceList`. I would rather the two paths agree.
If the `[8, 80]` result is itself the thing to fix, that is a question about both paths rather than about this capability: V1 returned 8 before this PR too, and this PR does not touch it. I am happy to take that as a follow-up, either as one change covering both paths or as a fix to V1 that this PR then aligns to. If you would rather see V1 fixed first, I can hold this one until that lands.
`PERMISSIVE` keeps the fields it did parse, so widening the parsed columns changes nothing there. `FAILFAST` throws whether or not the scans merged, because the merged scan reads the union of the two column sets and the scan that reads `b` already throws on its own. Both are now assertions in the suite rather than claims in a comment.
Your second suggestion is the one I would like to act on: the exception belongs in the capability's own javadoc, not only in a test comment and a migration-guide sentence. `TableCapability` is `@Evolving` public API and that sentence arrived with #57360, so I would rather not reword it unilaterally.
cc @peter-toth, who wrote #57360 and the plan-merging work under it. Does the "superset of their rows" sentence need a qualifier for sources whose parsing depends on which columns are read, and would you rather that land here or in a follow-up? I am happy either way.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57205][SQL] Declare SCAN_MERGING on the built-in file sources [spark]
Posted by "LuciferYang (via GitHub)" <gi...@apache.org>.
LuciferYang commented on PR #58340:
URL: https://github.com/apache/spark/pull/58340#issuecomment-5448875249
also cc @cloud-fan
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57205][SQL] Declare SCAN_MERGING on the built-in file sources [spark]
Posted by "LuciferYang (via GitHub)" <gi...@apache.org>.
LuciferYang commented on PR #58340:
URL: https://github.com/apache/spark/pull/58340#issuecomment-5448854949
Thanks for the review. Eight of the ten findings are addressed; the two on the `FileTable` comment have replies inline instead.
Fixed as suggested:
- The `nestedPruning=false` branch was vacuous, as you said. That arm now asserts `subqueryCounts`, `(1, 1)` merged against `(2, 0)` declined, and `distinctScans == 2` stays only on the `nestedPruning=true` arm where the two `readSchema`s differ. The comment says why the old assertion could not tell the two apart.
- `assertNoPlaceholderRelation` and its four call sites are gone. Your reasoning holds: `DataSourceV2Strategy`'s only batch-read case matches `DataSourceV2ScanRelation`, so a leaked bare relation fails planning inside `checkAnswer` before the helper runs.
- `v2Scans` no longer exists twice. It moved with `distinctScans` into a package-private `V2ScanMergingTestHelper` that both suites mix in. I left `PlanMergingSuite`'s inlined subquery counting alone to keep this diff to the duplication you pointed at.
- The redundant executions are gone: the parse-mode test binds the result once, and the differing-data-filters test writes its data outside the flag loop, matching the nested-fields test.
- `connector/avro` has coverage now. A test in `AvroV2Suite` asserts `AvroTable` carries the capability and that two scans differing only in their projected columns fuse into one reading the union.
- The migration guide says "for these shapes" where it claimed to match V1, so the partition-filter gap the suite pins does not read as closed.
Adjusted rather than taken as written:
- On the shared `CAPABILITIES`: `FileTable` now has a class-level scaladoc stating the inherited contract for subclass authors, which was your fallback suggestion. I kept the single declaration rather than repeating it in the six built-in tables, because a seventh format added later would then silently not participate, and that drift is harder to notice than the inherited opt-in.
- On the parse modes: I measured all three rather than extending the guide to cover them. `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 the two column sets and the scan reading the malformed column already throws on its own. The guide now records that neither mode changes instead of widening the warning, and both are assertions in the suite. Numbers are in the reply on the `FileTable` comment.
`FileSourceV2PlanMergingSuite`, `DSv2PlanMergingSuite` and `PlanMergingSuite` pass, 30 tests, along with the new Avro test and scalastyle on `sql` and `connector/avro`.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57205][SQL] Declare SCAN_MERGING on the built-in file sources [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on code in PR #58340:
URL: https://github.com/apache/spark/pull/58340#discussion_r3890184655
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/FileTable.scala:
##########
@@ -37,6 +37,18 @@ 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.
+ *
+ * A subclass opts in to the `SCAN_MERGING` capability by overriding [[supportsScanMerging]], which
+ * 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
Review Comment:
**Finding 13.** This is the right criterion, and it is stricter than the one `TableCapability.SCAN_MERGING` states. A third-party connector author reads only the javadoc.
`TableCapability.java:134-139`:
> By returning this capability a table declares a determinism contract: holding the scan options constant, the rows and columns a scan reads are fully determined by the filters pushed via `SupportsPushDownV2Filters` and the columns pruned via `SupportsPushDownRequiredColumns`.
A CSV table satisfies that as written. Its rows *are* fully determined by the pruned column set - that is exactly the dependence, and re-pruning to the same set does yield an equivalent scan. So `CSVTable` could declare the capability without contradicting a word of it, and the next paragraph's "the merged scan reads ... a superset of their rows" then simply does not follow from what the table promised.
What closes the gap is the monotonicity clause you wrote here: widening the pruned set must not change the rows or the values. Suggest adding it to the javadoc, right after the determinism sentence:
```java
* 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. A source whose parser decides what counts as a
* malformed record from the set of columns it was asked for does not meet this.
```
Different clause from the one at [r3879483408](https://github.com/apache/spark/pull/58340#discussion_r3879483408) - there I said not to weaken "a superset of their rows", and this asks to strengthen the sentence above it so that claim actually follows. Fine as a follow-up on SPARK-40259 if you would rather not widen this diff, but the capability is `@since 4.3.0` and unreleased, so it is cheaper now than after.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/planmerging/FileSourceV2PlanMergingSuite.scala:
##########
@@ -0,0 +1,703 @@
+/*
+ * 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.{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, 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
+
+/**
+ * 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. 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
+ * which read path the plan took before asserting anything about merging.
+ */
+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")
+
+ // 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"
+
+ 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)
+
+ 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`. 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,
+ 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 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}")
+ }
+ }
+
+ 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
+ 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 AQE as given and
+ * 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)
+ if (useV1) assertUsesFileSourceV1(df) else assertUsesFileSourceV2(df)
+ subqueryCounts(df)
+ }
+ }
+ }
+
+ 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: 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),
Review Comment:
**Finding 15.** This test makes two independent claims - the capability flips with the conf, and the merge that flip prevents would return wrong rows - and the first aborts before the second can report. That is the gap your mutation note records.
I measured the second half on its own. With `capabilities` changed to `if (supportsScanMerging)`, and the end-to-end block lifted into a scratch suite so nothing aborts first:
```
SCRATCH rows = [null,0]
SCRATCH distinctScans = 1
```
against `Row(45, 0)` and 2. So the assertion is effective and worth keeping; it is only unreportable where it sits.
Splitting the test in two - one on the capability, one on the rows - costs a `withTempPath` and makes the mutation name which half broke:
```scala
test("SPARK-57205: withhold SCAN_MERGING from a table whose reads are not strict") { ... }
test("SPARK-57205: a non-strict read keeps its scans separate") { ... }
```
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/FileTable.scala:
##########
@@ -111,7 +123,32 @@ 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 && hasStrictFileReads) {
+ 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
+
+ /**
+ * 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 = {
Review Comment:
**Finding 16.** #58411 lifts this exact predicate onto `FileSourceOptions.hasStrictFileReads` and points both existing spellings at it, `FileScanRDD` and the cache-repeatability check in `InMemoryRelation`. Once both PRs are in, this is the only place still writing it out.
```scala
private def hasStrictFileReads: Boolean =
new FileSourceOptions(options.asCaseSensitiveMap.asScala.toMap).hasStrictFileReads
```
Nothing to do now if this lands first - just worth a line in the description so the follow-up is not lost, since the scaladoc here already points at `FileScanRDD.hasStrictFileReads` as the matching predicate.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/planmerging/FileSourceV2PlanMergingSuite.scala:
##########
@@ -0,0 +1,703 @@
+/*
+ * 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.{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, 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
+
+/**
+ * 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. 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
+ * which read path the plan took before asserting anything about merging.
+ */
+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")
+
+ // 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"
+
+ 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)
+
+ 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`. 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,
+ 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 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}")
+ }
+ }
+
+ 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
+ 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 AQE as given and
+ * 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)
+ if (useV1) assertUsesFileSourceV1(df) else assertUsesFileSourceV2(df)
+ subqueryCounts(df)
+ }
+ }
+ }
+
+ 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: 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: ") {
+ 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.
+ 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: 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: ") {
+ 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}")
+ }
+ }
+ }
+ }
+ }
+
+ 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.
+ assert(scan.output.map(_.name).toSet == Set("a", "b"),
+ s"the merged scan should read the union of both columns; got ${scan.output}")
+ // 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")
+ }
+ }
+ }
+
+ 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}")
+ 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")
+ }
+ }
+ }
+
+ test("SPARK-57205: merge file scans with differing data filters only when dsv2 symmetric " +
+ "filter propagation is on") {
+ withTempPath { dir =>
+ val path = dir.getCanonicalPath
+ writeFlat("parquet", path)
+ 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
+ | (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}")
+ 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")
+ }
+ }
+ }
+ }
+ }
+ }
+ }
+
+ 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)
+ withFileView("parquet", path) {
+ Seq(true, false).foreach { nestedPruning =>
+ withClue(s"nestedSchemaPruning=$nestedPruning: ") {
+ 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 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}")
+ }
+ }
+ }
+ }
+ }
+ }
+ }
+
+ 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)
+ // 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,
+ 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)
+ // 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(
+ """
+ |SELECT
+ | (SELECT sum(a) FROM t1 WHERE c = 1),
+ | (SELECT sum(b) FROM t2 WHERE c = 1)
+ |""".stripMargin)
+
+ // 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" +
+ 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)),
+ "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")
+ }
+ }
+ }
+ }
+
+ 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
+
+ def withData(lines: Seq[String])(f: String => Unit): Unit =
+ withTempPath { dir =>
+ val path = dir.getCanonicalPath
+ lines.toDS().write.text(path)
+ f(path)
+ }
+
+ def rows(
+ format: String,
+ path: String,
+ schema: String,
+ mode: String,
+ query: String,
+ useV1: Boolean): 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),
+ 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))),
Review Comment:
**Finding 14.** This test's V1 arms pin the behaviour #58411 removes, so whichever of the two lands second breaks.
Measured rather than reasoned: I copied this file and `V2ScanMergingTestHelper.scala` into a worktree at #58411's head and ran this one test.
```
- SPARK-57205: CSV and JSON decline to merge, so their parsing stays per subquery *** FAILED ***
format=csv: (2, 0) did not equal (1, 1) unexpected subquery counts on V1:
```
It aborts on this line, so the rest never report, but they all move with it: `Row(8, 80)` becomes `Row(10, 80)`, `Row(1, 80)` becomes `Row(0, 80)`, and the FAILFAST `intercept[SparkException]` finds nothing to catch. Six V1 expectations plus the throw.
There is a second casualty in the description. Gap 4 opens with "CSV and JSON do not merge, while V1 does" and rests on "V1 has merged both for years and answers `[8, 80]`". After #58411 both paths decline and the gap is closed, not open.
The order is yours to pick, but it is worth stating in the description either way:
- #58411 first - then write these arms against the new V1 behaviour from the start, and gap 4 becomes "was a V1 bug, fixed by SPARK-59107".
- this first - then #58411 has to come back and change them.
Either way the V1 arms stop being a V1/V2 contrast once both are in, so they are better dropped than inverted; the V2 assertions carry the whole point of the test on their own.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57205][SQL] Declare SCAN_MERGING on the built-in file sources [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on code in PR #58340:
URL: https://github.com/apache/spark/pull/58340#discussion_r3881910283
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/FileTable.scala:
##########
@@ -111,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
Review Comment:
**Finding 8.** The criterion holds only while the read is strict.
Your reply on @dongjoon-hyun's thread justifies the four overrides with "a corrupt column chunk, or `datetimeRebaseModeInRead=EXCEPTION` meeting an ancient value, makes any format throw on a column the narrower scan had pruned". That is true only with `spark.sql.files.ignoreCorruptFiles=false`. With it on, `FilePartitionReader.next` catches any `RuntimeException` or `IOException` and returns `false`, dropping the rest of that file (`FilePartitionReader.scala:74`, via `DataSourceUtils.shouldIgnoreCorruptFileException`). So the merged scan returns fewer rows than the narrow scan did. That is the "silent and unrecoverable change to row membership" you gave as what separates CSV and JSON, reached by a format that declares the capability.
Measured on this head, parquet, `spark.sql.files.ignoreCorruptFiles=true`. `b` is written as a string and read as a long, so the reader throws only when it actually reads `b`; a corrupt column chunk is the same code path, just harder to build in a test.
```scala
spark.range(0, 10).selectExpr("id AS a", "cast(id AS string) AS b").write.parquet(path)
spark.read.schema("a long, b long").parquet(path).createOrReplaceTempView("t")
sql("SELECT (SELECT sum(a) FROM t), (SELECT count(b) FROM t)")
```
| path | default | `MergeSubplans` excluded |
|---|---|---|
| V2 | `[null, 0]` | `[45, 0]` |
| V1 | `[null, 0]` | `[45, 0]` |
`sum(a)` touches only healthy data and is correct at 45 today; after the merge it is `null`. V1 is the same, so this is V1 parity rather than a V1/V2 split - but it is the parity you declined to copy for CSV and JSON, on the same reasoning. ORC did not reproduce with this particular type mismatch; the exposure is any per-column read failure, so it is not parquet-specific.
The fix has a precedent in the tree: `InMemoryRelation` refuses to treat a file scan as repeatable under either flag (`InMemoryRelation.scala:378-398`, and `FileScanRDD.hasStrictFileReads`).
```scala
override def capabilities: java.util.Set[TableCapability] =
if (supportsScanMerging && hasStrictFileReads) {
FileTable.CAPABILITIES_WITH_SCAN_MERGING
} else {
FileTable.CAPABILITIES
}
/**
* Whether a read of this table is strict. A best-effort read is not reproducible: a failure on a
* column only the other scan projects is swallowed, so the merged scan would not read a superset
* of either input's rows.
*/
private def hasStrictFileReads: Boolean = {
val fileSourceOptions = new FileSourceOptions(options.asCaseSensitiveMap.asScala.toMap)
!fileSourceOptions.ignoreCorruptFiles && !fileSourceOptions.ignoreMissingFiles
}
```
`FileSourceOptions` is already imported here and it resolves the per-read option over the session conf, so both spellings are covered. Only `ignoreCorruptFiles` has the failure above - a missing file drops the same rows whatever is projected - so including `ignoreMissingFiles` is just matching the existing predicate; drop it if you prefer the narrower gate.
If you would rather keep the capability unconditional, then the migration-guide entry has to come back for this shape. Unlike the errors the criterion accepts, this one is a silent result change.
##########
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
Review Comment:
**Finding 9.** The first sentence is not what `AvroPartitionReaderFactory` does.
The deserializer is built from `readDataSchema`, i.e. the projection, not the full schema (`AvroPartitionReaderFactory.scala:101-111`). `positionalFieldMatching` is passed straight through, and `AvroUtils.AvroSchemaHelper.getAvroField` then resolves a catalyst field by its *position in that projected schema* (`AvroUtils.scala:463-469`). So the projection is exactly what decides which Avro field feeds which column, and widening it re-maps them.
Measured on this head. File fields `a`, `b`, `c` holding `id`, `100 * id`, `10000 * id` for ids 0 to 4, read with `positionalFieldMatching=true`, query `SELECT (SELECT sum(a) FROM t), (SELECT sum(c) FROM t)`:
| path | default | `MergeSubplans` excluded |
|---|---|---|
| V2 | `[10, 1000]` | `[10, 10]` |
| V1 | `[10, 1000]` | `[10, 10]` |
`sum(c)` is 100000. Both values are wrong, because Avro positional matching against a pruned schema is already broken without this PR: the `c`-only scan resolves position 0 and reads `a`. Merging cannot turn a correct answer into a wrong one here either, since the union of two prefixes is the longer prefix. So this is not a correctness regression, and I am not asking you to exclude Avro.
Two smaller asks. State the mechanism as it is - what makes merging safe for Avro is that the format has no record-level parse verdict, not that decoding ignores the projection - so a later reader does not build on a property the reader does not have. And "No query result changes" in the description needs a qualifier, or the pruning bug needs its own JIRA to point at.
@dongjoon-hyun named `positionalFieldMatching` as the Avro-specific risk on the coverage thread above; the new `AvroV2Suite` test does not reach it.
##########
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
Review Comment:
**Finding 11.** This override widens the seam from `protected` to public.
`FileTable.supportsScanMerging` is `protected`, and an override with no access modifier is public in Scala. So `ParquetTable`, `OrcTable`, `TextTable` and `AvroTable` each expose it as public API while `CSVTable` and `JsonTable` keep it protected. Either add the modifier to the four, or drop it from the base if the seam is meant to be part of the format-author surface.
```suggestion
override protected def supportsScanMerging: Boolean = true
```
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57205][SQL] Declare SCAN_MERGING on the built-in file sources [spark]
Posted by "LuciferYang (via GitHub)" <gi...@apache.org>.
LuciferYang commented on PR #58340:
URL: https://github.com/apache/spark/pull/58340#issuecomment-5488597808
Thank you @peter-toth @dongjoon-hyun
Since 4.3 has entered the RC phase, this pr has been merged only to `master` and `branch‑4.x`. @dongjoon-hyun If you have any new review comments, feel free to leave them, and I will address them in a follow‑up PR.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57205][SQL] Declare SCAN_MERGING on the built-in file sources [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on code in PR #58340:
URL: https://github.com/apache/spark/pull/58340#discussion_r3879483408
##########
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:
**Finding 6.** @LuciferYang, answering your question directly: no, I would not put a qualifier on "a superset of their rows".
That clause is not decoration, it is the soundness argument. The merge is only a rewrite because a projection and a filter above the merged scan can recover each original scan's result, and that needs the merged scan to still contain every row the narrow scan had. A source that can lose rows when a column joins the projection does not have the property. A qualifier admitting that would mean "declaring this capability permits Spark to change your results", and nothing downstream could rely on the premise again - which is the risk @dongjoon-hyun's comment is about.
Two facts have moved since you decided to keep CSV and JSON in, both measured on this head, both in my reply on the migration-guide thread: `PERMISSIVE` also changes (with `columnNameOfCorruptRecord` in the schema, `count(_corrupt_record)` goes 0 to 1 on csv and json), and `FAILFAST` also changes (a short CSV row makes the merged query throw where the unmerged one returned rows). So this is not one unusual mode. It reaches the default mode, and it can turn a working query into an error.
So I would take @dongjoon-hyun's shape, and one step further:
- `FileTable.CAPABILITIES` goes back to `EnumSet.of(BATCH_READ, BATCH_WRITE)`, with a `CAPABILITIES_WITH_SCAN_MERGING` beside it and a `protected def supportsScanMerging: Boolean = false` seam choosing between them (see my comment on the new scaladoc).
- `ParquetTable`, `OrcTable`, `TextTable` and `AvroTable` override it to `true`. `CSVTable` and `JsonTable` do not.
What that buys: the javadoc stays true of every source that declares the capability; no third-party `FileTable` subclass is opted into a contract its author never saw; and the migration-guide entry can go away, because no user-visible result change is left.
On your objection that a seventh format would then silently not participate - I think that is the right way round. A new format that does not merge is a missed optimization, visible as a slower query. A new format that merges while its parser is projection-sensitive is a wrong answer, visible as nothing. The default should fail in the cheap direction, and the four `supportsScanMerging = true` overrides sit next to `formatName` and `fallbackFileFormat`, which a new format has to fill in anyway.
The honest counter-argument, because it is a real one: this leaves V2 CSV and JSON merging less than V1 for column-only-differing scans, part of the gap this PR set out to close. My read is that V1's behaviour here is a bug rather than a target, since it makes one subquery's result depend on what a sibling subquery projects, and matching a bug is worse than a fourth documented gap. If you would rather fix V1 first and align V2 to it after, as you offered above, that order works for me too. What I would avoid is weakening the capability's contract so that the current shape becomes in-spec.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57205][SQL] Declare SCAN_MERGING on the built-in file sources [spark]
Posted by "LuciferYang (via GitHub)" <gi...@apache.org>.
LuciferYang commented on PR #58340:
URL: https://github.com/apache/spark/pull/58340#issuecomment-5461954729
Thanks @peter-toth. Finding 8 is taken, and it took Avro with it.
**8.** `capabilities` is now `supportsScanMerging && hasStrictFileReads`, with the predicate as you wrote it. I kept `ignoreMissingFiles` in it to match `FileScanRDD.hasStrictFileReads` on the physical side, and the gate is evaluated per call rather than cached, so a table built before the configuration was set still answers for the read that is running. The new test builds the table and the temp view outside the configuration scope for exactly that reason, covers both spellings, and asserts `[45, 0]` alongside two distinct scans.
**9.** The comment claimed the opposite of what the reader does, and it is rewritten around the property Avro does have: no record-level parse verdict. I also withheld the capability under `positionalFieldMatching`, so the contract holds for every source that declares it rather than only where a merge cannot make the answer worse. The option is read off the options map rather than through `AvroOptions`, whose constructor resolves `avroSchemaUrl` and would do I/O inside `capabilities`. The `AvroV2Suite` test now asserts the capability, the withholding under the option, and a projection-only merge. The underlying bug is SPARK-59108, with a runnable repro on the ticket, and the gate can go once it is fixed. ORC turned out to need no equivalent: `OrcUtils.requestedColumnIds` maps both positional cases through `dataSchema.fieldIndex(name)` and reports `canPruneCols = false`. On "No query result changes": with both gates in place it holds, so I dropped the migration-guide entry instead
of qualifying the sentence.
**10.** Corrected. Gap 4 now says CSV and JSON lose both shapes, differing projections and equal columns under differing filters, with your reason: `FileScan.equals` compares the normalized data filters. The V1 side is SPARK-59107, which pins four measured shapes, the three parse modes plus `ignoreCorruptFiles`.
**11.** Fixed on all four.
**12.** SPARK-57205's description is rewritten: the seam, the two exclusions, the two gates, the four gaps.
Mutation checks on the current head: flipping the three sql/core overrides to `false` fails 8 of the 15 new tests, and dropping `&& hasStrictFileReads` fails the strictness test at its first capability assertion. CI is green.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57205][SQL] Declare SCAN_MERGING on the built-in file sources [spark]
Posted by "LuciferYang (via GitHub)" <gi...@apache.org>.
LuciferYang commented on PR #58340:
URL: https://github.com/apache/spark/pull/58340#issuecomment-5474356468
Thanks, @peter-toth. All four are in at `b6c4b7b748b`, and CI is green on it.
**13.** In this PR rather than as a follow-up. I took your clause and added the second disqualifier from `FileTable`'s scaladoc, so the javadoc now names both reasons a built-in source is held back here: a parser that decides what counts as a malformed record from the set of columns it was asked for, and a source that resolves a column by its position in the projection. Those two are the whole gap between the two texts, and the capability is unreleased, so it is cheaper here than after.
**14.** Dropped rather than inverted, as you suggested. `rows` loses its `useV1` parameter, the three V1 expectations and the FAILFAST `intercept` go with it, and the comments now say what a merged scan would do instead of what V1 does. Gap 4 is rewritten around CSV and JSON getting no merging at all being deliberate, with the `[8, 80]` numbers attributed to the V1 path and to SPARK-59107 fixing them there, after which both paths decline these shapes alike. The order is in all three descriptions now: #58411, then this, then #58409.
One correction to the finding. "Every `useV1 = true` expectation here" does not hold for the two parquet ones: this suite pins `ignoreCorruptFiles` and `ignoreMissingFiles` to false in `sparkConf`, and parquet's parser does not depend on the projection, so #58411's gate never fires for them. Measured the way you did it, with this suite and the helper copied into a worktree at #58411's head: the CSV and JSON test passes there, and so does "V1 merges differing partition filters, V2 does not". So only the csv and json arms had to go, and the V1/V2 parity pair stays.
**15.** Split in two. Dropping `&& hasStrictFileReads` now fails both, the capability one at its first assertion and the rows one with correct answer `[45, 0]` against Spark answer `[null, 0]`, which is the number your scratch suite reported, now in a test that can report it. The three-overrides mutation is 8 of 16, and the eight that pass are six declines, the partition-filter test whose V2 side declines under the mutation too, and the new rows test, whose separate scans the mutation also produces.
**16.** In the description, since `FileSourceOptions.hasStrictFileReads` is not in master yet. Whichever way the two land, this ends up calling it.
Two more measurements on the sequencing, since #58409 belongs to the same set. Its five commits cherry-picked onto this head leave the `AvroV2Suite` case passing, so this PR and #58409 do not interact beyond the gate. And `git merge-tree` puts the only textual conflict among the three in `connector/avro/src/test/scala/org/apache/spark/sql/avro/AvroSuite.scala`, between this one and #58411, both of which append a test there, so whichever lands second needs a rebase.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57205][SQL] Declare SCAN_MERGING on the built-in file sources [spark]
Posted by "LuciferYang (via GitHub)" <gi...@apache.org>.
LuciferYang commented on PR #58340:
URL: https://github.com/apache/spark/pull/58340#issuecomment-5476267893
Thanks, @peter-toth. Both are in at `019cda6c095`.
**17.** Moved as you suggest, with the entry keeping its first three sentences plus your line on what the configuration adds. I named the requirement instead of opening with "the last of those conditions", since #58411 inserts the V1 condition into that same paragraph and the reference would point past it. Line 343 is untouched, and `git merge-tree` still auto-merges the doc after the insertion, so the only conflict with #58411 stays `AvroSuite.scala`. Line 331 I left as the section's summary, with the exception in the paragraph that enumerates the conditions; say so if you would rather it carried a pointer.
**18.** Split, `ignoreMissingFiles` now reading "to match the strictness predicate the file reader uses". That agrees with `FileTable.scala:143-145` and with #58411's wording.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57205][SQL] Declare SCAN_MERGING on the built-in file sources [spark]
Posted by "LuciferYang (via GitHub)" <gi...@apache.org>.
LuciferYang commented on code in PR #58340:
URL: https://github.com/apache/spark/pull/58340#discussion_r3880092918
##########
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:
@peter-toth thanks for measuring this, and sorry for the wrong table above. Both shapes reproduce on this head, on V1 as well as V2:
| shape | merged | `MergeSubplans` excluded |
|---|---|---|
| `PERMISSIVE`, `_corrupt_record` in schema, csv and json | `[1, 80]` | `[0, 80]` |
| `FAILFAST`, short CSV row | throws | `[10, 80]` |
My table measured one malformed shape and I generalized it. The `FAILFAST` reasoning I gave only covers malformedness belonging to a column; a token-count mismatch is a property of how many columns are parsed, and that is what the merge changes. With the short row the `a`-only scan parses one column against one token, so nothing is malformed until the merge makes it two.
That removes my objection. I weighed one non-default mode against V1 parity; it is a different trade when the default mode changes results and `FAILFAST` can turn a working query into an error. Your point that the superset clause is the soundness argument rather than decoration is well taken.
Finding 1 is right too: V1 returns `[8, 80]` as well, so falling back to `useV1SourceList` restores nothing. Only excluding `MergeSubplans` does.
I will fix the guide sentence and the same claim in my summary comment, and come back on the shape of the fix including your `supportsScanMerging` seam.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57205][SQL] Declare SCAN_MERGING on the built-in file sources [spark]
Posted by "LuciferYang (via GitHub)" <gi...@apache.org>.
LuciferYang commented on PR #58340:
URL: https://github.com/apache/spark/pull/58340#issuecomment-5488584064
**Merge Summary:**
- merged into master https://github.com/apache/spark/commit/484866bd80dcc153202103572f4c96ef6240ad87
- merged into branch-4.x https://github.com/apache/spark/commit/69e1cb10f43554035c6a8d13239dd7512e0b3a2f
*Posted by `merge_spark_pr.py`*
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57205][SQL] Declare SCAN_MERGING on the built-in file sources [spark]
Posted by "LuciferYang (via GitHub)" <gi...@apache.org>.
LuciferYang closed pull request #58340: [SPARK-57205][SQL] Declare SCAN_MERGING on the built-in file sources
URL: https://github.com/apache/spark/pull/58340
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57205][SQL] Declare SCAN_MERGING on the built-in file sources [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on PR #58340:
URL: https://github.com/apache/spark/pull/58340#issuecomment-5461061330
> To @peter-toth , if you don't mind, could you ask your agent to exclude the pre-existing comments or avoid to mention the other GitHub Account IDs more than once? It's because it seems that I received too many pings from your comments while you are talking with others (not me). And, please use direct hyperlinks to the comments instead of referring like `someone's thread` because there are multiple threads in the single PR.
Sorry @dongjoon-hyun for the pings and thanks for the feedback. I will adjust my review skill according that.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57205][SQL] Declare SCAN_MERGING on the built-in file sources [spark]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57205][SQL] Declare SCAN_MERGING on the built-in file sources [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on code in PR #58340:
URL: https://github.com/apache/spark/pull/58340#discussion_r3892890241
##########
docs/sql-performance-tuning.md:
##########
@@ -384,7 +384,7 @@ In TPC-DS benchmark runs, enabling symmetric filter propagation made `q9` and `q
<td><code>spark.sql.optimizer.mergeSubplans.filterPropagation.dsv2SymmetricFilterPropagation.enabled</code></td>
<td>false</td>
<td>
- 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 <code>spark.sql.optimizer.mergeSubplans.filterPropagation.symmetricFilterPropagation.enabled</code> 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 <code>Filter</code> re-checks the rest above the scan. This applies only to V2 sources that opt in to scan merging with the <code>SCAN_MERGING</code> 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 <code>spark.sql.optimizer.mergeSubplans.filterPropagation.symmetricFilterPropagation.enabled</code> 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 <code>Filter</code> re-checks the rest above the scan. This applies only to V2 sources that opt in to scan merging with the <code>SCAN_MERGING</code> 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 <code>spark.sql.sources.useV1SourceList</code>; 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 <code>spa
rk.sql.files.ignoreCorruptFiles</code> or <code>spark.sql.files.ignoreMissingFiles</code> 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 <code>positionalFieldMatching</code>, which resolves a column by its position in the projection.
Review Comment:
**Finding 17.** This paragraph is the only place the new behaviour is written down, and it sits in the cell of a configuration that is `false` by default. That configuration only adds the differing-data-filter shape. The projection-only merge and the two withholding rules hold whatever it is set to.
Meanwhile the prose a reader consults for when merging fires still states the old rule.
`docs/sql-performance-tuning.md:343`:
> Two subplans are merged when their plans match node by node: ... and the leaves must read the same input.
Line 331 opens the section the same way ("subplans that return a single row and read the same input"). Both were accurate while this cell ended with "no built-in source does". After this PR two leaves that read different column sets merge, under the default configuration, for Parquet, ORC, text and Avro on the V2 read path.
Suggest moving the format list and the withholding rules up into the prose, after line 343:
```markdown
On the DataSource V2 read path a source can go further and declare the `SCAN_MERGING` table capability, which lets 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.
```
and leaving this cell its first three sentences plus one line for what the configuration itself adds:
```html
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.
```
The `ignoreMissingFiles` wording above is finding 18's fix, folded in.
##########
docs/sql-performance-tuning.md:
##########
@@ -384,7 +384,7 @@ In TPC-DS benchmark runs, enabling symmetric filter propagation made `q9` and `q
<td><code>spark.sql.optimizer.mergeSubplans.filterPropagation.dsv2SymmetricFilterPropagation.enabled</code></td>
<td>false</td>
<td>
- 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 <code>spark.sql.optimizer.mergeSubplans.filterPropagation.symmetricFilterPropagation.enabled</code> 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 <code>Filter</code> re-checks the rest above the scan. This applies only to V2 sources that opt in to scan merging with the <code>SCAN_MERGING</code> 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 <code>spark.sql.optimizer.mergeSubplans.filterPropagation.symmetricFilterPropagation.enabled</code> 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 <code>Filter</code> re-checks the rest above the scan. This applies only to V2 sources that opt in to scan merging with the <code>SCAN_MERGING</code> 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 <code>spark.sql.sources.useV1SourceList</code>; 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 <code>spa
rk.sql.files.ignoreCorruptFiles</code> or <code>spark.sql.files.ignoreMissingFiles</code> 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 <code>positionalFieldMatching</code>, which resolves a column by its position in the projection.
Review Comment:
**Finding 18.** The "because" covers both flags as written, and only one of them does that.
> 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
`ignoreMissingFiles` swallows a `FileNotFoundException` (`sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/FilePartitionReader.scala:51`), which is file-level: the same rows go whatever either side projects. Your own scaladoc says it, `FileTable.scala:143-145`:
> `ignoreMissingFiles` drops the same rows whatever is projected, and is included to match `FileScanRDD.hasStrictFileReads`, the same predicate on the physical side.
So the reason belongs to `ignoreCorruptFiles`, and the second flag needs its own half-clause - "and when `spark.sql.files.ignoreMissingFiles` is true, to match the strictness predicate the file reader uses". Not reopening whether the flag belongs in the predicate; that is settled.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57205][SQL] Declare SCAN_MERGING on the built-in file sources [spark]
Posted by "LuciferYang (via GitHub)" <gi...@apache.org>.
LuciferYang commented on PR #58340:
URL: https://github.com/apache/spark/pull/58340#issuecomment-5476556912
@dongjoon-hyun Could you take another look when you get a chance? Or should I merge this PR first?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org