dongjoon-hyun commented on code in PR #55579:
URL: https://github.com/apache/spark/pull/55579#discussion_r4169230615
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/FileScan.scala:
##########
@@ -182,8 +185,80 @@ trait FileScan extends Scan
FilePartition.getFilePartitions(sparkSession, splitFiles, maxSplitBytes)
}
+ /**
+ * The partitions `partitions` planned, computed once per scan instance.
Both the plain read path
+ * and the runtime-filter path go through this, so a scan node that gets
runtime filters lists the
+ * files once and filters that listing, rather than reading the index twice
and risking two
+ * snapshots. A subclass still customizes `partitions`.
+ */
+ @transient private lazy val plannedPartitions: Seq[FilePartition] =
partitions
+
override def planInputPartitions(): Array[InputPartition] = {
- partitions.toArray
+ plannedPartitions.toArray
+ }
+
+ /**
+ * The partition columns Spark can derive a runtime filter on (SPARK-30628),
restricted to the
+ * ones `readSchema()` still exposes: a reference missing from the scan
relation output fails to
+ * resolve, and a pushed-down aggregate keeps only the partition columns it
groups by.
+ *
+ * A filter over one of them is evaluated against each file's partition
values, so the scan
+ * evaluates it in full and Spark does not evaluate it again after the scan
--
+ * `FileScanBuilder.pushFilters` already keeps compile-time partition
filters out of the post-scan
+ * filters for the same reason.
+ */
+ override def filterAttributes(): Array[NamedReference] = {
+ val readFields = readSchema().fieldNames.map(normalizeName).toSet
+ readPartitionSchema.fieldNames
+ .filter(name => readFields.contains(normalizeName(name)))
+ .map(FieldReference.column)
+ }
+
+ override def fullyPushedFilterAttributes(): Array[NamedReference] =
filterAttributes()
+
+ /**
+ * Plans the files whose partition values satisfy `expressions`, taken from
what `partitions`
+ * planned and bin-packed again for the surviving set.
+ *
+ * Filtering what was planned rather than listing the files again keeps one
listing per scan, and
+ * the files a subclass excluded in `partitions` stay excluded. How the
survivors group into
+ * partitions is decided here again, for their own size. Where each file was
cut into splits is
+ * not: `partitions` cut them for the whole set it listed, before these
filters existed.
+ *
+ * A partition value is fixed within a file, so every row of every partition
returned satisfies
+ * the expressions even though one partition can still hold files from
several partition
+ * directories.
+ */
+ override def planInputPartitionsWithRuntimeFilters(
+ expressions: Array[Expression]): Array[InputPartition] = {
+ val fieldIndex = readPartitionSchema.fieldNames.zipWithIndex
+ .map { case (name, i) => normalizeName(name) -> i }.toMap
+ // Evaluating these is the scan's only chance to apply them, since the
attributes are declared
+ // fully pushed and Spark drops the post-scan filter. A reference outside
the read partition
+ // schema has no value to evaluate against, so fail loudly rather than
return wrong rows. Spark
+ // screens against `filterAttributes()` before it gets here, which is a
subset of that schema.
+ val notApplicable = expressions.filterNot(
+ _.references.forall(a => fieldIndex.contains(normalizeName(a.name))))
+ if (notApplicable.nonEmpty) {
+ throw SparkException.internalError("A file scan can only apply a runtime
filter over its " +
+ s"read partition columns
${readPartitionSchema.fieldNames.mkString("[", ", ", "]")}, " +
+ s"got ${notApplicable.mkString(", ")}")
+ }
+ val bound = expressions.reduce(And).transform {
+ case a: Attribute => BoundReference(fieldIndex(normalizeName(a.name)),
a.dataType, a.nullable)
+ }
+ val predicate = Predicate.createInterpreted(bound)
+ val keptFiles = plannedPartitions
Review Comment:
**This path re-plans from `partitions` only and never consults an overridden
`planInputPartitions()`, so my [round-2
comment](https://github.com/apache/spark/pull/55579#discussion_r4048745686) is
only half addressed.** It said the old path "bypasses `partitions` /
`planInputPartitions()` entirely". An override of `partitions` is now honored,
but `BatchScanExec` still reads the static path through
`batch.planInputPartitions()` and always takes the reader factory from
`batch.createReaderFactory()`, while this method returns plain repacked
`FilePartition`s.
This already happens in the wild. Apache Sedona's
[`ShapefileScan`](https://github.com/apache/sedona/blob/3118f725c7e59ca0aaa534720a79f0e316e9ca6c/spark/common/src/main/scala/org/apache/sedona/sql/datasources/shapefile/ShapefileScan.scala#L79-L124)
extends `FileScan` and overrides `planInputPartitions()` to group each
shapefile's .shp/.shx/.dbf files into a `ShapefilePartition`, and
[`ShapefilePartitionReaderFactory`](https://github.com/apache/sedona/blob/3118f725c7e59ca0aaa534720a79f0e316e9ca6c/spark/common/src/main/scala/org/apache/sedona/sql/datasources/shapefile/ShapefilePartitionReaderFactory.scala#L106-L112)
throws `Unexpected partition type` for anything else. Its `RasterScan` does
the same with `RasterInputPartition`. Sedona [tests hive-partitioned
directories](https://github.com/apache/sedona/blob/3118f725c7e59ca0aaa534720a79f0e316e9ca6c/spark/common/src/test/scala/org/apache/sedona/sql/ShapefileTests.scala#L523),
so with this PR `SELECT * FROM shp WHERE part = (SELECT ma
x(id) FROM t)`, or a DPP join on `part`, drops the `FilterExec`, routes the
filter here, and every task fails. The same queries work today.
The choice has to be made at planning, because the `FilterExec` is already
gone by execution. Could the opt-in be explicit for the built-in formats (or
behind an overridable hook), as the alternative in that comment suggested,
instead of the default for every `FileScan`?
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/PushDownUtils.scala:
##########
@@ -230,48 +231,64 @@ object PushDownUtils extends Logging {
translatedFiltersPushed || partPredicatesPushed
- case catalystScan: SupportsRuntimeCatalystFiltering if
pushableFilters.nonEmpty =>
- // A runtime filter is normally evaluated twice: the source prunes
with it, and the
- // FilterExec above the scan applies it again. The two have to agree,
so this screen
- // pushes only predicates the source can be trusted to evaluate in
Spark's place.
- //
- // But pushing a non-deterministic one would let the source prune on
its own coin flip and
- // Spark flip again for the rows that survive, so we handle that
specially.
- //
- // Not pushing a filter is safe only while its FilterExec still
evaluates it. A fully
- // pushed filter has none, so the source is its only evaluator. That
is why
- // DataSourceV2Strategy deletes a FilterExec at planning time only for
a filter we push
- // below, and we use the same method (isPushablePartitionFilter) to
determine if we
- // should push it.
- //
- // A DPP filter degrades to TrueLiteral once its subquery is pruned
away, so it matches
- // every row. The V2 path above drops these implicitly, since
translateRuntimeFilterV2
- // returns None; here we push Catalyst expressions directly, so we
remove them explicitly.
- val catalystFilters = pushableFilters
- .flatMap(unwrapRuntimeFilterExpression)
- .filterNot(_ == Literal.TrueLiteral)
- if (catalystFilters.nonEmpty) {
- catalystScan.filter(catalystFilters.toArray)
- true
- } else {
- false
- }
-
case _ =>
Review Comment:
**`pushRuntimeFilters` now silently returns `false` for a Catalyst-only
scan, which is every V2 file scan, while `DataSourceV2Strategy` has already
removed the `FilterExec` of a fully pushed scalar-subquery filter.** Before
this PR a Catalyst scan reached its own branch here and was filtered. External
code that still follows the old "push, then
`scan.toBatch.planInputPartitions()`" pattern keeps compiling, reads every
partition, and has nothing left to filter the rows. Apache Gluten's
[`AbstractBatchScanExec`](https://github.com/apache/gluten/blob/39fdc5f584668ad58178714644ba199d37787c2c/shims/spark42/src/main/scala/org/apache/spark/sql/execution/datasources/v2/AbstractBatchScanExec.scala#L69-L75)
shim does exactly that, so a shim derived from it for a Spark with this PR
would return wrong rows for `WHERE part = (SELECT max(...) ...)`. The scaladoc
at L175 ("does not go through this method") is also inaccurate, since
`replanWithRuntimeFilters` routes every Catalyst scan without ru
ntime filters through here.
Could this fail closed instead, e.g. throw an internal error that points to
`replanWithRuntimeFilters` when a `SupportsRuntimeCatalystFiltering` scan
reaches here with pushable filters?
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/FileScan.scala:
##########
@@ -43,6 +45,7 @@ trait FileScan extends Scan
with Batch
with SupportsReportStatistics
with SupportsMetadata
+ with SupportsRuntimeCatalystFiltering
Review Comment:
**Mixing `SupportsRuntimeCatalystFiltering` into the trait turns every
subclass that already implements `SupportsRuntimeFiltering` into a
dual-interface scan, and such a scan now fails every query, not only DPP
queries.** DataSourceV2Strategy.scala:182 evaluates
`relation.runtimeFilterAttrs` for every scan relation, which runs
`checkRuntimeFilteringInterfaces()` and throws `INTERNAL_ERROR`.
The PR description lists this with "No in-tree subclass does", but
out-of-tree ones exist. Apache Kyuubi's Spark Hive connector has
[`HiveScan(...) extends FileScan with
SupportsRuntimeFiltering`](https://github.com/apache/kyuubi/blob/cebcfd7eb4a4d1651defc998c16054cd8fd4f1e4/extensions/spark/kyuubi-spark-connector-hive/src/main/scala/org/apache/kyuubi/spark/connector/hive/read/HiveScan.scala#L42-L52),
and YTsaurus SPYT has [`YtScan ... extends FileScan with
SupportsReportPartitioning with
SupportsRuntimeFiltering`](https://github.com/ytsaurus/ytsaurus-spyt/blob/7c2db2b5808ef3001b5a0ce86bcbd53babe9565c/data-source/src/main/scala/org/apache/spark/sql/v2/YtScan.scala#L44-L46).
On a Spark with this PR, a plain `SELECT * FROM t` through the Kyuubi
connector fails at planning. The subclass cannot opt out of an interface it
inherits from the trait, and dropping its own `SupportsRuntimeFiltering` would
give up its re-listing through `fileIndex.listHiveFiles(partitionFilters ++
runtimeFilt
ers, ...)`.
Could the built-in scans opt in individually instead of the trait, so that
these connectors keep working?
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/connector/SupportsRuntimeCatalystFiltering.scala:
##########
@@ -40,9 +39,15 @@ trait SupportsRuntimeCatalystFiltering extends Scan {
/**
* Returns attributes this scan can be filtered by at runtime.
*
- * Spark will call [[filter]] if it can derive a runtime filter for any of
these attributes.
- * Each reference must resolve against the scan relation output when Spark
builds it. Attributes
- * pruned out of [[Scan.readSchema]] fail to resolve.
+ * Spark will call [[planInputPartitionsWithRuntimeFilters]] if it can
derive a runtime filter
+ * for any of these attributes. Each reference must resolve against the scan
relation output
+ * when Spark builds it. Attributes pruned out of [[Scan.readSchema]] fail
to resolve.
+ *
+ * A dynamic partition pruning filter has no post-scan evaluator at all:
Spark removes it from
+ * the post-scan filters whether or not the attribute is also in
[[fullyPushedFilterAttributes]],
+ * because the join it was derived from already implies it. So an attribute
exposed here must be
+ * one this scan can filter on exactly, not approximately.
[[fullyPushedFilterAttributes]]
Review Comment:
Minor: the conclusion here does not follow from its premise. If "the join it
was derived from already implies it", a scan that keeps extra rows, by pruning
conservatively or not at all, is still correct; only dropping a row that
satisfies the filter is wrong. That is what PushDownUtils.scala:247-248 says
too ("Dropping it costs pruning and not correctness"), and what the in-memory
fixture does: it keeps a partition on evaluation failure, and
`InMemoryCatalystRuntimeFilterTable` declares `part` filterable under
`days(part)` without ever pruning on it. As written, an implementor that can
only prune conservatively, e.g. on a transform or bucket source column or a
nested field, would leave the column out and lose DPP for no correctness gain.
Could this say instead that the scan may return rows that fail the filter,
because Spark re-applies it (by the join, the rewrite, or the `FilterExec`
unless fully pushed), but must never drop a row that satisfies it?
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2FileSourceDPPSuite.scala:
##########
@@ -0,0 +1,658 @@
+/*
+ * 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.connector
+
+import java.io.File
+
+import org.apache.spark.{SparkException, SparkThrowable}
+import org.apache.spark.sql.{DataFrame, QueryTest, Row}
+import org.apache.spark.sql.catalyst.expressions.{AttributeReference,
DynamicPruningExpression, DynamicPruningSubquery, EqualTo, Expression, Literal}
+import org.apache.spark.sql.catalyst.plans.logical.{Filter, LogicalPlan}
+import org.apache.spark.sql.execution.FilterExec
+import org.apache.spark.sql.execution.adaptive.{AdaptiveSparkPlanHelper,
DisableAdaptiveExecutionSuite, EnableAdaptiveExecutionSuite}
+import org.apache.spark.sql.execution.datasources.FilePartition
+import org.apache.spark.sql.execution.datasources.v2.{BatchScanExec, FileScan}
+import org.apache.spark.sql.execution.datasources.v2.parquet.ParquetScan
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.test.SharedSparkSession
+import org.apache.spark.sql.types.{IntegerType, LongType}
+
+/**
+ * Tests for SPARK-30628: subquery partition pruning and DPP for V2 file
sources.
+ *
+ * These tests exercise V2 file scans (e.g. ParquetScan) -- distinct from the
in-memory V2
+ * catalog used by [[org.apache.spark.sql.DynamicPartitionPruningV2Suite]] and
from the V2
+ * iterative-pushdown work covered by
[[DataSourceV2EnhancedRuntimePartitionFilterSuite]].
+ */
+abstract class DataSourceV2FileSourceDPPSuiteBase extends QueryTest
+ with SharedSparkSession
+ with AdaptiveSparkPlanHelper {
+
+ // Standard V2-only conf for DPP tests on file sources. AQE is set by the
concrete subclasses,
+ // which mix in DisableAdaptiveExecutionSuite /
EnableAdaptiveExecutionSuite: the DPP subquery is
+ // planned by PlanDynamicPruningFilters without AQE and by
PlanAdaptiveDynamicPruningFilters with
+ // it, and those are two separate rules.
+ private def withDppV2Conf[T](thunk: => T): T = {
+ withSQLConf(
+ SQLConf.USE_V1_SOURCE_LIST.key -> "",
+ SQLConf.DYNAMIC_PARTITION_PRUNING_ENABLED.key -> "true",
+ SQLConf.DYNAMIC_PARTITION_PRUNING_REUSE_BROADCAST_ONLY.key -> "false",
+ SQLConf.DYNAMIC_PARTITION_PRUNING_FALLBACK_FILTER_RATIO.key ->
"2")(thunk)
+ }
+
+ // Writes a partitioned `fact` and an unpartitioned `dim` table in `format`
and registers them
+ // as temp views. `fact` has 100 rows across 10 partitions (10 rows each);
`dim` has 10 rows
+ // with `dim_id` and `dim_val` both in [0, 10). The read schema is given
explicitly because csv
+ // infers strings, and a cast on the join key changes whether DPP fires at
all.
+ private def writeFactAndDim(dir: File, format: String = "parquet"): Unit = {
+ val factPath = new File(dir, "fact").getCanonicalPath
+ val dimPath = new File(dir, "dim").getCanonicalPath
+ spark.range(100)
+ .selectExpr("id", "id % 10 AS part")
+ .write.format(format).partitionBy("part").save(factPath)
+ spark.read.format(format).schema("id long, part int").load(factPath)
+ .createOrReplaceTempView("fact")
+ spark.range(10)
+ .selectExpr("id AS dim_id", "id AS dim_val")
+ .write.format(format).save(dimPath)
+ spark.read.format(format).schema("dim_id long, dim_val long").load(dimPath)
+ .createOrReplaceTempView("dim")
+ }
+
+ // The BatchScanExec reading `fact`, the only partitioned table of the two.
The lookup descends
+ // into AQE query stages, so it finds the scan whichever stage it ended up
in.
+ private def factScanOf(df: DataFrame): BatchScanExec = {
+ val executedPlan = df.queryExecution.executedPlan
+ collectFirst(executedPlan) {
+ case b: BatchScanExec if b.scan.isInstanceOf[FileScan] &&
+ b.scan.asInstanceOf[FileScan].readPartitionSchema.nonEmpty => b
+ }.getOrElse(fail("no fact BatchScanExec found in:\n" +
executedPlan.treeString))
+ }
+
+ private def fileScanOf(df: DataFrame): FileScan =
factScanOf(df).scan.asInstanceOf[FileScan]
+
+ private def collectDppFilters(plan: LogicalPlan): Seq[Expression] = {
+ plan.collect {
+ case f: Filter if
f.condition.exists(_.isInstanceOf[DynamicPruningSubquery]) => f.condition
+ }
+ }
+
+ test("DPP inserts DynamicPruningSubquery for V2 parquet partitioned table") {
+ withDppV2Conf {
+ withTempDir { dir =>
+ writeFactAndDim(dir)
+ val df = sql(
+ """SELECT f.id FROM fact f JOIN dim d
+ |ON f.part = d.dim_id WHERE d.dim_val = 7""".stripMargin)
+ val optimized = df.queryExecution.optimizedPlan
+ assert(collectDppFilters(optimized).nonEmpty,
+ "expected DynamicPruningSubquery in optimized plan over V2 parquet,
got plan:\n" +
+ optimized.treeString)
+ }
+ }
+ }
+
+ test("DPP does not fire when join key is a data column (non-partition)") {
+ withDppV2Conf {
+ withTempDir { dir =>
+ writeFactAndDim(dir)
+ // Join on f.id (data column), not f.part (partition column). `f.part`
stays in the
+ // projection so the fact scan still reads a partition column: without
it the scan declares
+ // nothing filterable for a reason that has nothing to do with the
join key.
+ val df = sql(
+ """SELECT f.id, f.part FROM fact f JOIN dim d
+ |ON f.id = d.dim_id WHERE d.dim_val = 7""".stripMargin)
+ val optimized = df.queryExecution.optimizedPlan
+ assert(collectDppFilters(optimized).isEmpty,
+ "DPP should not fire on non-partition join keys, got plan:\n" +
+ optimized.treeString)
+ assert(df.collect().length === 1)
+ // And the scan is one that could have been pruned, had the join key
been its partition
+ // column: a filterAttributes() reporting every read column would make
DPP fire above, and
+ // the fact scan would read fewer than all 100 rows.
+ assert(fileScanOf(df).filterAttributes().flatMap(_.fieldNames).toSeq
=== Seq("part"))
+ val numOutputRows = factScanOf(df).metrics("numOutputRows").value
+ assert(numOutputRows === 100,
+ s"expected the fact scan to read every row, got $numOutputRows")
+ }
+ }
+ }
+
+ test("DPP fires when partitioned fact is on the right side of the join") {
+ withDppV2Conf {
+ withTempDir { dir =>
+ writeFactAndDim(dir)
+ // dim on left, fact on right; join on partition column
+ val df = sql(
+ """SELECT f.id FROM dim d JOIN fact f
+ |ON d.dim_id = f.part WHERE d.dim_val = 7""".stripMargin)
+ val optimized = df.queryExecution.optimizedPlan
+ assert(collectDppFilters(optimized).nonEmpty,
+ "expected DPP on right-side partitioned fact, got plan:\n" +
+ optimized.treeString)
+ checkAnswer(df, (0 until 10).map(i => Row(i * 10 + 7)))
+ val numOutputRows = factScanOf(df).metrics("numOutputRows").value
+ assert(numOutputRows === 10,
+ s"expected the fact scan to read only partition 7, got
$numOutputRows rows")
+ }
+ }
+ }
+
+ test("DPP filter reaches BatchScanExec.runtimeFilters as
DynamicPruningExpression") {
+ withDppV2Conf {
+ withTempDir { dir =>
+ writeFactAndDim(dir)
+ val df = sql(
+ """SELECT f.id FROM fact f JOIN dim d
+ |ON f.part = d.dim_id WHERE d.dim_val = 7""".stripMargin)
+ df.collect()
+ val factScan = factScanOf(df)
+ // A DPP filter that degraded to true is also a
DynamicPruningExpression, so match on the
+ // payload: only a subquery child means the filter can actually prune
anything.
+ val dppFilters = factScan.runtimeFilters.collect {
+ case d @ DynamicPruningExpression(child) if
!child.isInstanceOf[Literal] => d
+ }
+ assert(dppFilters.nonEmpty,
+ "expected the fact scan's runtimeFilters to carry a
DynamicPruningExpression over a " +
+ s"subquery, got ${factScan.runtimeFilters}")
+ }
+ }
+ }
+
+ test("DPP prunes input partitions at runtime for v2 parquet") {
+ // Fact has 100 rows across 10 partitions (10 rows each). DPP filter
d.dim_val = 7 selects
+ // exactly partition 7 -> 10 rows. Without runtime pruning, all 100 rows
are read.
+ withDppV2Conf {
+ withTempDir { dir =>
+ writeFactAndDim(dir)
+ val df = sql(
+ """SELECT f.id FROM fact f JOIN dim d
+ |ON f.part = d.dim_id WHERE d.dim_val = 7""".stripMargin)
+ val rows = df.collect()
+ // Correctness: 10 fact rows match (partition 7), each multiplied by 1
dim row.
+ assert(rows.length == 10,
+ s"expected 10 rows after join+filter, got ${rows.length}")
+ // Pruning: fact scan should have read only the 10 rows in partition 7.
+ val numOutputRows = factScanOf(df).metrics("numOutputRows").value
+ assert(numOutputRows == 10,
+ s"expected fact scan to read 10 rows after DPP, got $numOutputRows.
plan:\n" +
+ df.queryExecution.executedPlan.treeString)
+ }
+ }
+ }
+
+ test("DPP that prunes every partition leaves the scan with no input
partition") {
+ // `fact5` has partitions 0..4 while the dim row selects 7, so the build
side is not empty and
+ // the surviving partition set is. That combination is what reaches the
zero-partition
+ // DataSourceRDD: BatchScanExec's EmptyRDDWithPartitions branch keys off
SinglePartition, which
+ // a file scan never reports.
+ withDppV2Conf {
+ withTempDir { dir =>
+ writeFactAndDim(dir)
+ val factPath = new File(dir, "fact5").getCanonicalPath
+ spark.range(50).selectExpr("id", "id % 5 AS part")
+ .write.format("parquet").partitionBy("part").save(factPath)
+ spark.read.format("parquet").schema("id long, part int").load(factPath)
+ .createOrReplaceTempView("fact5")
+ val df = sql(
+ """SELECT f.id FROM fact5 f JOIN dim d
+ |ON f.part = d.dim_id WHERE d.dim_val = 7""".stripMargin)
+ assert(df.collect().isEmpty)
+ val factScan = factScanOf(df)
+ assert(factScan.filteredPartitions.flatten.isEmpty,
+ s"expected no input partition, got
${factScan.filteredPartitions.flatten.size}")
+ // What the assertion above does not see: BatchScanExec turning an
empty partition set into
+ // an RDD with one empty partition still satisfies it.
+ assert(factScan.inputRDD.partitions.isEmpty,
+ s"expected an RDD with no partition, got
${factScan.inputRDD.partitions.length}")
+ val numOutputRows = factScan.metrics("numOutputRows").value
+ assert(numOutputRows == 0,
+ s"expected the fact scan to read nothing, got $numOutputRows rows")
+ }
+ }
+ }
+
+ test("a DPP filter a union branch cannot apply is dropped instead of failing
the query") {
+ // PartitionPruning resolves a union key through the first branch only,
and the filter it
+ // inserts is then pushed positionally into every branch. So the
unpartitioned branch's scan
+ // is handed a filter over a column it never declared filterable. Dropping
it costs pruning
+ // on that branch and nothing else, since the join re-applies it; letting
it through means an
+ // INTERNAL_ERROR from a query that runs today.
+ withDppV2Conf {
+ withTempDir { dir =>
+ writeFactAndDim(dir)
+ val otherPath = new File(dir, "fact_other").getCanonicalPath
+ spark.range(100).selectExpr("id", "id % 10 AS other")
+ .write.format("parquet").save(otherPath)
+ spark.read.format("parquet").schema("id long, other
long").load(otherPath)
+ .createOrReplaceTempView("fact_other")
+ val df = sql(
+ """SELECT u.id FROM (
+ | SELECT id, part FROM fact UNION ALL SELECT id, other FROM
fact_other
+ |) u JOIN dim d ON u.part = d.dim_id WHERE d.dim_val =
7""".stripMargin)
+ val expected = (0 until 10).map(i => Row(i * 10 + 7))
+ checkAnswer(df, expected ++ expected)
+ // The filter did reach the branch that cannot use it: that is what
makes this a regression
+ // test rather than a query that happens to work. Both the
unpartitioned branch and `dim`
+ // are scans with no partition column, so ask for the one that was
handed a live DPP filter.
+ val unusable = collect(df.queryExecution.executedPlan) {
+ case b: BatchScanExec if b.scan.isInstanceOf[FileScan] &&
+ b.scan.asInstanceOf[FileScan].readPartitionSchema.isEmpty &&
+ b.runtimeFilters.exists {
+ case DynamicPruningExpression(child) =>
!child.isInstanceOf[Literal]
+ case _ => false
+ } => b
+ }
+ assert(unusable.size == 1,
+ "expected the DPP filter to reach exactly the branch that cannot
apply it, got " +
+ unusable.map(_.scan.description()).mkString("\n"))
+ // And the branch that can use it still pruned.
+ val numOutputRows = factScanOf(df).metrics("numOutputRows").value
+ assert(numOutputRows == 10,
+ s"expected the partitioned branch to read 10 rows, got
$numOutputRows")
+ }
+ }
+ }
+
+ test("a partition value the runtime filter cannot evaluate fails the query")
{
+ // A partition value the filter cannot be evaluated against has to fail
the query: keeping the
+ // file would return rows that nothing filters, since the filter is
declared fully pushed. What
+ // this pins is that the failure is not swallowed. The exposure is the
same one a compile-time
+ // partition filter over the same column has, and it is why this path must
not gain the
+ // eval-failure tolerance the iterative PartitionPredicate path has. That
no FilterExec is left
+ // above the scan is pinned separately.
+ withDppV2Conf {
+ withSQLConf(SQLConf.ANSI_ENABLED.key -> "true") {
+ withTempDir { dir =>
+ writeFactAndDim(dir)
+ val strPath = new File(dir, "fact_str").getCanonicalPath
+ spark.range(10)
+ .selectExpr("id", "CASE WHEN id = 3 THEN 'abc' ELSE CAST(id AS
STRING) END AS part")
+ .write.format("parquet").partitionBy("part").save(strPath)
+ spark.read.format("parquet").schema("id long, part
string").load(strPath)
+ .createOrReplaceTempView("fact_str")
+ val e = intercept[SparkThrowable] {
+ sql(
+ """SELECT f.id FROM fact_str f
+ |WHERE f.part = (SELECT max(dim_id) FROM dim WHERE dim_val =
7)""".stripMargin)
+ .collect()
+ }
+ assert(e.getCondition === "CAST_INVALID_INPUT",
+ s"expected the failing cast to surface, got ${e.getCondition}")
+ }
+ }
+ }
+ }
+
+ test("a subclass that narrows the file set in partitions keeps it under a
runtime filter") {
+ // The runtime-filter path filters what `partitions` returned, so an
override of it is honored.
+ // Listing the files again instead would hand the excluded ones back as
soon as a filter fired,
+ // and with the filter declared fully pushed nothing above the scan would
remove them.
+ withDppV2Conf {
+ withTempDir { dir =>
+ writeFactAndDim(dir)
+ val df = sql("SELECT f.id, f.part FROM fact f")
+ df.collect()
+ val scan = fileScanOf(df).asInstanceOf[ParquetScan]
+ // Keeps the files of one partition directory, which the second filter
below does not
+ // select, and counts how often the scan asks for its partitions.
+ var partitionsCalls = 0
+ val narrowed = new ParquetScan(scan.sparkSession, scan.hadoopConf,
scan.fileIndex,
+ scan.dataSchema, scan.readDataSchema, scan.readPartitionSchema,
scan.pushedFilters,
+ scan.options, scan.pushedAggregate, scan.partitionFilters,
scan.dataFilters,
+ scan.pushedVariantExtractions) {
+ override protected def partitions: Seq[FilePartition] = {
+ partitionsCalls += 1
+ super.partitions.map { part =>
+ part.copy(files = part.files.filter(_.partitionValues.getInt(0)
== 3))
+ }.filter(_.files.nonEmpty)
+ }
+ }
+ val partAttr = AttributeReference("part", IntegerType)()
+ val kept = narrowed.planInputPartitionsWithRuntimeFilters(
+ Array(EqualTo(partAttr, Literal(3))))
+
assert(kept.flatMap(_.asInstanceOf[FilePartition].files.map(_.partitionValues.getInt(0)))
+ .distinct.toSeq === Seq(3))
+ // And a filter selecting a directory the override dropped comes back
with nothing.
+ assert(narrowed.planInputPartitionsWithRuntimeFilters(
+ Array(EqualTo(partAttr, Literal(7)))).isEmpty)
+ // Both calls filtered one listing, which is what keeps the plan and
the execution on one
+ // snapshot of the file index.
+ assert(partitionsCalls === 1, s"expected one listing, got
$partitionsCalls")
+ }
+ }
+ }
+
+ test("scalar subquery on partition column of v2 parquet prunes partitions") {
+ // Scalar subquery (SELECT max(dim_id) FROM dim WHERE dim_val = 7)
evaluates to 7.
+ // WHERE f.part = 7 should select exactly partition 7 -> 10 rows. Without
partition
+ // pruning, the scan reads all 100 rows and the post-filter keeps 10.
+ withDppV2Conf {
+ withTempDir { dir =>
+ writeFactAndDim(dir)
+ val df = sql(
+ """SELECT f.id FROM fact f
+ |WHERE f.part = (SELECT max(dim_id) FROM dim WHERE dim_val =
7)""".stripMargin)
+ val rows = df.collect()
+ assert(rows.length == 10,
+ s"expected 10 rows after scalar-subquery filter, got ${rows.length}")
+ val numOutputRows = factScanOf(df).metrics("numOutputRows").value
+ assert(numOutputRows == 10,
+ s"expected fact scan to read 10 rows after scalar-subquery pruning,
got " +
+ s"$numOutputRows. plan:\n" +
df.queryExecution.executedPlan.treeString)
+ }
+ }
+ }
+
+ test("filterAttributes reports the read partition columns") {
+ withDppV2Conf {
+ withTempDir { dir =>
+ writeFactAndDim(dir)
+ val df = sql("SELECT f.id, f.part FROM fact f")
+ df.collect()
+ val declared =
fileScanOf(df).filterAttributes().flatMap(_.fieldNames).toSeq
+ assert(declared == Seq("part"),
+ s"expected the partition column to be declared runtime-filterable,
got $declared")
+ }
+ }
+ }
+
+ test("filterAttributes reports nothing once a pushed aggregate drops the
partition columns") {
+ // With an aggregate pushed down, ParquetScan.readSchema() returns only
the aggregate schema, so
+ // the partition columns are not in the scan relation output. Spark
resolves every declared
+ // runtime-filter attribute against that output and fails the query if one
is missing, so
+ // filterAttributes() must not report them.
+ withDppV2Conf {
+ withSQLConf(SQLConf.PARQUET_AGGREGATE_PUSHDOWN_ENABLED.key -> "true") {
+ withTempDir { dir =>
+ writeFactAndDim(dir)
+ val df = sql("SELECT count(part) FROM fact")
+ checkAnswer(df, Row(100))
+ val declared =
fileScanOf(df).filterAttributes().flatMap(_.fieldNames).toSeq
+ assert(declared.isEmpty,
+ s"expected no runtime-filterable attribute under a pushed
aggregate, got $declared")
+ }
+ }
+ }
+ }
+
+ test("filterAttributes keeps a partition column a pushed aggregate groups
by") {
+ // Aggregate pushdown only happens when the GROUP BY set covers every
partition column, and
+ // those columns go into the pushed-down schema, so they stay in the scan
relation output and
+ // stay runtime-filterable. This is the other side of the intersection
from the test above.
+ withDppV2Conf {
+ withSQLConf(SQLConf.PARQUET_AGGREGATE_PUSHDOWN_ENABLED.key -> "true") {
+ withTempDir { dir =>
+ writeFactAndDim(dir)
+ val df = sql("SELECT part, count(*) FROM fact GROUP BY part")
+ checkAnswer(df, (0 until 10).map(p => Row(p, 10)))
+ val scan = fileScanOf(df)
+ // Without the pushdown the assertion below says nothing: `part` is
in the read partition
+ // schema either way. Pin that the aggregate really went into the
scan.
+ assert(scan.asInstanceOf[ParquetScan].pushedAggregate.isDefined,
+ s"expected the aggregate to be pushed into the scan, got
${scan.description()}")
+ val declared = scan.filterAttributes().flatMap(_.fieldNames).toSeq
+ assert(declared == Seq("part"),
+ s"expected the group-by partition column to stay declared, got
$declared")
+ }
+ }
+ }
+ }
+
+ test("scalar subquery on a partition column is not re-evaluated after the
scan") {
+ // FileScan declares its partition columns in fullyPushedFilterAttributes:
applying such a
+ // filter selects partition directories, so every row the scan returns
satisfies it and Spark
+ // drops the post-scan FilterExec, the same treatment a compile-time
partition filter gets.
+ withDppV2Conf {
+ withTempDir { dir =>
+ writeFactAndDim(dir)
+ val df = sql(
+ """SELECT f.id FROM fact f
+ |WHERE f.part = (SELECT max(dim_id) FROM dim WHERE dim_val =
7)""".stripMargin)
+ val rows = df.collect()
+ assert(rows.length == 10, s"expected 10 rows, got ${rows.length}")
+ val executedPlan = df.queryExecution.executedPlan
+ assert(collect(executedPlan) { case f: FilterExec => f }.isEmpty,
+ "a fully pushed partition filter should leave no FilterExec above
the scan, got:\n" +
+ executedPlan.treeString)
+ }
+ }
+ }
+
+ test("a runtime filter is ANDed with the compile-time partition filters, not
substituted") {
+ // The two halves are applied in different places: `partitions` puts
`partitionFilters` through
+ // `fileIndex.listFiles`, and `planInputPartitionsWithRuntimeFilters`
filters what that planned.
+ // Both are the scan's only evaluator: FileScanBuilder.pushFilters returns
just the data filters
+ // as post-scan filters, and a partition-column runtime filter is declared
fully pushed, so
+ // dropping either half returns rows that nothing filters out.
+ withDppV2Conf {
+ withTempDir { dir =>
+ writeFactAndDim(dir)
+ // min(dim_id) over dim_val > 6 is 7, so this keeps partitions 7 and
8: 7 from the runtime
+ // filter's lower bound, 8 from the compile-time upper bound.
+ val df = sql(
+ """SELECT f.id FROM fact f
+ |WHERE f.part <= 8
+ | AND f.part >= (SELECT min(dim_id) FROM dim WHERE dim_val >
6)""".stripMargin)
+ checkAnswer(df, (0 until 10).flatMap(i => Seq(Row(i * 10 + 7), Row(i *
10 + 8))))
+ val numOutputRows = factScanOf(df).metrics("numOutputRows").value
+ assert(numOutputRows == 20,
+ s"expected the fact scan to read only partitions 7 and 8, got
$numOutputRows rows")
+ }
+ }
+ }
+
+ test("runtime partition pruning survives a mixed-case partition column
reference") {
+ // `f.PART` is resolved to the schema's `part` before it ever becomes a
runtime filter, so this
+ // pins the end-to-end path rather than any name matching: what the scan
receives is identical
+ // to the lower-case query's. The test below is the one that pins matching
by exact name.
Review Comment:
Nit: "The test below is the one that pins matching by exact name" is stale.
Since the guard moved to `normalizeName`, the test below asserts that `PART`
binds and returns partition 7; only `id` hits the guard. The PR description has
the same leftover in its test section ("or with the partition column spelled
differently, hits the scan's guard"). Could we reword both, e.g. "pins how the
scan matches the name" here?
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/FileScan.scala:
##########
@@ -182,8 +185,80 @@ trait FileScan extends Scan
FilePartition.getFilePartitions(sparkSession, splitFiles, maxSplitBytes)
}
+ /**
+ * The partitions `partitions` planned, computed once per scan instance.
Both the plain read path
+ * and the runtime-filter path go through this, so a scan node that gets
runtime filters lists the
+ * files once and filters that listing, rather than reading the index twice
and risking two
+ * snapshots. A subclass still customizes `partitions`.
+ */
+ @transient private lazy val plannedPartitions: Seq[FilePartition] =
partitions
+
override def planInputPartitions(): Array[InputPartition] = {
- partitions.toArray
+ plannedPartitions.toArray
+ }
+
+ /**
+ * The partition columns Spark can derive a runtime filter on (SPARK-30628),
restricted to the
+ * ones `readSchema()` still exposes: a reference missing from the scan
relation output fails to
+ * resolve, and a pushed-down aggregate keeps only the partition columns it
groups by.
+ *
+ * A filter over one of them is evaluated against each file's partition
values, so the scan
+ * evaluates it in full and Spark does not evaluate it again after the scan
--
+ * `FileScanBuilder.pushFilters` already keeps compile-time partition
filters out of the post-scan
+ * filters for the same reason.
+ */
+ override def filterAttributes(): Array[NamedReference] = {
+ val readFields = readSchema().fieldNames.map(normalizeName).toSet
+ readPartitionSchema.fieldNames
+ .filter(name => readFields.contains(normalizeName(name)))
+ .map(FieldReference.column)
+ }
+
+ override def fullyPushedFilterAttributes(): Array[NamedReference] =
filterAttributes()
Review Comment:
**With the partition columns declared fully pushed, a scalar-subquery
partition filter disappears from `EXPLAIN FORMATTED` and from the SQL UI, which
uses the formatted mode by default.** Before this PR the predicate showed as
the `FilterExec` condition. That node is gone now, and the formatted output
never prints `BatchScanExec.runtimeFilters`: the tree line comes from
`simpleStringWithNodeId`, and the details come from
`DataSourceV2ScanExecBase.verboseStringWithOperatorId`, where
`PartitionFilters` (L139) lists only compile-time filters. For `SELECT id FROM
fact WHERE part = (SELECT max(dim_id) FROM dim WHERE dim_val = 7)`, the plan
shows the hosting subquery but no `part = ...` anywhere. V1 shows the same
filter under `PartitionFilters`.
Could `BatchScanExec` override `verboseStringWithOperatorId` to add
`RuntimeFilters: [...]` when non-empty, with an `ExplainSuite` case for a V2
file source?
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2FileSourceDPPSuite.scala:
##########
@@ -0,0 +1,658 @@
+/*
+ * 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.connector
+
+import java.io.File
+
+import org.apache.spark.{SparkException, SparkThrowable}
+import org.apache.spark.sql.{DataFrame, QueryTest, Row}
+import org.apache.spark.sql.catalyst.expressions.{AttributeReference,
DynamicPruningExpression, DynamicPruningSubquery, EqualTo, Expression, Literal}
+import org.apache.spark.sql.catalyst.plans.logical.{Filter, LogicalPlan}
+import org.apache.spark.sql.execution.FilterExec
+import org.apache.spark.sql.execution.adaptive.{AdaptiveSparkPlanHelper,
DisableAdaptiveExecutionSuite, EnableAdaptiveExecutionSuite}
+import org.apache.spark.sql.execution.datasources.FilePartition
+import org.apache.spark.sql.execution.datasources.v2.{BatchScanExec, FileScan}
+import org.apache.spark.sql.execution.datasources.v2.parquet.ParquetScan
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.test.SharedSparkSession
+import org.apache.spark.sql.types.{IntegerType, LongType}
+
+/**
+ * Tests for SPARK-30628: subquery partition pruning and DPP for V2 file
sources.
+ *
+ * These tests exercise V2 file scans (e.g. ParquetScan) -- distinct from the
in-memory V2
+ * catalog used by [[org.apache.spark.sql.DynamicPartitionPruningV2Suite]] and
from the V2
+ * iterative-pushdown work covered by
[[DataSourceV2EnhancedRuntimePartitionFilterSuite]].
+ */
+abstract class DataSourceV2FileSourceDPPSuiteBase extends QueryTest
+ with SharedSparkSession
+ with AdaptiveSparkPlanHelper {
+
+ // Standard V2-only conf for DPP tests on file sources. AQE is set by the
concrete subclasses,
+ // which mix in DisableAdaptiveExecutionSuite /
EnableAdaptiveExecutionSuite: the DPP subquery is
+ // planned by PlanDynamicPruningFilters without AQE and by
PlanAdaptiveDynamicPruningFilters with
+ // it, and those are two separate rules.
+ private def withDppV2Conf[T](thunk: => T): T = {
+ withSQLConf(
+ SQLConf.USE_V1_SOURCE_LIST.key -> "",
+ SQLConf.DYNAMIC_PARTITION_PRUNING_ENABLED.key -> "true",
+ SQLConf.DYNAMIC_PARTITION_PRUNING_REUSE_BROADCAST_ONLY.key -> "false",
+ SQLConf.DYNAMIC_PARTITION_PRUNING_FALLBACK_FILTER_RATIO.key ->
"2")(thunk)
+ }
+
+ // Writes a partitioned `fact` and an unpartitioned `dim` table in `format`
and registers them
+ // as temp views. `fact` has 100 rows across 10 partitions (10 rows each);
`dim` has 10 rows
+ // with `dim_id` and `dim_val` both in [0, 10). The read schema is given
explicitly because csv
+ // infers strings, and a cast on the join key changes whether DPP fires at
all.
+ private def writeFactAndDim(dir: File, format: String = "parquet"): Unit = {
+ val factPath = new File(dir, "fact").getCanonicalPath
+ val dimPath = new File(dir, "dim").getCanonicalPath
+ spark.range(100)
+ .selectExpr("id", "id % 10 AS part")
+ .write.format(format).partitionBy("part").save(factPath)
Review Comment:
**Every fixture has the single partition column `part`, so projected
`partitionValues` are never combined with a runtime filter.** `partitions`
projects the values to `readPartitionSchema` only when a strict subset of the
partition columns is read (FileScan.scala:159-160), and the new binding depends
on `readPartitionSchema` ordinals (FileScan.scala:234-235). Binding by
`fileIndex.partitionSchema` ordinals, as V1 and an earlier revision of this
method did against unprojected values, would read the wrong slot here, and
every current test would still pass because the ordinal is 0 in both schemas.
With no `FilterExec` left, that means wrong rows. For example, with
`partitionBy("p1", "p2", "p3")` and `SELECT id, p3 FROM t WHERE p2 = (SELECT
...)`, `p2` would read the `p3` slot.
Could we add that case, plus a DPP join on `p2`, asserting
`readPartitionSchema.fieldNames.toSeq === Seq("p2", "p3")` so the projection
stays engaged?
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/FileScan.scala:
##########
@@ -182,8 +185,80 @@ trait FileScan extends Scan
FilePartition.getFilePartitions(sparkSession, splitFiles, maxSplitBytes)
}
+ /**
+ * The partitions `partitions` planned, computed once per scan instance.
Both the plain read path
+ * and the runtime-filter path go through this, so a scan node that gets
runtime filters lists the
+ * files once and filters that listing, rather than reading the index twice
and risking two
+ * snapshots. A subclass still customizes `partitions`.
+ */
+ @transient private lazy val plannedPartitions: Seq[FilePartition] =
partitions
+
override def planInputPartitions(): Array[InputPartition] = {
- partitions.toArray
+ plannedPartitions.toArray
+ }
+
+ /**
+ * The partition columns Spark can derive a runtime filter on (SPARK-30628),
restricted to the
+ * ones `readSchema()` still exposes: a reference missing from the scan
relation output fails to
+ * resolve, and a pushed-down aggregate keeps only the partition columns it
groups by.
+ *
+ * A filter over one of them is evaluated against each file's partition
values, so the scan
+ * evaluates it in full and Spark does not evaluate it again after the scan
--
+ * `FileScanBuilder.pushFilters` already keeps compile-time partition
filters out of the post-scan
+ * filters for the same reason.
+ */
+ override def filterAttributes(): Array[NamedReference] = {
+ val readFields = readSchema().fieldNames.map(normalizeName).toSet
+ readPartitionSchema.fieldNames
+ .filter(name => readFields.contains(normalizeName(name)))
+ .map(FieldReference.column)
+ }
+
+ override def fullyPushedFilterAttributes(): Array[NamedReference] =
filterAttributes()
+
+ /**
+ * Plans the files whose partition values satisfy `expressions`, taken from
what `partitions`
+ * planned and bin-packed again for the surviving set.
+ *
+ * Filtering what was planned rather than listing the files again keeps one
listing per scan, and
+ * the files a subclass excluded in `partitions` stay excluded. How the
survivors group into
+ * partitions is decided here again, for their own size. Where each file was
cut into splits is
+ * not: `partitions` cut them for the whole set it listed, before these
filters existed.
+ *
+ * A partition value is fixed within a file, so every row of every partition
returned satisfies
+ * the expressions even though one partition can still hold files from
several partition
+ * directories.
+ */
+ override def planInputPartitionsWithRuntimeFilters(
+ expressions: Array[Expression]): Array[InputPartition] = {
+ val fieldIndex = readPartitionSchema.fieldNames.zipWithIndex
+ .map { case (name, i) => normalizeName(name) -> i }.toMap
+ // Evaluating these is the scan's only chance to apply them, since the
attributes are declared
+ // fully pushed and Spark drops the post-scan filter. A reference outside
the read partition
+ // schema has no value to evaluate against, so fail loudly rather than
return wrong rows. Spark
+ // screens against `filterAttributes()` before it gets here, which is a
subset of that schema.
+ val notApplicable = expressions.filterNot(
+ _.references.forall(a => fieldIndex.contains(normalizeName(a.name))))
+ if (notApplicable.nonEmpty) {
+ throw SparkException.internalError("A file scan can only apply a runtime
filter over its " +
+ s"read partition columns
${readPartitionSchema.fieldNames.mkString("[", ", ", "]")}, " +
+ s"got ${notApplicable.mkString(", ")}")
+ }
+ val bound = expressions.reduce(And).transform {
+ case a: Attribute => BoundReference(fieldIndex(normalizeName(a.name)),
a.dataType, a.nullable)
+ }
+ val predicate = Predicate.createInterpreted(bound)
+ val keptFiles = plannedPartitions
+ .flatMap(_.files.filter(file => predicate.eval(file.partitionValues)))
+ .sortBy(_.length)(implicitly[Ordering[Long]].reverse)
+ // Pack the survivors instead of handing back the planned partitions with
files removed:
+ // `partitions` sized those bins for the whole file set, so a scan that
pruned most of its files
+ // would run the same few tasks over a fraction of the data. V1 packs the
pruned set the same
+ // way, from `dynamicallySelectedPartitions`.
+ val openCostInBytes = conf.filesOpenCostInBytes
+ val maxSplitBytes = FilePartition.maxSplitBytes(
+ sparkSession, keptFiles.map(_.length + openCostInBytes).sum)
+ FilePartition.getFilePartitions(sparkSession, keptFiles,
maxSplitBytes).toArray
Review Comment:
**Correcting my own [round-3
suggestion](https://github.com/apache/spark/pull/55579#issuecomment-5768680813):
the repack keeps only the file set of a `partitions` override, not its
grouping or partition type.** There I wrote that both could be kept, but
`getFilePartitions` re-bins the survivors into plain `FilePartition`s. A
subclass that mixes in `SupportsReportPartitioning` and returns
`HasPartitionKey` splits from `partitions` would hit "Data source must have
preserved the original partitioning during runtime filtering"
(PushDownUtils.scala:334-337) as soon as a DPP or scalar-subquery filter fires;
YTsaurus SPYT's `YtScan` already mixes `SupportsReportPartitioning` into a
`FileScan`. A subclass that reports an ordering with one sorted file per split
loses it once Next Fit packs several files together. The PR description's
"keeps whatever a subclass did in `partitions`" and the test comment at
DataSourceV2FileSourceDPPSuite.scala:303 overstate this as well.
Could `filterAttributes()` return nothing when the scan reports partitioning
or ordering (or the opt-in be per format), so that the repack only applies to
scans whose layout Spark owns?
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/connector/SupportsRuntimeCatalystFiltering.scala:
##########
@@ -72,35 +80,34 @@ trait SupportsRuntimeCatalystFiltering extends Scan {
def fullyPushedFilterAttributes(): Array[NamedReference] = Array.empty
/**
- * Filters this scan using runtime Catalyst expressions.
+ * Plans this scan's input partitions with runtime Catalyst expressions
applied.
+ *
+ * The provided expressions must be interpreted as a set of predicates that
are ANDed together,
+ * and applied on top of the filters already pushed at query compilation.
Implementations may use
+ * them to plan fewer [[InputPartition]]s than
[[org.apache.spark.sql.connector.read.Batch]]
+ * would without them.
*
- * The provided expressions must be interpreted as a set of predicates that
are ANDed together.
- * Implementations may use the expressions to prune initially planned
- * [[org.apache.spark.sql.connector.read.InputPartition]]s.
+ * Spark calls this method at execution time, and only when at least one
runtime filter survives
+ * screening; the partitions it returns replace the ones that scan node
would otherwise have
+ * read. Spark may also call `Batch.planInputPartitions()` on the same scan,
so neither method may
+ * assume it is the only one used. One scan instance can back several scan
nodes (e.g. the two
+ * branches of a group-based UPDATE), each with its own runtime filters, so
an implementation
Review Comment:
Minor: with the old mutating `filter`, a group-based (`ReplaceData`) scan
narrowed itself, so its write could ask the scan what had been read. This
contract now forbids carrying state from one call to the next, but it does not
say how such a write learns which groups were read. The in-memory fixture
needed the new `readPartitions`, a union of every call's result, for exactly
this reason. Since every `FileScan` now also enables
`RowLevelOperationRuntimeGroupFiltering` for a row-level operation built on it,
a connector whose commit replaces the files of the scan it configured would
delete groups that were never read. Could this paragraph say that a group-based
scan has to record what each call returned, and treat a node that read through
`planInputPartitions()` as having read everything?
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/FileScan.scala:
##########
@@ -182,8 +185,80 @@ trait FileScan extends Scan
FilePartition.getFilePartitions(sparkSession, splitFiles, maxSplitBytes)
}
+ /**
+ * The partitions `partitions` planned, computed once per scan instance.
Both the plain read path
+ * and the runtime-filter path go through this, so a scan node that gets
runtime filters lists the
+ * files once and filters that listing, rather than reading the index twice
and risking two
+ * snapshots. A subclass still customizes `partitions`.
+ */
+ @transient private lazy val plannedPartitions: Seq[FilePartition] =
partitions
Review Comment:
Minor: as a Scala 2.13 `lazy val`, this runs `partitions`, including a
subclass override, while holding the scan instance's monitor, which
`planInputPartitions()` never did before. An override that hands the work to
another thread and waits, e.g. `Await.result(Future(super.partitions),
timeout)`, deadlocks planning as soon as that thread first touches another lazy
val of the same scan, such as `codecFactory` through `isSplitable` in a
`TextBasedFileScan`. Nothing in-tree does this. Could the memo be lock-free
(e.g. an `AtomicReference` set with `compareAndSet`), or at least documented on
`partitions`?
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2FileSourceDPPSuite.scala:
##########
@@ -0,0 +1,658 @@
+/*
+ * 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.connector
+
+import java.io.File
+
+import org.apache.spark.{SparkException, SparkThrowable}
+import org.apache.spark.sql.{DataFrame, QueryTest, Row}
+import org.apache.spark.sql.catalyst.expressions.{AttributeReference,
DynamicPruningExpression, DynamicPruningSubquery, EqualTo, Expression, Literal}
+import org.apache.spark.sql.catalyst.plans.logical.{Filter, LogicalPlan}
+import org.apache.spark.sql.execution.FilterExec
+import org.apache.spark.sql.execution.adaptive.{AdaptiveSparkPlanHelper,
DisableAdaptiveExecutionSuite, EnableAdaptiveExecutionSuite}
+import org.apache.spark.sql.execution.datasources.FilePartition
+import org.apache.spark.sql.execution.datasources.v2.{BatchScanExec, FileScan}
+import org.apache.spark.sql.execution.datasources.v2.parquet.ParquetScan
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.test.SharedSparkSession
+import org.apache.spark.sql.types.{IntegerType, LongType}
+
+/**
+ * Tests for SPARK-30628: subquery partition pruning and DPP for V2 file
sources.
+ *
+ * These tests exercise V2 file scans (e.g. ParquetScan) -- distinct from the
in-memory V2
+ * catalog used by [[org.apache.spark.sql.DynamicPartitionPruningV2Suite]] and
from the V2
+ * iterative-pushdown work covered by
[[DataSourceV2EnhancedRuntimePartitionFilterSuite]].
+ */
+abstract class DataSourceV2FileSourceDPPSuiteBase extends QueryTest
+ with SharedSparkSession
+ with AdaptiveSparkPlanHelper {
+
+ // Standard V2-only conf for DPP tests on file sources. AQE is set by the
concrete subclasses,
+ // which mix in DisableAdaptiveExecutionSuite /
EnableAdaptiveExecutionSuite: the DPP subquery is
+ // planned by PlanDynamicPruningFilters without AQE and by
PlanAdaptiveDynamicPruningFilters with
+ // it, and those are two separate rules.
+ private def withDppV2Conf[T](thunk: => T): T = {
+ withSQLConf(
+ SQLConf.USE_V1_SOURCE_LIST.key -> "",
+ SQLConf.DYNAMIC_PARTITION_PRUNING_ENABLED.key -> "true",
+ SQLConf.DYNAMIC_PARTITION_PRUNING_REUSE_BROADCAST_ONLY.key -> "false",
+ SQLConf.DYNAMIC_PARTITION_PRUNING_FALLBACK_FILTER_RATIO.key ->
"2")(thunk)
+ }
+
+ // Writes a partitioned `fact` and an unpartitioned `dim` table in `format`
and registers them
+ // as temp views. `fact` has 100 rows across 10 partitions (10 rows each);
`dim` has 10 rows
+ // with `dim_id` and `dim_val` both in [0, 10). The read schema is given
explicitly because csv
+ // infers strings, and a cast on the join key changes whether DPP fires at
all.
+ private def writeFactAndDim(dir: File, format: String = "parquet"): Unit = {
+ val factPath = new File(dir, "fact").getCanonicalPath
+ val dimPath = new File(dir, "dim").getCanonicalPath
+ spark.range(100)
+ .selectExpr("id", "id % 10 AS part")
+ .write.format(format).partitionBy("part").save(factPath)
+ spark.read.format(format).schema("id long, part int").load(factPath)
+ .createOrReplaceTempView("fact")
+ spark.range(10)
+ .selectExpr("id AS dim_id", "id AS dim_val")
+ .write.format(format).save(dimPath)
+ spark.read.format(format).schema("dim_id long, dim_val long").load(dimPath)
+ .createOrReplaceTempView("dim")
+ }
+
+ // The BatchScanExec reading `fact`, the only partitioned table of the two.
The lookup descends
+ // into AQE query stages, so it finds the scan whichever stage it ended up
in.
+ private def factScanOf(df: DataFrame): BatchScanExec = {
+ val executedPlan = df.queryExecution.executedPlan
+ collectFirst(executedPlan) {
+ case b: BatchScanExec if b.scan.isInstanceOf[FileScan] &&
+ b.scan.asInstanceOf[FileScan].readPartitionSchema.nonEmpty => b
+ }.getOrElse(fail("no fact BatchScanExec found in:\n" +
executedPlan.treeString))
+ }
+
+ private def fileScanOf(df: DataFrame): FileScan =
factScanOf(df).scan.asInstanceOf[FileScan]
+
+ private def collectDppFilters(plan: LogicalPlan): Seq[Expression] = {
+ plan.collect {
+ case f: Filter if
f.condition.exists(_.isInstanceOf[DynamicPruningSubquery]) => f.condition
+ }
+ }
+
+ test("DPP inserts DynamicPruningSubquery for V2 parquet partitioned table") {
+ withDppV2Conf {
+ withTempDir { dir =>
+ writeFactAndDim(dir)
+ val df = sql(
+ """SELECT f.id FROM fact f JOIN dim d
+ |ON f.part = d.dim_id WHERE d.dim_val = 7""".stripMargin)
+ val optimized = df.queryExecution.optimizedPlan
+ assert(collectDppFilters(optimized).nonEmpty,
+ "expected DynamicPruningSubquery in optimized plan over V2 parquet,
got plan:\n" +
+ optimized.treeString)
+ }
+ }
+ }
+
+ test("DPP does not fire when join key is a data column (non-partition)") {
+ withDppV2Conf {
+ withTempDir { dir =>
+ writeFactAndDim(dir)
+ // Join on f.id (data column), not f.part (partition column). `f.part`
stays in the
+ // projection so the fact scan still reads a partition column: without
it the scan declares
+ // nothing filterable for a reason that has nothing to do with the
join key.
+ val df = sql(
+ """SELECT f.id, f.part FROM fact f JOIN dim d
+ |ON f.id = d.dim_id WHERE d.dim_val = 7""".stripMargin)
+ val optimized = df.queryExecution.optimizedPlan
+ assert(collectDppFilters(optimized).isEmpty,
+ "DPP should not fire on non-partition join keys, got plan:\n" +
+ optimized.treeString)
+ assert(df.collect().length === 1)
+ // And the scan is one that could have been pruned, had the join key
been its partition
+ // column: a filterAttributes() reporting every read column would make
DPP fire above, and
+ // the fact scan would read fewer than all 100 rows.
+ assert(fileScanOf(df).filterAttributes().flatMap(_.fieldNames).toSeq
=== Seq("part"))
+ val numOutputRows = factScanOf(df).metrics("numOutputRows").value
+ assert(numOutputRows === 100,
+ s"expected the fact scan to read every row, got $numOutputRows")
+ }
+ }
+ }
+
+ test("DPP fires when partitioned fact is on the right side of the join") {
+ withDppV2Conf {
+ withTempDir { dir =>
+ writeFactAndDim(dir)
+ // dim on left, fact on right; join on partition column
+ val df = sql(
+ """SELECT f.id FROM dim d JOIN fact f
+ |ON d.dim_id = f.part WHERE d.dim_val = 7""".stripMargin)
+ val optimized = df.queryExecution.optimizedPlan
+ assert(collectDppFilters(optimized).nonEmpty,
+ "expected DPP on right-side partitioned fact, got plan:\n" +
+ optimized.treeString)
+ checkAnswer(df, (0 until 10).map(i => Row(i * 10 + 7)))
+ val numOutputRows = factScanOf(df).metrics("numOutputRows").value
+ assert(numOutputRows === 10,
+ s"expected the fact scan to read only partition 7, got
$numOutputRows rows")
+ }
+ }
+ }
+
+ test("DPP filter reaches BatchScanExec.runtimeFilters as
DynamicPruningExpression") {
+ withDppV2Conf {
+ withTempDir { dir =>
+ writeFactAndDim(dir)
+ val df = sql(
+ """SELECT f.id FROM fact f JOIN dim d
+ |ON f.part = d.dim_id WHERE d.dim_val = 7""".stripMargin)
+ df.collect()
+ val factScan = factScanOf(df)
+ // A DPP filter that degraded to true is also a
DynamicPruningExpression, so match on the
+ // payload: only a subquery child means the filter can actually prune
anything.
+ val dppFilters = factScan.runtimeFilters.collect {
+ case d @ DynamicPruningExpression(child) if
!child.isInstanceOf[Literal] => d
+ }
+ assert(dppFilters.nonEmpty,
+ "expected the fact scan's runtimeFilters to carry a
DynamicPruningExpression over a " +
+ s"subquery, got ${factScan.runtimeFilters}")
+ }
+ }
+ }
+
+ test("DPP prunes input partitions at runtime for v2 parquet") {
+ // Fact has 100 rows across 10 partitions (10 rows each). DPP filter
d.dim_val = 7 selects
+ // exactly partition 7 -> 10 rows. Without runtime pruning, all 100 rows
are read.
+ withDppV2Conf {
+ withTempDir { dir =>
+ writeFactAndDim(dir)
+ val df = sql(
+ """SELECT f.id FROM fact f JOIN dim d
+ |ON f.part = d.dim_id WHERE d.dim_val = 7""".stripMargin)
+ val rows = df.collect()
+ // Correctness: 10 fact rows match (partition 7), each multiplied by 1
dim row.
+ assert(rows.length == 10,
+ s"expected 10 rows after join+filter, got ${rows.length}")
+ // Pruning: fact scan should have read only the 10 rows in partition 7.
+ val numOutputRows = factScanOf(df).metrics("numOutputRows").value
+ assert(numOutputRows == 10,
+ s"expected fact scan to read 10 rows after DPP, got $numOutputRows.
plan:\n" +
+ df.queryExecution.executedPlan.treeString)
+ }
+ }
+ }
+
+ test("DPP that prunes every partition leaves the scan with no input
partition") {
+ // `fact5` has partitions 0..4 while the dim row selects 7, so the build
side is not empty and
+ // the surviving partition set is. That combination is what reaches the
zero-partition
+ // DataSourceRDD: BatchScanExec's EmptyRDDWithPartitions branch keys off
SinglePartition, which
+ // a file scan never reports.
+ withDppV2Conf {
+ withTempDir { dir =>
+ writeFactAndDim(dir)
+ val factPath = new File(dir, "fact5").getCanonicalPath
+ spark.range(50).selectExpr("id", "id % 5 AS part")
+ .write.format("parquet").partitionBy("part").save(factPath)
+ spark.read.format("parquet").schema("id long, part int").load(factPath)
+ .createOrReplaceTempView("fact5")
+ val df = sql(
+ """SELECT f.id FROM fact5 f JOIN dim d
+ |ON f.part = d.dim_id WHERE d.dim_val = 7""".stripMargin)
+ assert(df.collect().isEmpty)
+ val factScan = factScanOf(df)
+ assert(factScan.filteredPartitions.flatten.isEmpty,
+ s"expected no input partition, got
${factScan.filteredPartitions.flatten.size}")
+ // What the assertion above does not see: BatchScanExec turning an
empty partition set into
+ // an RDD with one empty partition still satisfies it.
+ assert(factScan.inputRDD.partitions.isEmpty,
+ s"expected an RDD with no partition, got
${factScan.inputRDD.partitions.length}")
+ val numOutputRows = factScan.metrics("numOutputRows").value
+ assert(numOutputRows == 0,
+ s"expected the fact scan to read nothing, got $numOutputRows rows")
+ }
+ }
+ }
+
+ test("a DPP filter a union branch cannot apply is dropped instead of failing
the query") {
+ // PartitionPruning resolves a union key through the first branch only,
and the filter it
+ // inserts is then pushed positionally into every branch. So the
unpartitioned branch's scan
+ // is handed a filter over a column it never declared filterable. Dropping
it costs pruning
+ // on that branch and nothing else, since the join re-applies it; letting
it through means an
+ // INTERNAL_ERROR from a query that runs today.
+ withDppV2Conf {
+ withTempDir { dir =>
+ writeFactAndDim(dir)
+ val otherPath = new File(dir, "fact_other").getCanonicalPath
+ spark.range(100).selectExpr("id", "id % 10 AS other")
+ .write.format("parquet").save(otherPath)
+ spark.read.format("parquet").schema("id long, other
long").load(otherPath)
+ .createOrReplaceTempView("fact_other")
+ val df = sql(
+ """SELECT u.id FROM (
+ | SELECT id, part FROM fact UNION ALL SELECT id, other FROM
fact_other
+ |) u JOIN dim d ON u.part = d.dim_id WHERE d.dim_val =
7""".stripMargin)
+ val expected = (0 until 10).map(i => Row(i * 10 + 7))
+ checkAnswer(df, expected ++ expected)
+ // The filter did reach the branch that cannot use it: that is what
makes this a regression
+ // test rather than a query that happens to work. Both the
unpartitioned branch and `dim`
+ // are scans with no partition column, so ask for the one that was
handed a live DPP filter.
+ val unusable = collect(df.queryExecution.executedPlan) {
+ case b: BatchScanExec if b.scan.isInstanceOf[FileScan] &&
+ b.scan.asInstanceOf[FileScan].readPartitionSchema.isEmpty &&
+ b.runtimeFilters.exists {
+ case DynamicPruningExpression(child) =>
!child.isInstanceOf[Literal]
+ case _ => false
+ } => b
+ }
+ assert(unusable.size == 1,
+ "expected the DPP filter to reach exactly the branch that cannot
apply it, got " +
+ unusable.map(_.scan.description()).mkString("\n"))
+ // And the branch that can use it still pruned.
+ val numOutputRows = factScanOf(df).metrics("numOutputRows").value
+ assert(numOutputRows == 10,
+ s"expected the partitioned branch to read 10 rows, got
$numOutputRows")
+ }
+ }
+ }
+
+ test("a partition value the runtime filter cannot evaluate fails the query")
{
+ // A partition value the filter cannot be evaluated against has to fail
the query: keeping the
+ // file would return rows that nothing filters, since the filter is
declared fully pushed. What
+ // this pins is that the failure is not swallowed. The exposure is the
same one a compile-time
+ // partition filter over the same column has, and it is why this path must
not gain the
+ // eval-failure tolerance the iterative PartitionPredicate path has. That
no FilterExec is left
+ // above the scan is pinned separately.
+ withDppV2Conf {
+ withSQLConf(SQLConf.ANSI_ENABLED.key -> "true") {
+ withTempDir { dir =>
+ writeFactAndDim(dir)
+ val strPath = new File(dir, "fact_str").getCanonicalPath
+ spark.range(10)
+ .selectExpr("id", "CASE WHEN id = 3 THEN 'abc' ELSE CAST(id AS
STRING) END AS part")
+ .write.format("parquet").partitionBy("part").save(strPath)
+ spark.read.format("parquet").schema("id long, part
string").load(strPath)
+ .createOrReplaceTempView("fact_str")
+ val e = intercept[SparkThrowable] {
+ sql(
+ """SELECT f.id FROM fact_str f
+ |WHERE f.part = (SELECT max(dim_id) FROM dim WHERE dim_val =
7)""".stripMargin)
+ .collect()
+ }
+ assert(e.getCondition === "CAST_INVALID_INPUT",
+ s"expected the failing cast to surface, got ${e.getCondition}")
+ }
+ }
+ }
+ }
+
+ test("a subclass that narrows the file set in partitions keeps it under a
runtime filter") {
+ // The runtime-filter path filters what `partitions` returned, so an
override of it is honored.
+ // Listing the files again instead would hand the excluded ones back as
soon as a filter fired,
+ // and with the filter declared fully pushed nothing above the scan would
remove them.
+ withDppV2Conf {
+ withTempDir { dir =>
+ writeFactAndDim(dir)
+ val df = sql("SELECT f.id, f.part FROM fact f")
+ df.collect()
+ val scan = fileScanOf(df).asInstanceOf[ParquetScan]
+ // Keeps the files of one partition directory, which the second filter
below does not
+ // select, and counts how often the scan asks for its partitions.
+ var partitionsCalls = 0
+ val narrowed = new ParquetScan(scan.sparkSession, scan.hadoopConf,
scan.fileIndex,
+ scan.dataSchema, scan.readDataSchema, scan.readPartitionSchema,
scan.pushedFilters,
+ scan.options, scan.pushedAggregate, scan.partitionFilters,
scan.dataFilters,
+ scan.pushedVariantExtractions) {
+ override protected def partitions: Seq[FilePartition] = {
+ partitionsCalls += 1
+ super.partitions.map { part =>
+ part.copy(files = part.files.filter(_.partitionValues.getInt(0)
== 3))
+ }.filter(_.files.nonEmpty)
+ }
+ }
+ val partAttr = AttributeReference("part", IntegerType)()
+ val kept = narrowed.planInputPartitionsWithRuntimeFilters(
+ Array(EqualTo(partAttr, Literal(3))))
+
assert(kept.flatMap(_.asInstanceOf[FilePartition].files.map(_.partitionValues.getInt(0)))
+ .distinct.toSeq === Seq(3))
+ // And a filter selecting a directory the override dropped comes back
with nothing.
+ assert(narrowed.planInputPartitionsWithRuntimeFilters(
+ Array(EqualTo(partAttr, Literal(7)))).isEmpty)
+ // Both calls filtered one listing, which is what keeps the plan and
the execution on one
+ // snapshot of the file index.
+ assert(partitionsCalls === 1, s"expected one listing, got
$partitionsCalls")
Review Comment:
**This test does not pin the single listing across both paths that its
comment and the PR description describe.** It calls
`planInputPartitionsWithRuntimeFilters` twice and never
`narrowed.planInputPartitions()`, so reverting FileScan.scala:197 to the pre-PR
`partitions.toArray` keeps it green, even though a real query would then list,
split and pack twice: at planning through `supportsColumnar`, and again at
execution. The second filter (`part = 7`, a directory the override already
dropped) also cannot catch state carried over from the first call, because
narrowing the memo or ANDing successive calls returns empty too.
Could the test call `narrowed.planInputPartitions()` first, then filter
`part = 7` (expect empty) and `part = 3` (expect `Seq(3)`), keeping
`partitionsCalls === 1`?
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/PushDownUtils.scala:
##########
@@ -230,48 +231,64 @@ object PushDownUtils extends Logging {
translatedFiltersPushed || partPredicatesPushed
- case catalystScan: SupportsRuntimeCatalystFiltering if
pushableFilters.nonEmpty =>
- // A runtime filter is normally evaluated twice: the source prunes
with it, and the
- // FilterExec above the scan applies it again. The two have to agree,
so this screen
- // pushes only predicates the source can be trusted to evaluate in
Spark's place.
- //
- // But pushing a non-deterministic one would let the source prune on
its own coin flip and
- // Spark flip again for the rows that survive, so we handle that
specially.
- //
- // Not pushing a filter is safe only while its FilterExec still
evaluates it. A fully
- // pushed filter has none, so the source is its only evaluator. That
is why
- // DataSourceV2Strategy deletes a FilterExec at planning time only for
a filter we push
- // below, and we use the same method (isPushablePartitionFilter) to
determine if we
- // should push it.
- //
- // A DPP filter degrades to TrueLiteral once its subquery is pruned
away, so it matches
- // every row. The V2 path above drops these implicitly, since
translateRuntimeFilterV2
- // returns None; here we push Catalyst expressions directly, so we
remove them explicitly.
- val catalystFilters = pushableFilters
- .flatMap(unwrapRuntimeFilterExpression)
- .filterNot(_ == Literal.TrueLiteral)
- if (catalystFilters.nonEmpty) {
- catalystScan.filter(catalystFilters.toArray)
- true
- } else {
- false
- }
-
case _ =>
false
}
}
+ /**
+ * Screens runtime filters for a [[SupportsRuntimeCatalystFiltering]] scan
and unwraps them to the
+ * Catalyst predicates the scan should plan with. No translation to
connector predicates happens
+ * here: what a scan can apply is the scan's own business.
+ *
+ * Only filters over attributes the scan declared filterable are kept.
`DataSourceV2Strategy`
+ * routes a dynamic pruning filter without checking that declaration, and
the key of a filter
+ * inserted over one union branch is pushed positionally into the others, so
a scan can be handed
+ * a filter over a column it never declared. Dropping it costs pruning and
not correctness: its
+ * rows are filtered anyway, by the join a DPP filter was derived from, and
by the rewrite
+ * re-applying its own condition for a row-level operation's group filter.
For a filter derived
+ * from a scalar subquery this drops nothing: the strategy already required
its references to be a
+ * subset of the same declaration.
+ *
+ * The determinism screen is the other half. A scalar subquery filter is
evaluated twice, by the
+ * source and by the FilterExec above the scan, so pushing a
non-deterministic one would let the
+ * two disagree. `DataSourceV2Strategy` deletes that FilterExec only for a
filter this method
+ * keeps, which is why both sites decide with `isPushablePartitionFilter`.
+ *
+ * A DPP filter degrades to TrueLiteral once its subquery is pruned away, so
it matches every row.
+ * The V2 path drops these implicitly, since translateRuntimeFilterV2
returns None; here the
+ * expressions reach the scan directly, so we remove them explicitly.
+ */
+ private def catalystRuntimeFilters(
+ scan: SupportsRuntimeCatalystFiltering,
+ runtimeFilters: Seq[Expression],
+ output: Seq[AttributeReference]): Seq[Expression] = {
+ val filterAttrs = V2ExpressionUtils.resolveDataSourceRuntimeFilterRefs(
+ scan.filterAttributes(),
+ output,
+ "filterAttributes()",
+ scan.getClass.getName)
+ runtimeFilters
+ .filter(f => f.references.subsetOf(filterAttrs))
Review Comment:
Minor: the union case is handled here at execution, but
DataSourceV2Strategy.scala:201 could screen it at planning, the same way
DataSourceV2Strategy.scala:187 already screens scalar-subquery filters by
`relation.runtimeFilterAttrs`. As it is, the `fact_other` scan in the new union
test keeps a live `dynamicpruningexpression(other IN dynamicpruning#..)`. It
shows in EXPLAIN on a scan with no partition columns, it takes part in
`equals`/`doCanonicalize` so the node no longer matches an otherwise identical
scan for reuse, and its subquery is still planned. The new test then pins that
the filter reaches the branch (`unusable.size == 1`). Could the strategy filter
`dynamicFilters` by `relation.runtimeFilterAttrs` for
`SupportsRuntimeCatalystFiltering` scans, keeping this line as a defensive
check and flipping that assertion?
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/FileScan.scala:
##########
@@ -182,8 +185,80 @@ trait FileScan extends Scan
FilePartition.getFilePartitions(sparkSession, splitFiles, maxSplitBytes)
}
+ /**
+ * The partitions `partitions` planned, computed once per scan instance.
Both the plain read path
+ * and the runtime-filter path go through this, so a scan node that gets
runtime filters lists the
+ * files once and filters that listing, rather than reading the index twice
and risking two
+ * snapshots. A subclass still customizes `partitions`.
+ */
+ @transient private lazy val plannedPartitions: Seq[FilePartition] =
partitions
+
override def planInputPartitions(): Array[InputPartition] = {
- partitions.toArray
+ plannedPartitions.toArray
+ }
+
+ /**
+ * The partition columns Spark can derive a runtime filter on (SPARK-30628),
restricted to the
+ * ones `readSchema()` still exposes: a reference missing from the scan
relation output fails to
+ * resolve, and a pushed-down aggregate keeps only the partition columns it
groups by.
+ *
+ * A filter over one of them is evaluated against each file's partition
values, so the scan
+ * evaluates it in full and Spark does not evaluate it again after the scan
--
+ * `FileScanBuilder.pushFilters` already keeps compile-time partition
filters out of the post-scan
+ * filters for the same reason.
+ */
+ override def filterAttributes(): Array[NamedReference] = {
Review Comment:
Minor: with partition columns declared here, `PartitionPruning` now inserts
DPP for V2 file scans, but the SPARK-38148 rule in
`CleanupDynamicPruningFilters` cannot see a static `part = <literal>` on the
same key, because `FileScanBuilder` pushes it fully into the scan instead of
leaving it in the `Filter`. So `SELECT /*+ broadcast(d) */ f.id FROM fact f
JOIN dim d ON f.part = d.dim_id WHERE f.part = 3 AND d.dim_val = 7` keeps a DPP
that can prune nothing, where V1 removes it. That costs a
`SubqueryBroadcastExec` and a re-filter and re-pack of partition 3, and, with
`reuseBroadcastOnly=false` on a non-broadcast join, a second run of the dim
side. Could the rule's `DataSourceV2ScanRelation` case also read equalities
from `DataSourceV2ScanRelation.pushedFilters`, with the SPARK-38148 cases run
against V2 parquet?
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2CatalystRuntimeFilterSuite.scala:
##########
@@ -862,21 +883,17 @@ private class PartitioningBreakingScan(
afterFilter: Seq[InputPartition])
extends Scan with Batch with SupportsRuntimeCatalystFiltering {
- private var filtered = false
-
override def readSchema(): StructType = new StructType().add("part",
IntegerType)
override def toBatch: Batch = this
- override def planInputPartitions(): Array[InputPartition] =
- if (filtered) afterFilter.toArray else initialPartitions.toArray
+ override def planInputPartitions(): Array[InputPartition] =
initialPartitions.toArray
Review Comment:
Nit: after this change nothing reaches `initialPartitions`, the `Batch`
mix-in, `toBatch`, `planInputPartitions()` or `createReaderFactory()`. The only
caller goes through the Catalyst branch of `replanWithRuntimeFilters`, which
calls only `filterAttributes()` and `planInputPartitionsWithRuntimeFilters`,
and passes `originalPartitions = Seq.empty`. Could this become
`PartitioningBreakingScan(afterFilter: Seq[InputPartition]) extends
SupportsRuntimeCatalystFiltering` like the other scans in this file, dropping
`Batch` and `PartitionReaderFactory` from the imports?
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/FileScan.scala:
##########
@@ -182,8 +185,80 @@ trait FileScan extends Scan
FilePartition.getFilePartitions(sparkSession, splitFiles, maxSplitBytes)
}
+ /**
+ * The partitions `partitions` planned, computed once per scan instance.
Both the plain read path
+ * and the runtime-filter path go through this, so a scan node that gets
runtime filters lists the
+ * files once and filters that listing, rather than reading the index twice
and risking two
+ * snapshots. A subclass still customizes `partitions`.
+ */
+ @transient private lazy val plannedPartitions: Seq[FilePartition] =
partitions
+
override def planInputPartitions(): Array[InputPartition] = {
- partitions.toArray
+ plannedPartitions.toArray
+ }
+
+ /**
+ * The partition columns Spark can derive a runtime filter on (SPARK-30628),
restricted to the
+ * ones `readSchema()` still exposes: a reference missing from the scan
relation output fails to
+ * resolve, and a pushed-down aggregate keeps only the partition columns it
groups by.
+ *
+ * A filter over one of them is evaluated against each file's partition
values, so the scan
+ * evaluates it in full and Spark does not evaluate it again after the scan
--
+ * `FileScanBuilder.pushFilters` already keeps compile-time partition
filters out of the post-scan
+ * filters for the same reason.
+ */
+ override def filterAttributes(): Array[NamedReference] = {
Review Comment:
Minor: the new DPP also changes which runtime filter V2 file scans get on
shuffle joins. `PartitionPruning` inserts it with `onlyInBroadcast = true` by
default, `InjectRuntimeFilter` then skips the bloom filter for that key
(`!hasDynamicPruningSubquery(...)`), and on a sort-merge join
`PlanDynamicPruningFilters` turns the DPP into `true`. A V2 query that used to
get `might_contain(bloom, xxhash64(part))` before the shuffle now gets no
runtime filter at all. This matches V1, and the real fix belongs in
`InjectRuntimeFilter`, but could the user-facing section of the description
mention it?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [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]
