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: [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]
