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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
