This is an automated email from the ASF dual-hosted git repository.
LuciferYang pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new 69e1cb10f435 [SPARK-57205][SQL] Declare SCAN_MERGING on the built-in
file sources
69e1cb10f435 is described below
commit 69e1cb10f43554035c6a8d13239dd7512e0b3a2f
Author: YangJie <[email protected]>
AuthorDate: Tue Sep 1 11:44:33 2026 +0800
[SPARK-57205][SQL] Declare SCAN_MERGING on the built-in file sources
### What changes were proposed in this pull request?
`FileTable` gains a `supportsScanMerging` seam, and `ParquetTable`,
`OrcTable`, `TextTable` and `AvroTable` override it. Those four then take part
in the DSv2 scan merging added by #57360 (SPARK-40259): `PlanMerger` drives
`V2ScanRelationPushDown.rebuildScan` to rebuild a merged scan, and a source
supplies no merge logic of its own. It only declares that widening the set of
columns pruned on its builder, with the scan options and pushed filters held
constant, changes neither which row [...]
`CSVTable` and `JsonTable` do not override it. Their parsers are handed the
columns the scan asked for and decide from that set what counts as a malformed
record, so a merged scan reading the union of two column sets can drop or
rewrite rows the narrower scan returned. Measured, csv and json alike: with
`mode=DROPMALFORMED` and a record malformed only in the other subquery's
column, `sum(a)` is 8 where two separate scans give 10; with `PERMISSIVE`, the
default, and `_corrupt_record` i [...]
Two further gates keep the contract true of the formats that do declare it.
`FileTable` withholds the capability when
`spark.sql.files.ignoreCorruptFiles` or `spark.sql.files.ignoreMissingFiles` is
set, matching `FileScanRDD.hasStrictFileReads` on the physical side. Under a
non-strict read a failure in a column that only the sibling subquery projects
is swallowed and the rest of that file's rows go with it, so the merged scan
returns fewer rows than the narrow one did. Measured on parquet, V1 and V2
alike: `SELECT (SELECT sum(a) FROM t), (SELECT count(b) FROM [...]
`AvroTable` withholds it under `positionalFieldMatching`.
`AvroPartitionReaderFactory` builds the deserializer from the pruned read
schema while the Avro side stays the full Avro schema, so under that option
catalyst field *i* of the projection takes Avro field *i* of that schema, and
widening the projection changes the values a column comes back with. That is a
bug in its own right, filed as SPARK-59108 (#58409), which lands after this one
and deletes this gate with it. ORC needs no [...]
The rest is a new test suite, a package-private `V2ScanMergingTestHelper`
shared with `DSv2PlanMergingSuite`, one test in `AvroV2Suite`, and one
documentation update.
Landing order: #58411 (SPARK-59107) first, then this one, then #58409
(SPARK-59108), which deletes the `AvroTable` gate. Nothing here depends on the
first step, since this PR no longer asserts anything about how V1 merges CSV
and JSON. Once #58411 is in, `FileTable.hasStrictFileReads` should call the
`FileSourceOptions.hasStrictFileReads` it adds rather than spell the predicate
out a third time, beside `FileScanRDD` and the cache-repeatability check in
`InMemoryRelation`.
### 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, so two subquery scans over the same path are
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 it does fire [...]
On TPC-DS at scale factor 100, with `spark.sql.sources.useV1SourceList`
cleared, two queries change: q9 goes from 16 distinct scans to 6 under the
default configuration and to 2 with `dsv2SymmetricFilterPropagation` on, and
q28 from 6 to 1 with that configuration on. Of the 99 v1.4 queries, 95 ran and
the other 93 of those are unchanged. Wall clock for q9 went from 78.1s to 40.3s
by default and to 21.6s with the configuration on, and q28 from 61.3s to 34.4s;
measured on `local[1]` wit [...]
### 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 only: subqueries over the same Parquet, ORC, text or Avro table
that differ only in their projected columns now collapse into a single scan
reading the union of those columns. No query result changes, which is what the
exclusions and the two gates above are for, and why this needs no
migration-guide entry. The `dsv2SymmetricFilterPropagation` entry in
`docs/sql-performance-tuning.md` said "no built-in source does", which this PR
falsifies. The four formats and the two withh [...]
Four gaps are left in place. The first three are follow-ups #57360 already
lists.
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.
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. V1 does not merge
this shape either, because `SchemaPruning` rewrites each side's `dataSchema`
and the two relations stop being canonically equal.
3. Differing data filters need
`spark.sql.optimizer.mergeSubplans.filterPropagation.dsv2SymmetricFilterPropagation.enabled`,
which defaults to false. On the V2 path that configuration alone is enough,
where V1 needs the broader `symmetricFilterPropagation`.
4. CSV and JSON get no merging at all, which is deliberate rather than a
follow-up. It costs them two shapes: scans whose projected columns differ, and
scans reading the same columns under different filters, since `FileScan.equals`
compares the normalized data filters so those are not canonically equal either
and do not fall back to plain reuse. The V1 path merges both today and answers
`[8, 80]` where two separate scans answer `[10, 80]`, so one subquery's result
depends on what a si [...]
### How was this patch tested?
New suite `FileSourceV2PlanMergingSuite`, 16 tests. Every built-in file
table is checked on the side it belongs to; a non-strict read is checked to
withhold the capability on both strictness configurations, and separately to
keep its scans separate and return `[45, 0]`, with the table and the temp view
built outside the configuration scope so that a cached gate would fail the
test. Those are two tests rather than one so that a mutation names which half
it broke, the capability asserti [...]
Then V1/V2 parity on three shapes and the partition-filter gap pinned as
the one shape where they disagree, both on parquet, whose reads this suite pins
strict and whose parser does not depend on the projection, so SPARK-59107
leaves them merging on V1. The CSV and JSON parse behaviour is pinned on the V2
side alone, on three shapes: `DROPMALFORMED` gives `[10, 80]`, `PERMISSIVE`
with `_corrupt_record` in the schema gives `[0, 80]`, and `FAILFAST` with a
short CSV row returns rows rat [...]
Every test asserts which read path the plan took 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
configuration alone would quietly run V1 on both sides.
One test in `AvroV2Suite`, because `AvroTable` lives in the module that has
it on the classpath: it asserts the capability, that `positionalFieldMatching`
withholds it, and that two scans differing only in their projected columns fuse
into one reading the union.
Mutation checks. Turning the three sql/core overrides to `false` fails 8 of
the 16 tests; of the 8 that pass, 6 assert a decline, one asserts that V1
merges differing partition filters where V2 does not and V2 declines under the
mutation too, and one asserts that a non-strict read keeps its scans separate,
which the mutation also produces. Dropping `&& hasStrictFileReads` from
`capabilities` fails both strictness tests, the capability one at its first
assertion and the rows one with ` [...]
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
Closes #58340 from LuciferYang/SPARK-57205.
Authored-by: YangJie <[email protected]>
Signed-off-by: yangjie01 <[email protected]>
(cherry picked from commit 484866bd80dcc153202103572f4c96ef6240ad87)
Signed-off-by: yangjie01 <[email protected]>
---
.../org/apache/spark/sql/v2/avro/AvroTable.scala | 15 +-
.../org/apache/spark/sql/avro/AvroSuite.scala | 47 +-
docs/sql-performance-tuning.md | 4 +-
.../sql/connector/catalog/TableCapability.java | 6 +
.../sql/execution/datasources/v2/FileTable.scala | 45 +-
.../execution/datasources/v2/orc/OrcTable.scala | 6 +
.../datasources/v2/parquet/ParquetTable.scala | 6 +
.../execution/datasources/v2/text/TextTable.scala | 4 +
.../planmerging/DSv2PlanMergingSuite.scala | 9 +-
.../planmerging/FileSourceV2PlanMergingSuite.scala | 697 +++++++++++++++++++++
.../planmerging/V2ScanMergingTestHelper.scala | 41 ++
11 files changed, 869 insertions(+), 11 deletions(-)
diff --git
a/connector/avro/src/main/scala/org/apache/spark/sql/v2/avro/AvroTable.scala
b/connector/avro/src/main/scala/org/apache/spark/sql/v2/avro/AvroTable.scala
index abcea9a2a238..7b6db3c16c5b 100644
--- a/connector/avro/src/main/scala/org/apache/spark/sql/v2/avro/AvroTable.scala
+++ b/connector/avro/src/main/scala/org/apache/spark/sql/v2/avro/AvroTable.scala
@@ -21,7 +21,7 @@ import scala.jdk.CollectionConverters._
import org.apache.hadoop.fs.FileStatus
import org.apache.spark.sql.SparkSession
-import org.apache.spark.sql.avro.AvroUtils
+import org.apache.spark.sql.avro.{AvroOptions, AvroUtils}
import org.apache.spark.sql.connector.write.{LogicalWriteInfo, Write,
WriteBuilder}
import org.apache.spark.sql.execution.datasources.FileFormat
import org.apache.spark.sql.execution.datasources.v2.FileTable
@@ -52,4 +52,17 @@ case class AvroTable(
override def supportsDataType(dataType: DataType): Boolean =
AvroUtils.supportsDataType(dataType)
override def formatName: String = "Avro"
+
+ // Avro has no record-level parse verdict: a record is either decodable or
the read fails, and
+ // there is no mode that drops or rewrites a record based on the columns
asked for. The `mode`
+ // option in AvroOptions is read by from_avro and schema_of_avro, not by
this scan.
+ //
+ // `positionalFieldMatching` is the exception. AvroPartitionReaderFactory
builds the deserializer
+ // from the pruned read schema while the Avro side stays the full Avro
schema, so under that
+ // option catalyst field i of the projection takes Avro field i of that
schema, and widening the
+ // projection changes the values a column comes back with. Read the option
off the map rather than
+ // through AvroOptions, whose constructor resolves `avroSchemaUrl` and would
do I/O here, and read
+ // it leniently so a malformed value still fails where Avro reports it
rather than here.
+ override protected def supportsScanMerging: Boolean =
+
!"true".equalsIgnoreCase(options.get(AvroOptions.POSITIONAL_FIELD_MATCHING))
}
diff --git
a/connector/avro/src/test/scala/org/apache/spark/sql/avro/AvroSuite.scala
b/connector/avro/src/test/scala/org/apache/spark/sql/avro/AvroSuite.scala
index 813401773834..f4e27dd7f038 100644
--- a/connector/avro/src/test/scala/org/apache/spark/sql/avro/AvroSuite.scala
+++ b/connector/avro/src/test/scala/org/apache/spark/sql/avro/AvroSuite.scala
@@ -42,9 +42,10 @@ import
org.apache.spark.sql.catalyst.expressions.AttributeReference
import org.apache.spark.sql.catalyst.plans.logical.Filter
import org.apache.spark.sql.catalyst.util.DateTimeTestUtils
import
org.apache.spark.sql.catalyst.util.DateTimeTestUtils.{withDefaultTimeZone, LA,
UTC}
+import org.apache.spark.sql.connector.catalog.TableCapability
import org.apache.spark.sql.execution.{FormattedMode, SparkPlan}
import org.apache.spark.sql.execution.datasources.{CommonFileDataSourceSuite,
DataSource, FilePartition}
-import org.apache.spark.sql.execution.datasources.v2.{BatchScanExec,
FileDataSourceV2, FileTable}
+import org.apache.spark.sql.execution.datasources.v2.{BatchScanExec,
DataSourceV2ScanRelation, FileDataSourceV2, FileTable}
import org.apache.spark.sql.functions._
import org.apache.spark.sql.internal.LegacyBehaviorPolicy
import org.apache.spark.sql.internal.LegacyBehaviorPolicy._
@@ -3937,6 +3938,50 @@ class AvroV2Suite extends AvroSuite with
ExplainSuiteHelper {
s"V2 formatName '${v2Table.formatName}' != V1 toString
'${v1Format.toString}'")
}
+ test("SPARK-57205: Avro V2 declares SCAN_MERGING and merges scans differing
only in columns") {
+ // AvroTable withholds the capability under positionalFieldMatching,
because the deserializer is
+ // built from the pruned read schema while the Avro side stays unpruned,
so catalyst field i of
+ // the projection takes Avro field i of that schema and widening the
projection shifts values.
+ // FileTable also withholds it when the reads are not strict, so pin that
rather than inherit.
+ withSQLConf(
+ SQLConf.IGNORE_CORRUPT_FILES.key -> "false",
+ SQLConf.IGNORE_MISSING_FILES.key -> "false") {
+ val v2Provider = DataSource.lookupDataSourceV2("avro",
spark.sessionState.conf)
+ assert(v2Provider.isDefined)
+ val dsV2 = v2Provider.get.asInstanceOf[FileDataSourceV2]
+ val v2Table = dsV2.getTable(
+ new StructType(), Array.empty, JCollections.emptyMap[String, String]())
+ assert(v2Table.capabilities().contains(TableCapability.SCAN_MERGING))
+ val positional = dsV2.getTable(new StructType(), Array.empty,
+ JCollections.singletonMap("positionalFieldMatching", "true"))
+ assert(!positional.capabilities().contains(TableCapability.SCAN_MERGING))
+
+ withTempPath { dir =>
+ val path = dir.getCanonicalPath
+ spark.range(0, 20).selectExpr("id AS a", "id * 2 AS b", "id % 3 AS c")
+ .write.format("avro").save(path)
+ withTempView("avro_scan_merging") {
+
spark.read.format("avro").load(path).createOrReplaceTempView("avro_scan_merging")
+ val df = sql(
+ """
+ |SELECT
+ | (SELECT sum(a) FROM avro_scan_merging WHERE c = 1),
+ | (SELECT sum(b) FROM avro_scan_merging WHERE c = 1)
+ |""".stripMargin)
+ checkAnswer(df, Row(70, 140))
+ val scans = df.queryExecution.optimizedPlan.collectWithSubqueries {
+ case s: DataSourceV2ScanRelation => s
+ }
+ assert(scans.map(_.canonicalized).distinct.length == 1,
+ s"the two Avro scans should be fused into
one:\n${df.queryExecution.optimizedPlan}")
+ // c is read because the filter stays above the merged scan.
+ assert(scans.head.output.map(_.name).toSet == Set("a", "b", "c"),
+ s"the merged scan should read the union of both columns; got
${scans.head.output}")
+ }
+ }
+ }
+ }
+
test("Geospatial types are not supported in Avro") {
withTempDir { dir =>
// Temporary directory for writing the test data.
diff --git a/docs/sql-performance-tuning.md b/docs/sql-performance-tuning.md
index fee8801086c4..eff53e60d8c1 100644
--- a/docs/sql-performance-tuning.md
+++ b/docs/sql-performance-tuning.md
@@ -342,6 +342,8 @@ They are merged into one aggregate that computes `min` and
`max` together, so `s
Two subplans are merged when their plans match node by node: `Project` lists
are unioned, `Aggregate`s must have the same grouping and use the same
aggregation implementation (so a `min` is not merged with a `collect_list`),
`Filter`s must have the same condition, `Join`s must have the same type,
condition and hints, and the leaves must read the same input. Subplans that
differ only in their `WHERE` conditions can be merged as well, by turning each
side's condition into a boolean column [...]
+On the DataSource V2 read path the requirement that the leaves read the same
input is relaxed for a source that declares the `SCAN_MERGING` table
capability: two leaves that differ only in their projected columns merge into a
single scan reading the union of those columns. Among the built-in file formats
Parquet, ORC, text and Avro declare it; a format reaches its V2 read path only
when it is removed from `spark.sql.sources.useV1SourceList`. A file table
withholds the capability when `sp [...]
+
When only one of the two subplans has a filter, merging is always beneficial,
because the unfiltered side reads all the data anyway. This case is on by
default, unless the filter has to cross a `Join` to reach the aggregate, which
needs the through-join configuration below. When both sides have a filter (the
symmetric case), the merged scan filter becomes `OR(f1, f2)`, which is less
selective than either original filter and can therefore read more data - for
example when the filters prun [...]
Still, it is worth considering on queries that compute several differently
filtered aggregates over the same table, which is a common analytical shape:
@@ -384,7 +386,7 @@ In TPC-DS benchmark runs, enabling symmetric filter
propagation made `q9` and `q
<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
on [...]
+ 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
on [...]
</td>
<td>4.3.0</td>
</tr>
diff --git
a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/TableCapability.java
b/sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/TableCapability.java
index 599e801f2a07..e98cf7c11a05 100644
---
a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/TableCapability.java
+++
b/sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/TableCapability.java
@@ -139,6 +139,12 @@ public enum TableCapability {
* obtaining a fresh {@link org.apache.spark.sql.connector.read.ScanBuilder}
with the same options
* and re-applying the same pushed filters and pruned columns yields an
equivalent scan.
* <p>
+ * Determinism alone is not enough: widening the set of pruned columns, with
the options and
+ * pushed filters held constant, must not change which rows the scan returns
nor the values it
+ * returns for the columns already asked for. It may at most surface a read
error. A source whose
+ * parser decides what counts as a malformed record from the set of columns
it was asked for does
+ * not meet this, and neither does one that resolves a column by its
position in the projection.
+ * <p>
* Given that contract, Spark builds the merged scan itself: it prunes a
fresh ScanBuilder to the
* union of both read schemas, re-pushes the (possibly OR-widened) filters,
and builds. The merged
* scan reads the union of the two scans' columns and a superset of their
rows; each original
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/FileTable.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/FileTable.scala
index 8941a4c8d8c7..41b4a6196d5f 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/FileTable.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/FileTable.scala
@@ -37,6 +37,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
+ * returns for the columns it was already asked for. It may at most surface a
read error. A format
+ * whose parser decides what counts as a malformed record from the set of
columns it was asked for
+ * does not meet that, and neither does one that resolves a column by its
position in the
+ * projection. The capability is also withheld from a table whose reads are
not strict, see
+ * `hasStrictFileReads`.
+ */
abstract class FileTable(
sparkSession: SparkSession,
options: CaseInsensitiveStringMap,
@@ -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 = {
+ val fileSourceOptions = new
FileSourceOptions(options.asCaseSensitiveMap.asScala.toMap)
+ !fileSourceOptions.ignoreCorruptFiles &&
!fileSourceOptions.ignoreMissingFiles
+ }
/**
* When possible, this method should return the schema of the given `files`.
When the format
@@ -182,4 +219,10 @@ abstract class FileTable(
object FileTable {
private val CAPABILITIES = util.EnumSet.of(BATCH_READ, BATCH_WRITE)
+
+ // For the formats that override supportsScanMerging. `fileIndex` is a lazy
val, so every scan
+ // built from one table lists the same files, and `newScanBuilder` returns a
fresh builder over
+ // `mergedOptions(options)`.
+ private val CAPABILITIES_WITH_SCAN_MERGING =
+ util.EnumSet.of(BATCH_READ, BATCH_WRITE, SCAN_MERGING)
}
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/orc/OrcTable.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/orc/OrcTable.scala
index 08cd89fdacc6..b3aaf10ccc2b 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/orc/OrcTable.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/orc/OrcTable.scala
@@ -68,4 +68,10 @@ case class OrcTable(
}
override def formatName: String = "ORC"
+
+ // A row is decoded from the columns the scan asked for, so under strict
file reads reading more
+ // columns can only surface an error, never silently change which rows come
back. When the read is
+ // not strict that error is swallowed and the rest of the file's rows go
with it, which is why
+ // FileTable withholds the capability there.
+ override protected def supportsScanMerging: Boolean = true
}
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/parquet/ParquetTable.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/parquet/ParquetTable.scala
index 67052c201a9d..768e1ce329f5 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/parquet/ParquetTable.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/parquet/ParquetTable.scala
@@ -70,4 +70,10 @@ case class ParquetTable(
}
override def formatName: String = "Parquet"
+
+ // A row is decoded from the column chunks the scan asked for, so under
strict file reads reading
+ // more columns can only surface an error, never silently change which rows
come back. When the
+ // read is not strict that error is swallowed and the rest of the file's
rows go with it, which is
+ // why FileTable withholds the capability there.
+ override protected def supportsScanMerging: Boolean = true
}
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/text/TextTable.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/text/TextTable.scala
index 289c41ca1d0b..5fcb4d3b3af2 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/text/TextTable.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/text/TextTable.scala
@@ -50,4 +50,8 @@ case class TextTable(
dataType.isInstanceOf[StringType]
override def formatName: String = "Text"
+
+ // The schema is a single `value` column, and every line -- or every file
under `wholetext` --
+ // becomes a row, so there is no parse step whose outcome a wider column set
could change.
+ override protected def supportsScanMerging: Boolean = true
}
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/planmerging/DSv2PlanMergingSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/planmerging/DSv2PlanMergingSuite.scala
index 86bbe951ab9b..b426b27acac1 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/execution/planmerging/DSv2PlanMergingSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/planmerging/DSv2PlanMergingSuite.scala
@@ -23,7 +23,7 @@ import org.apache.spark.sql.{DataFrame, QueryTest, Row}
import org.apache.spark.sql.catalyst.plans.physical.KeyedPartitioning
import org.apache.spark.sql.connector.FakeV2ProviderWithCustomSchema
import
org.apache.spark.sql.connector.catalog.{InMemoryScanMergingPartitionFilterCatalog,
InMemoryScanMergingReportingCatalog}
-import org.apache.spark.sql.execution.datasources.v2.{BatchScanExec,
DataSourceV2Relation, DataSourceV2ScanRelation}
+import org.apache.spark.sql.execution.datasources.v2.{BatchScanExec,
DataSourceV2Relation}
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.test.SharedSparkSession
@@ -37,7 +37,7 @@ import org.apache.spark.sql.test.SharedSparkSession
* mis-classify it as non-strict and decline the merge (leaving two scans).
*/
class DSv2PlanMergingSuite extends QueryTest with SharedSparkSession
- with BeforeAndAfter {
+ with BeforeAndAfter with V2ScanMergingTestHelper {
private val v2Source = classOf[FakeV2ProviderWithCustomSchema].getName
private val tbl = "scanmerge.t"
@@ -56,11 +56,6 @@ class DSv2PlanMergingSuite extends QueryTest with
SharedSparkSession
spark.conf.unset("spark.sql.catalog.scanmergereport")
}
- private def v2Scans(df: DataFrame): Seq[DataSourceV2ScanRelation] =
- df.queryExecution.optimizedPlan.collectWithSubqueries {
- case s: DataSourceV2ScanRelation => s
- }
-
// A successful DSv2 merge builds the scan and leaves NO bare
DataSourceV2Relation in the plan.
// A leaked deferred scan (e.g. if a future recursion arm forwarded
`deferredScan` without
// building it) would surface as an unbuilt placeholder relation the read
path cannot plan --
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/planmerging/FileSourceV2PlanMergingSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/planmerging/FileSourceV2PlanMergingSuite.scala
new file mode 100644
index 000000000000..51fb31be8ee3
--- /dev/null
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/planmerging/FileSourceV2PlanMergingSuite.scala
@@ -0,0 +1,697 @@
+/*
+ * 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
+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
+ 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")
+ }
+ }
+ }
+ }
+ }
+ }
+
+ // Separate from the capability test above so that a change breaking the
gate names which half it
+ // broke: an assertion that aborts on the capability never reports on the
rows.
+ test("SPARK-57205: a non-strict read keeps its scans separate") {
+ withTempPath { dir =>
+ val path = dir.getCanonicalPath
+ // b is written as a string and read as a long, so the reader fails only
once it reads b.
+ spark.range(0, 10).selectExpr("id AS a", "cast(id AS string) AS
b").write.parquet(path)
+ // The scans stay separate, so the a-only scan never reads b and sum(a)
is still exact. If
+ // they merged, reading b would fail, ignoreCorruptFiles would swallow
it and drop the rest of
+ // the file, and sum(a) would come back null over rows nothing above the
scan removed. The
+ // 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. Neither table declares SCAN_MERGING, so
each subquery keeps
+ // its own scan. Every shape below is one where a merged scan would return
something else, so
+ // adding the capability back to either table fails this test. The V1 path
does merge them
+ // today, and does return something else, which is SPARK-59107.
+ val typeErrorCsv = Seq("0,0", "1,10", "2,BAD", "3,30", "4,40")
+ val typeErrorJson = Seq(
+ """{"a":0,"b":0}""",
+ """{"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): 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, schema = Some(schema),
+ options = Map("mode" -> mode, "columnNameOfCorruptRecord" ->
"_corrupt_record")) {
+ val df = sql(query)
+ assertUsesFileSourceV2(df)
+ val result = df.collect().toSeq
+ // Asserted after collect() so that AQE has finalized and a merged
subquery's reuse would
+ // be visible in the plan. Pinning this alongside the rows
attributes them to the merge
+ // decision itself.
+ assert(subqueryCounts(df) == ((2, 0)),
+ s"the two subqueries should keep their own
scans:\n${df.queryExecution.executedPlan}")
+ result
+ }
+ }
+
+ Seq("csv" -> typeErrorCsv, "json" -> typeErrorJson).foreach { case
(format, lines) =>
+ withClue(s"format=$format: ") {
+ withData(lines) { path =>
+ // DROPMALFORMED. The a-only scan never parses b, so the record
survives for sum(a). A
+ // merged scan would parse the union and drop it for both, giving 8.
+ assert(rows(format, path, "a long, b long", "DROPMALFORMED",
sumQuery) ==
+ Seq(Row(10, 80)))
+
+ // PERMISSIVE, the default mode, with the corrupt-record column in
the schema. The
+ // subquery that reads a and the corrupt column does not flag the
record; a merged scan
+ // would parse b and populate the column for a row that subquery
counted as clean.
+ assert(rows(format, path, "a long, b long, _corrupt_record string",
"PERMISSIVE",
+ corruptQuery) == Seq(Row(0, 80)))
+ }
+ }
+ }
+
+ // FAILFAST, CSV only: JSON has no arity check, so a missing field is
null, not malformed. Each
+ // narrow scan matches the short row, so the query returns rows; a merged
scan would parse two
+ // columns against a one-token row and throw, turning a working query into
an error.
+ withData(shortRowCsv) { path =>
+ assert(rows("csv", path, "a long, b long", "FAILFAST", sumQuery) ==
Seq(Row(10, 80)))
+ }
+ }
+}
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/planmerging/V2ScanMergingTestHelper.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/planmerging/V2ScanMergingTestHelper.scala
new file mode 100644
index 000000000000..b822d65ac0ad
--- /dev/null
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/planmerging/V2ScanMergingTestHelper.scala
@@ -0,0 +1,41 @@
+/*
+ * 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 is consistent with a
+ * merge and more than one means it was declined. Scans that read the same
columns and carry the
+ * same filters canonicalize equal either way, so use subquery counts for
those.
+ */
+ protected def distinctScans(df: DataFrame): Int =
v2Scans(df).map(_.canonicalized).distinct.length
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
