dongjoon-hyun commented on code in PR #55579:
URL: https://github.com/apache/spark/pull/55579#discussion_r4503842447
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/PushDownUtils.scala:
##########
@@ -230,48 +238,92 @@ 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 catalystScan: SupportsRuntimeCatalystFiltering
+ if runtimeFilters.nonEmpty &&
implementsOnlyDeprecatedFilter(catalystScan) =>
+ filterWithDeprecatedMethod(catalystScan, runtimeFilters, output)
+
+ case catalystScan: SupportsRuntimeCatalystFiltering =>
+ // Resolved only when a filter could be fully pushed, so a node with
no such filter does
+ // not ask the scan.
+ lazy val fullyPushedAttrs =
V2ExpressionUtils.resolveDataSourceRuntimeFilterRefs(
+ catalystScan.fullyPushedFilterAttributes(),
+ output,
+ "fullyPushedFilterAttributes()",
+ catalystScan.getClass.getName)
+ val evaluatedOnlyByScan =
+ runtimeFilters.filter(isFullyPushedRuntimeFilter(_,
fullyPushedAttrs))
+ if (evaluatedOnlyByScan.nonEmpty) {
+ throw SparkException.internalError(
+ s"${scan.getClass.getName} fully evaluates the runtime filters " +
+ s"${evaluatedOnlyByScan.mkString(", ")}, so they must go through "
+
+ "PushDownUtils.replanWithRuntimeFilters, not pushRuntimeFilters.")
}
+ false
case _ =>
false
}
}
+ /**
+ * Screens runtime filters for a [[SupportsRuntimeCatalystFiltering]] scan
and unwraps them to the
+ * Catalyst predicates the scan plans with, untranslated.
+ *
+ * A filter is kept only if it references attributes the scan declared
filterable and passes
+ * `isPushablePartitionFilter`, the check `DataSourceV2Strategy` uses to
drop a pushed filter's
+ * `FilterExec`. A DPP filter degraded to true is removed, as the V2 path
does by not translating
+ * it.
+ */
+ 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))
+ .filter(isPushablePartitionFilter(_, includeSubquery = true))
+ .flatMap(unwrapRuntimeFilterExpression)
+ .filterNot(_ == Literal.TrueLiteral)
+ }
+
+ /**
+ * Whether `scan` gets `planInputPartitionsWithRuntimeFilters` from the
default in
+ * [[SupportsRuntimeCatalystFiltering]], such as a class compiled against
Spark 4.3 or a Java
+ * class implementing the trait directly that overrides only `filter`. A
Scala class that mixes in
+ * the trait gets a forwarder, so it and its subclasses take the general
path.
+ */
+ private def implementsOnlyDeprecatedFilter(scan:
SupportsRuntimeCatalystFiltering): Boolean =
+ scan.getClass.getMethod("planInputPartitionsWithRuntimeFilters",
classOf[Array[Expression]])
Review Comment:
**26. [Low] `implementsOnlyDeprecatedFilter` can fail a query with
`NoClassDefFoundError`**
`getMethod` builds the public methods of each class it searches, starting
with the scan's own class, and that resolves every type in their signatures. So
a `SupportsRuntimeCatalystFiltering` scan with a public method over a type that
is missing at runtime, e.g. one from an optional dependency, now fails every
query that has a runtime filter on it with `NoClassDefFoundError` at L242 or
L356, while 4.3.0 called `filter` without reflection and ran the query. The
lookup also runs again on every execution of every such scan node.
`CodeGenerator.leavesGenCodeToExpression` (CodeGenerator.scala:2929-2936) makes
the same inherited-default check behind a `ClassValue` cache and `catch { case
_: LinkageError => false }`. Could this do the same? With `false` such a scan
takes the general path, whose default still calls `filter`.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Strategy.scala:
##########
@@ -226,22 +227,29 @@ class DataSourceV2Strategy(session: SparkSession) extends
Strategy with Predicat
}
// Screen with the same test pushdown applies, or a filter dropped here
and rejected there
// would be evaluated nowhere.
- val fullyPushedRuntimeFilters = scalarSubqueryFilters.filter { f =>
- f.references.subsetOf(relation.fullyPushedRuntimeFilterAttrs) &&
- PushDownUtils.isPushablePartitionFilter(f, includeSubquery = true)
- }
+ val fullyPushedRuntimeFilters = scalarSubqueryFilters.filter(
+ PushDownUtils.isFullyPushedRuntimeFilter(_,
relation.fullyPushedRuntimeFilterAttrs))
// dynamicFilters need no such check: a DynamicPruningSubquery over a
non-deterministic
// filtering plan is itself non-deterministic, so
CleanupDynamicPruningFilters has already
// rewritten it to TrueLiteral by the time we get here.
+ // A Catalyst runtime-filtering scan only applies a filter over
attributes it declared, so a
+ // DPP filter over any other column is dropped here rather than planned
for nothing. One
+ // PartitionPruning inserted over a union lands in every branch,
including one that cannot
+ // use it; the join it was derived from still applies it.
+ val usableDynamicFilters = relation.scan match {
+ case _: SupportsRuntimeCatalystFiltering =>
+
dynamicFilters.filter(_.references.subsetOf(relation.runtimeFilterAttrs))
Review Comment:
**31. [Nit] An unpartitioned branch of a union gets no runtime filter when
the DPP filter is kept**
When `PartitionPruning` inserts a DPP filter over a union of a partitioned
V2 file table and an unpartitioned one, `InjectRuntimeFilter` finds it on the
key and adds no bloom filter (InjectRuntimeFilter.scala:423), the filter then
lands in both branches (L236-238), and this line drops the copy in the
unpartitioned branch. When that DPP filter is kept, e.g. as its own subquery
with `reuseBroadcastOnly=false`, the unpartitioned branch gets no runtime
filter, while on master, where a V2 file scan gets no DPP filter, the bloom
filter goes into both branches where its checks pass. SPARK-60702 does not
change this, since such a DPP filter counts as kept. The results are the same.
Could the description list it with the bloom-filter bullet?
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/dynamicpruning/PartitionPruning.scala:
##########
@@ -233,6 +233,10 @@ object PartitionPruning extends Rule[LogicalPlan] with
PredicateHelper with Join
private def hasSelectivePredicate(plan: LogicalPlan): Boolean = plan.exists {
case f: Filter => isLikelySelective(f.condition)
+ // A built-in V2 file scan takes its partition filters in without leaving
a `Filter` above it,
Review Comment:
**29. [Low] A V2 file dim filtered only on its partition column cannot be a
bloom filter's creation side**
Item 12 fixed this for DPP, but
`InjectRuntimeFilter.extractSelectiveFilterOverScan` makes the same check for a
bloom filter's creation side: it counts a selective predicate only at a
`Filter` (InjectRuntimeFilter.scala:140) and accepts a leaf only after one
(L237). A V2 file dim filtered only on its partition column keeps no `Filter`
above its scan, so it cannot be the creation side, while V1 can, since
`PruneFileSourcePartitions` keeps the `Filter`. This already holds on master
for any DSv2 dim, but it matters more after merging master. With SPARK-60702, a
key whose `onlyInBroadcast` DPP filter physical planning drops now gets a bloom
filter when this path finds a creation side. So with the default
`reuseBroadcastOnly` on a shuffle join, the query of "DPP for a dimension
filtered only on its partition column, as in V1" would get a bloom filter in
V1, where its size checks pass, and no runtime filter at all in V2.
`InjectRuntimeFilter` is in sql/catalyst and cannot see this trait
, but it could read `DataSourceV2ScanRelation.pushedFilters`, as the check
below does. InjectRuntimeFilter.scala is outside this diff, so I left this
here. Could it be a follow-up JIRA, or listed in the description next to the
bloom-filter bullet?
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/RuntimePartitionPruningFileScan.scala:
##########
@@ -0,0 +1,108 @@
+/*
+ * 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.datasources.v2
+
+import java.util.Locale
+
+import scala.util.control.NonFatal
+
+import org.apache.spark.SparkException
+import org.apache.spark.sql.catalyst.expressions.{Attribute, BoundReference,
Expression, Predicate}
+import org.apache.spark.sql.connector.expressions.{FieldReference,
NamedReference}
+import org.apache.spark.sql.connector.read.InputPartition
+import org.apache.spark.sql.execution.datasources.FilePartition
+import org.apache.spark.sql.internal.connector.SupportsRuntimeCatalystFiltering
+import org.apache.spark.util.TransientBestEffortLazyVal
+
+/**
+ * A [[FileScan]] that prunes its partition directories with the runtime
filters Spark derives over
+ * its partition columns (SPARK-30628). The built-in file formats mix this in.
+ *
+ * Both read paths start from one listing of the file index and split and pack
the files they
+ * keep through `filePartitions`, so `partitions` and `planInputPartitions()`
are final here. A
+ * scan that reports a partitioning or an ordering should not mix this in.
+ */
+trait RuntimePartitionPruningFileScan extends FileScan with
SupportsRuntimeCatalystFiltering {
+
+ /**
+ * The file index listing and its packing, each taken once per scan so
planning and execution
+ * read one snapshot. A `lazy val` would compute them while holding the
scan's monitor.
+ */
+ private val listing = new TransientBestEffortLazyVal(() =>
+ fileIndex.listFiles(partitionFilters, dataFilters))
+
+ private val allPartitions = new TransientBestEffortLazyVal(() =>
filePartitions(listing()))
+
+ // Final so that a subclass of a built-in scan cannot change one read path
without the other.
+ final override protected def partitions: Seq[FilePartition] = allPartitions()
+
+ final override def planInputPartitions(): Array[InputPartition] =
allPartitions().toArray
+
+ /**
+ * The read partition columns that `readSchema()` still exposes; a
pushed-down aggregate keeps
+ * only the ones it groups by. None is declared fully pushed, so a scalar
subquery filter over
+ * them is evaluated again above the scan.
+ */
+ override def filterAttributes(): Array[NamedReference] = {
+ val readFields = readSchema().fieldNames.map(normalizePartitionName).toSet
+ readPartitionSchema.fieldNames
+ .filter(name => readFields.contains(normalizePartitionName(name)))
+ .map(FieldReference.column)
+ }
+
+ /**
+ * Keeps the listed partition directories whose values satisfy
`expressions`, then splits and
+ * packs their files for their own size, as V1 does for
`dynamicallySelectedPartitions`.
+ *
+ * A filter that fails to evaluate on a directory keeps it, since the join
or the `FilterExec`
+ * above the scan applies the filter again.
+ */
+ override def planInputPartitionsWithRuntimeFilters(
Review Comment:
**28. [Nit] `planInputPartitionsWithRuntimeFilters` is still open to a
subclass**
L50 makes `partitions` and `planInputPartitions()` final "so that a subclass
of a built-in scan cannot change one read path without the other", and the
description says such a subclass "can change the packing only by overriding
`filePartitions`". This method is not final, though, so a subclass can still
change the runtime-filtered path alone. Could it be `final override def` as
well? Nothing in-tree overrides it on a built-in scan.
The two finals are also a binary change for such a subclass. scalac emits
them as `final` forwarders in each built-in scan class, so a subclass built
against 4.3.0 that overrides `planInputPartitions()` now fails at class loading
("overrides final method"), and a recompiled one fails to compile. MiMa does
not report it, since `org.apache.spark.sql.execution.*` is excluded
(MimaExcludes.scala:191). Could the user-facing section mention both finals?
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/RuntimePartitionPruningFileScan.scala:
##########
@@ -0,0 +1,108 @@
+/*
+ * 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.datasources.v2
+
+import java.util.Locale
+
+import scala.util.control.NonFatal
+
+import org.apache.spark.SparkException
+import org.apache.spark.sql.catalyst.expressions.{Attribute, BoundReference,
Expression, Predicate}
+import org.apache.spark.sql.connector.expressions.{FieldReference,
NamedReference}
+import org.apache.spark.sql.connector.read.InputPartition
+import org.apache.spark.sql.execution.datasources.FilePartition
+import org.apache.spark.sql.internal.connector.SupportsRuntimeCatalystFiltering
+import org.apache.spark.util.TransientBestEffortLazyVal
+
+/**
+ * A [[FileScan]] that prunes its partition directories with the runtime
filters Spark derives over
+ * its partition columns (SPARK-30628). The built-in file formats mix this in.
+ *
+ * Both read paths start from one listing of the file index and split and pack
the files they
+ * keep through `filePartitions`, so `partitions` and `planInputPartitions()`
are final here. A
+ * scan that reports a partitioning or an ordering should not mix this in.
+ */
+trait RuntimePartitionPruningFileScan extends FileScan with
SupportsRuntimeCatalystFiltering {
+
+ /**
+ * The file index listing and its packing, each taken once per scan so
planning and execution
+ * read one snapshot. A `lazy val` would compute them while holding the
scan's monitor.
+ */
+ private val listing = new TransientBestEffortLazyVal(() =>
+ fileIndex.listFiles(partitionFilters, dataFilters))
+
+ private val allPartitions = new TransientBestEffortLazyVal(() =>
filePartitions(listing()))
+
+ // Final so that a subclass of a built-in scan cannot change one read path
without the other.
+ final override protected def partitions: Seq[FilePartition] = allPartitions()
+
+ final override def planInputPartitions(): Array[InputPartition] =
allPartitions().toArray
+
+ /**
+ * The read partition columns that `readSchema()` still exposes; a
pushed-down aggregate keeps
+ * only the ones it groups by. None is declared fully pushed, so a scalar
subquery filter over
+ * them is evaluated again above the scan.
+ */
+ override def filterAttributes(): Array[NamedReference] = {
Review Comment:
**24. [Low] A group-based row-level operation on a built-in scan now gets a
group filter that the scan does not record**
With the partition columns declared here,
`RowLevelOperationRuntimeGroupFiltering` injects a group filter into a
group-based (`ReplaceData`) operation whose scan is a built-in file scan, since
it checks only `runtimeFilterAttrs.nonEmpty`
(RowLevelOperationRuntimeGroupFiltering.scala:67-72). The read then gets only
the partition directories that have a matching row. The trait asks a
group-based scan whose write replaces the groups that were read to record what
each call returned (SupportsRuntimeCatalystFiltering.scala:96-98), but this one
records nothing, and `planInputPartitions()` still returns every directory. So
a connector whose `SupportsRowLevelOperations` builds its scan from a built-in
one, and whose write replaces the groups that scan plans, e.g. from
`planInputPartitions()` or the file index, would now rewrite only the matching
partitions of `DELETE FROM t WHERE x = 5` but replace all of them, which loses
the rows of the others. On master the scan has no runtime filter at
tributes, so the rule does nothing and the read covers every group. No in-tree
`FileTable` supports row-level operations, and I did not find such a connector
on GitHub, so this is a contract gap rather than a known break.
Could `canInjectGroupFilters` skip a `RuntimePartitionPruningFileScan`, as
on master, or could the scaladoc at L67-73 say that the scan does not record
what each call returned, so it must not back such a write?
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/PushDownUtils.scala:
##########
@@ -293,11 +347,21 @@ object PushDownUtils extends Logging {
output: Seq[AttributeReference],
keyedPartitioning: Option[KeyedPartitioning],
originalPartitions: => Seq[InputPartition]): Seq[Option[InputPartition]]
= {
- val filtered = pushRuntimeFilters(scan, runtimeFilters, table, output)
+ val (filtered, newPartitions) = scan match {
+ // A Catalyst scan is asked even when screening leaves no filter, unless
it also implements
+ // SupportsRuntimeV2Filtering, which pushRuntimeFilters rejects, or
declares only the
+ // deprecated `filter`, which pushRuntimeFilters calls as Spark 4.3 did.
+ case catalystScan: SupportsRuntimeCatalystFiltering
+ if runtimeFilters.nonEmpty &&
!catalystScan.isInstanceOf[SupportsRuntimeV2Filtering] &&
+ !implementsOnlyDeprecatedFilter(catalystScan) =>
Review Comment:
**25. [Low] Follow-up to my item 10: a Scala scan recompiled against this
trait still plans again when screening removes every runtime filter**
This guard sends a scan to `pushRuntimeFilters` only when its classes do not
declare `planInputPartitionsWithRuntimeFilters`, i.e. one built against 4.3.0
or a Java one (L293-298). A Scala scan recompiled against this trait gets a
mixin forwarder for it, so one that still overrides only `filter`, which the
description says needs only `override` on it, takes this branch, as
PushDownUtils.scala:182-185 says. When screening removes every runtime filter,
the default then skips `filter` but plans again through
`toBatch.planInputPartitions()`
(SupportsRuntimeCatalystFiltering.scala:120-121) instead of reading
`originalPartitions`. That is item 10 again for every such scan recompiled from
now on: one more planning per scan node on a sort-merge join under the default
DPP settings, and, for a scan instance that another node already narrowed
through `filter`, a read of the narrowed set, the SPARK-60752 class from item
14.
The test at DataSourceV2CatalystRuntimeFilterSuite.scala:568-579 cannot see
this. `LegacyFilterScan` is such a recompiled Scala scan, although its doc says
"A scan written against Spark 4.3" (L965), and the test passes
`scan.toBatch.planInputPartitions()` as `originalPartitions`, so a new plan and
the planned partitions agree, and L576 reads as if the scan took the 4.3.0
route.
Reflection sees a forwarder as a declaration, so I do not see a check here
that tells the two apart. Could the scaladoc of `filter`
(SupportsRuntimeCatalystFiltering.scala:124-128) and the description say that
such a scan should implement `planInputPartitionsWithRuntimeFilters`, since it
otherwise plans again when screening removes every runtime filter, and could
L576 and L965 say that this Scala fixture takes the general path?
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/RuntimePartitionPruningFileScan.scala:
##########
@@ -0,0 +1,108 @@
+/*
+ * 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.datasources.v2
+
+import java.util.Locale
+
+import scala.util.control.NonFatal
+
+import org.apache.spark.SparkException
+import org.apache.spark.sql.catalyst.expressions.{Attribute, BoundReference,
Expression, Predicate}
+import org.apache.spark.sql.connector.expressions.{FieldReference,
NamedReference}
+import org.apache.spark.sql.connector.read.InputPartition
+import org.apache.spark.sql.execution.datasources.FilePartition
+import org.apache.spark.sql.internal.connector.SupportsRuntimeCatalystFiltering
+import org.apache.spark.util.TransientBestEffortLazyVal
+
+/**
+ * A [[FileScan]] that prunes its partition directories with the runtime
filters Spark derives over
+ * its partition columns (SPARK-30628). The built-in file formats mix this in.
+ *
+ * Both read paths start from one listing of the file index and split and pack
the files they
+ * keep through `filePartitions`, so `partitions` and `planInputPartitions()`
are final here. A
+ * scan that reports a partitioning or an ordering should not mix this in.
+ */
+trait RuntimePartitionPruningFileScan extends FileScan with
SupportsRuntimeCatalystFiltering {
Review Comment:
**27. [Low] A subclass of a built-in scan that adds
`SupportsRuntimeV2Filtering` now fails every query**
The six built-in scans are case classes that are not final, and through this
trait they now implement `SupportsRuntimeCatalystFiltering`. A subclass that
adds `SupportsRuntimeFiltering` or `SupportsRuntimeV2Filtering`, e.g. `class
MyScan(...) extends ParquetScan(...) with SupportsRuntimeV2Filtering`, then
implements both, and even a plain `SELECT` fails with INTERNAL_ERROR, since
DataSourceV2Strategy.scala:218 reads `runtimeFilterAttrs`, which calls
`checkRuntimeFilteringInterfaces()` (DataSourceV2Relation.scala:225-229 and
392-400). On master such a subclass works, and its runtime filters reach its
own `filter`. The per-format opt-in keeps a `FileScan` subclass as on master,
but not a subclass of a built-in scan. As with item 9, I did not find one on
GitHub, so this is a contract gap rather than a known break. Could the
user-facing section of the description say so, next to the finals of item 28?
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/RuntimePartitionPruningFileScan.scala:
##########
@@ -0,0 +1,108 @@
+/*
+ * 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.datasources.v2
+
+import java.util.Locale
+
+import scala.util.control.NonFatal
+
+import org.apache.spark.SparkException
+import org.apache.spark.sql.catalyst.expressions.{Attribute, BoundReference,
Expression, Predicate}
+import org.apache.spark.sql.connector.expressions.{FieldReference,
NamedReference}
+import org.apache.spark.sql.connector.read.InputPartition
+import org.apache.spark.sql.execution.datasources.FilePartition
+import org.apache.spark.sql.internal.connector.SupportsRuntimeCatalystFiltering
+import org.apache.spark.util.TransientBestEffortLazyVal
+
+/**
+ * A [[FileScan]] that prunes its partition directories with the runtime
filters Spark derives over
+ * its partition columns (SPARK-30628). The built-in file formats mix this in.
+ *
+ * Both read paths start from one listing of the file index and split and pack
the files they
+ * keep through `filePartitions`, so `partitions` and `planInputPartitions()`
are final here. A
+ * scan that reports a partitioning or an ordering should not mix this in.
+ */
+trait RuntimePartitionPruningFileScan extends FileScan with
SupportsRuntimeCatalystFiltering {
+
+ /**
+ * The file index listing and its packing, each taken once per scan so
planning and execution
+ * read one snapshot. A `lazy val` would compute them while holding the
scan's monitor.
+ */
+ private val listing = new TransientBestEffortLazyVal(() =>
Review Comment:
**30. [Low] Every built-in file scan keeps its file listing for the plan's
lifetime**
`listing` holds the `Seq[PartitionDirectory]` from `fileIndex.listFiles` for
as long as the scan lives, on every built-in file scan, including an
unpartitioned table, a scan that reads no partition column and a plan without
runtime filters, none of which reads it again after `allPartitions`.
`listFiles` wraps each file in a new `FileStatusWithMetadata` per call
(FileIndex.scala:103-104), so this keeps an object per file on top of
`allPartitions`, tens of MB for a table with a million files, for as long as a
`QueryExecution` or a cached plan holds the scan. After
`InMemoryFileIndex.refresh()`, a retained scan also keeps the previous
listing's `FileStatus`es alive. The merge-base drops the listing after packing.
Could `listing` be memoized only when `filterAttributes()` is not empty, and
otherwise stay local to `allPartitions`?
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/RuntimePartitionPruningFileScan.scala:
##########
@@ -0,0 +1,108 @@
+/*
+ * 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.datasources.v2
+
+import java.util.Locale
+
+import scala.util.control.NonFatal
+
+import org.apache.spark.SparkException
+import org.apache.spark.sql.catalyst.expressions.{Attribute, BoundReference,
Expression, Predicate}
+import org.apache.spark.sql.connector.expressions.{FieldReference,
NamedReference}
+import org.apache.spark.sql.connector.read.InputPartition
+import org.apache.spark.sql.execution.datasources.FilePartition
+import org.apache.spark.sql.internal.connector.SupportsRuntimeCatalystFiltering
+import org.apache.spark.util.TransientBestEffortLazyVal
+
+/**
+ * A [[FileScan]] that prunes its partition directories with the runtime
filters Spark derives over
+ * its partition columns (SPARK-30628). The built-in file formats mix this in.
+ *
+ * Both read paths start from one listing of the file index and split and pack
the files they
+ * keep through `filePartitions`, so `partitions` and `planInputPartitions()`
are final here. A
+ * scan that reports a partitioning or an ordering should not mix this in.
+ */
+trait RuntimePartitionPruningFileScan extends FileScan with
SupportsRuntimeCatalystFiltering {
+
+ /**
+ * The file index listing and its packing, each taken once per scan so
planning and execution
+ * read one snapshot. A `lazy val` would compute them while holding the
scan's monitor.
+ */
+ private val listing = new TransientBestEffortLazyVal(() =>
+ fileIndex.listFiles(partitionFilters, dataFilters))
+
+ private val allPartitions = new TransientBestEffortLazyVal(() =>
filePartitions(listing()))
+
+ // Final so that a subclass of a built-in scan cannot change one read path
without the other.
+ final override protected def partitions: Seq[FilePartition] = allPartitions()
+
+ final override def planInputPartitions(): Array[InputPartition] =
allPartitions().toArray
+
+ /**
+ * The read partition columns that `readSchema()` still exposes; a
pushed-down aggregate keeps
+ * only the ones it groups by. None is declared fully pushed, so a scalar
subquery filter over
+ * them is evaluated again above the scan.
+ */
+ override def filterAttributes(): Array[NamedReference] = {
+ val readFields = readSchema().fieldNames.map(normalizePartitionName).toSet
+ readPartitionSchema.fieldNames
+ .filter(name => readFields.contains(normalizePartitionName(name)))
+ .map(FieldReference.column)
+ }
+
+ /**
+ * Keeps the listed partition directories whose values satisfy
`expressions`, then splits and
+ * packs their files for their own size, as V1 does for
`dynamicallySelectedPartitions`.
+ *
+ * A filter that fails to evaluate on a directory keeps it, since the join
or the `FilterExec`
+ * above the scan applies the filter again.
+ */
+ override def planInputPartitionsWithRuntimeFilters(
+ expressions: Array[Expression]): Array[InputPartition] = {
+ if (expressions.isEmpty) return allPartitions().toArray
+ // Directory values follow the table's partition schema, not the read one,
and bind as V1
+ // binds them: with that schema's types, nullable.
+ val partitionSchema = fileIndex.partitionSchema
+ val fieldIndex = partitionSchema.fieldNames.zipWithIndex
+ .map { case (name, i) => normalizePartitionName(name) -> i }.toMap
+ val predicates = expressions.toSeq.map { e =>
+ Predicate.createInterpreted(e.transform {
+ case a: Attribute =>
+ // PushDownUtils passes only filters over `filterAttributes()`.
+ val i = fieldIndex.getOrElse(normalizePartitionName(a.name),
+ throw SparkException.internalError("A file scan can only apply a
runtime filter " +
+ s"over its partition columns
${partitionSchema.fieldNames.mkString("[", ", ", "]")}" +
+ s", got $e"))
+ BoundReference(i, partitionSchema(i).dataType, nullable = true)
+ })
+ }
+ val all = listing()
+ val kept = all.filter { dir =>
+ predicates.forall { predicate =>
+ try predicate.eval(dir.values) catch { case NonFatal(_) => true }
+ }
+ }
+ if (kept.length == all.length) allPartitions().toArray else
filePartitions(kept).toArray
Review Comment:
**32. [Nit] A scan that DPP prunes still packs every file at planning**
When a runtime filter keeps few directories, the node packs twice: planning
packs every listed file through `allPartitions`, and this line packs the
survivors. Planning has to, because `FileScan` keeps the default
`columnarSupportMode()`, `PARTITION_DEFINED`, so `supportsColumnar`
(DataSourceV2ScanExecBase.scala:200-210), which DataSourceV2Strategy.scala:275
asks, reads `inputPartitions`. With DPP keeping 10 of 10,000 partitions of a
table with a million files, the driver splits, sorts and packs a million files
at planning only to learn one boolean. V1 packs only
`dynamicallySelectedPartitions`. The merge-base pays the same at planning, so
this is not a regression. Could a follow-up have the built-in scans report
`columnarSupportMode()` themselves? Their reader factories decide columnar
reads from the schema and the conf, not from the partition.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/PushDownUtils.scala:
##########
@@ -230,48 +238,92 @@ 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 catalystScan: SupportsRuntimeCatalystFiltering
+ if runtimeFilters.nonEmpty &&
implementsOnlyDeprecatedFilter(catalystScan) =>
+ filterWithDeprecatedMethod(catalystScan, runtimeFilters, output)
+
+ case catalystScan: SupportsRuntimeCatalystFiltering =>
+ // Resolved only when a filter could be fully pushed, so a node with
no such filter does
+ // not ask the scan.
+ lazy val fullyPushedAttrs =
V2ExpressionUtils.resolveDataSourceRuntimeFilterRefs(
+ catalystScan.fullyPushedFilterAttributes(),
+ output,
+ "fullyPushedFilterAttributes()",
+ catalystScan.getClass.getName)
+ val evaluatedOnlyByScan =
+ runtimeFilters.filter(isFullyPushedRuntimeFilter(_,
fullyPushedAttrs))
+ if (evaluatedOnlyByScan.nonEmpty) {
+ throw SparkException.internalError(
+ s"${scan.getClass.getName} fully evaluates the runtime filters " +
+ s"${evaluatedOnlyByScan.mkString(", ")}, so they must go through "
+
+ "PushDownUtils.replanWithRuntimeFilters, not pushRuntimeFilters.")
}
+ false
case _ =>
false
}
}
+ /**
+ * Screens runtime filters for a [[SupportsRuntimeCatalystFiltering]] scan
and unwraps them to the
+ * Catalyst predicates the scan plans with, untranslated.
+ *
+ * A filter is kept only if it references attributes the scan declared
filterable and passes
+ * `isPushablePartitionFilter`, the check `DataSourceV2Strategy` uses to
drop a pushed filter's
+ * `FilterExec`. A DPP filter degraded to true is removed, as the V2 path
does by not translating
+ * it.
+ */
+ 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:
**36. [Nit] The `filterAttributes()` screen at execution no longer drops
anything Spark planned**
Since 2acbef9 the strategy keeps only the DPP filters over
`relation.runtimeFilterAttrs` for a Catalyst scan
(DataSourceV2Strategy.scala:239-243) and routes only scalar subquery filters
over them (L218-224). `BatchScanExec` is built only at
DataSourceV2Strategy.scala:269, and the copies that DPP planning makes keep the
same columns, so this `subsetOf` cannot drop a filter of a node Spark planned.
It only guards a node built elsewhere, at the cost of resolving
`filterAttributes()` again on each execution. The `true` removal at L290 is
still needed. Could the scaladoc at L268-276 say that the strategy already
drops these? Alternatively, `CleanupDynamicPruningFilters` could drop a DPP
filter whose `pruningKey.references` are not in `r.runtimeFilterAttrs` for
every `DataSourceV2ScanRelation`, which would also keep it from the `filter` of
a `SupportsRuntimeV2Filtering` branch, which gets it on master.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/dynamicpruning/CleanupDynamicPruningFilters.scala:
##########
@@ -79,9 +90,14 @@ object CleanupDynamicPruningFilters extends
Rule[LogicalPlan] with PredicateHelp
case p @ NodeWithOnlyDeterministicProjectAndFilter(
HiveTableRelation(_, _, _, _, _)) =>
removeUnnecessaryDynamicPruningSubquery(p)
- case p @ NodeWithOnlyDeterministicProjectAndFilter(
- _: DataSourceV2ScanRelation) =>
- removeUnnecessaryDynamicPruningSubquery(p)
+ // Only a built-in file scan is known to prune what it plans with a
partition filter it fully
+ // took. Another DSv2 source may apply such a filter row by row and
leave the pruning to DPP.
Review Comment:
**35. [Nit] The reason at L93-94 does not hold for the `Filter` path of the
same rule**
L93-94 keep the pushed path to built-in file scans because another DSv2
source "may apply such a filter row by row and leave the pruning to DPP". But
the `Filter` path of this rule (L62-65), which master already applies to every
`DataSourceV2ScanRelation`, drops the DPP filter for an equality left in a
`Filter` above the scan, which the source did not take and so cannot prune by.
The rule thus trusts a DSv2 source less when it took the equality than when it
did not. Keeping other sources as on master after #60053 is a fine reason by
itself; could the comment say that instead?
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/dynamicpruning/PartitionPruning.scala:
##########
@@ -233,6 +233,10 @@ object PartitionPruning extends Rule[LogicalPlan] with
PredicateHelper with Join
private def hasSelectivePredicate(plan: LogicalPlan): Boolean = plan.exists {
case f: Filter => isLikelySelective(f.condition)
+ // A built-in V2 file scan takes its partition filters in without leaving
a `Filter` above it,
+ // where V1 keeps one.
+ case r: DataSourceV2ScanRelation if
r.scan.isInstanceOf[RuntimePartitionPruningFileScan] =>
Review Comment:
**34. [Nit] Follow-up to my item 12: the filtering side need not be a
built-in file scan**
The `isInstanceOf[RuntimePartitionPruningFileScan]` here came from my
suggestion in item 12, but the filtering side does not need it. This only asks
whether a likely selective predicate restricts the dim, and a predicate that
any DSv2 source took in fully, leaving no `Filter`, does so just as one in a
`Filter` does. The scope in `CleanupDynamicPruningFilters` is about what the
pruned scan guarantees, which does not apply here. So a dim on a connector scan
with a fully pushed `region = 'EU'` still gets no DPP, as on master. Could this
be `case r: DataSourceV2ScanRelation =>
r.pushedFilters.exists(isLikelySelective)`? It is a benefit heuristic, so the
results stay the same, though with `reuseBroadcastOnly=false` it can add DPP
subqueries for such dims.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Strategy.scala:
##########
@@ -226,22 +227,29 @@ class DataSourceV2Strategy(session: SparkSession) extends
Strategy with Predicat
}
// Screen with the same test pushdown applies, or a filter dropped here
and rejected there
// would be evaluated nowhere.
- val fullyPushedRuntimeFilters = scalarSubqueryFilters.filter { f =>
- f.references.subsetOf(relation.fullyPushedRuntimeFilterAttrs) &&
- PushDownUtils.isPushablePartitionFilter(f, includeSubquery = true)
- }
+ val fullyPushedRuntimeFilters = scalarSubqueryFilters.filter(
+ PushDownUtils.isFullyPushedRuntimeFilter(_,
relation.fullyPushedRuntimeFilterAttrs))
// dynamicFilters need no such check: a DynamicPruningSubquery over a
non-deterministic
// filtering plan is itself non-deterministic, so
CleanupDynamicPruningFilters has already
// rewritten it to TrueLiteral by the time we get here.
+ // A Catalyst runtime-filtering scan only applies a filter over
attributes it declared, so a
+ // DPP filter over any other column is dropped here rather than planned
for nothing. One
+ // PartitionPruning inserted over a union lands in every branch,
including one that cannot
+ // use it; the join it was derived from still applies it.
+ val usableDynamicFilters = relation.scan match {
+ case _: SupportsRuntimeCatalystFiltering =>
+
dynamicFilters.filter(_.references.subsetOf(relation.runtimeFilterAttrs))
+ case _ => dynamicFilters
+ }
// A group-based read keeps only the group filter, for the reason above.
PartitionPruning
// plans a DynamicPruningSubquery from the join of one read, so it is a
filter of one read
// too. With an IN subquery it lands on the read of the updated rows of
an UPDATE. With a
// NOT EXISTS subquery the negated condition becomes a semi join, so it
lands on the read
// of the rows to copy over.
val runtimeFilters = if (groupBasedRead) {
- dynamicFilters.filterNot(_.isInstanceOf[DynamicPruningSubquery])
+ usableDynamicFilters.filterNot(_.isInstanceOf[DynamicPruningSubquery])
} else {
- dynamicFilters ++ scalarSubqueryFilters
+ usableDynamicFilters ++ scalarSubqueryFilters
Review Comment:
**33. [Nit] A scalar subquery filter over a partition column is planned
twice**
Since e721ae7 the file scans declare no partition column fully pushed (item
2), so `part = (SELECT ...)` goes into `runtimeFilters` here and also stays in
the `Filter` above (L265), as for a `SupportsRuntimeV2Filtering` scan on
master. Under the defaults, subquery reuse runs it once, but with
`spark.sql.execution.reuseSubquery=false` it runs twice, and if its source
changes in between, the scan prunes with one value while the `Filter` keeps
another. V1 and the merge-base evaluate it once. That is a fair price for item
2, but could the user-facing section say that such a filter now shows up both
in `RuntimeFilters` and in the `Filter` above the scan?
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/RuntimePartitionPruningFileScan.scala:
##########
@@ -0,0 +1,108 @@
+/*
+ * 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.datasources.v2
+
+import java.util.Locale
+
+import scala.util.control.NonFatal
+
+import org.apache.spark.SparkException
+import org.apache.spark.sql.catalyst.expressions.{Attribute, BoundReference,
Expression, Predicate}
+import org.apache.spark.sql.connector.expressions.{FieldReference,
NamedReference}
+import org.apache.spark.sql.connector.read.InputPartition
+import org.apache.spark.sql.execution.datasources.FilePartition
+import org.apache.spark.sql.internal.connector.SupportsRuntimeCatalystFiltering
+import org.apache.spark.util.TransientBestEffortLazyVal
+
+/**
+ * A [[FileScan]] that prunes its partition directories with the runtime
filters Spark derives over
+ * its partition columns (SPARK-30628). The built-in file formats mix this in.
+ *
+ * Both read paths start from one listing of the file index and split and pack
the files they
+ * keep through `filePartitions`, so `partitions` and `planInputPartitions()`
are final here. A
+ * scan that reports a partitioning or an ordering should not mix this in.
+ */
+trait RuntimePartitionPruningFileScan extends FileScan with
SupportsRuntimeCatalystFiltering {
+
+ /**
+ * The file index listing and its packing, each taken once per scan so
planning and execution
+ * read one snapshot. A `lazy val` would compute them while holding the
scan's monitor.
+ */
+ private val listing = new TransientBestEffortLazyVal(() =>
+ fileIndex.listFiles(partitionFilters, dataFilters))
+
+ private val allPartitions = new TransientBestEffortLazyVal(() =>
filePartitions(listing()))
+
+ // Final so that a subclass of a built-in scan cannot change one read path
without the other.
+ final override protected def partitions: Seq[FilePartition] = allPartitions()
+
+ final override def planInputPartitions(): Array[InputPartition] =
allPartitions().toArray
+
+ /**
+ * The read partition columns that `readSchema()` still exposes; a
pushed-down aggregate keeps
+ * only the ones it groups by. None is declared fully pushed, so a scalar
subquery filter over
+ * them is evaluated again above the scan.
+ */
+ override def filterAttributes(): Array[NamedReference] = {
+ val readFields = readSchema().fieldNames.map(normalizePartitionName).toSet
+ readPartitionSchema.fieldNames
+ .filter(name => readFields.contains(normalizePartitionName(name)))
+ .map(FieldReference.column)
+ }
+
+ /**
+ * Keeps the listed partition directories whose values satisfy
`expressions`, then splits and
+ * packs their files for their own size, as V1 does for
`dynamicallySelectedPartitions`.
+ *
+ * A filter that fails to evaluate on a directory keeps it, since the join
or the `FilterExec`
+ * above the scan applies the filter again.
+ */
+ override def planInputPartitionsWithRuntimeFilters(
+ expressions: Array[Expression]): Array[InputPartition] = {
+ if (expressions.isEmpty) return allPartitions().toArray
+ // Directory values follow the table's partition schema, not the read one,
and bind as V1
+ // binds them: with that schema's types, nullable.
+ val partitionSchema = fileIndex.partitionSchema
+ val fieldIndex = partitionSchema.fieldNames.zipWithIndex
+ .map { case (name, i) => normalizePartitionName(name) -> i }.toMap
+ val predicates = expressions.toSeq.map { e =>
+ Predicate.createInterpreted(e.transform {
+ case a: Attribute =>
+ // PushDownUtils passes only filters over `filterAttributes()`.
+ val i = fieldIndex.getOrElse(normalizePartitionName(a.name),
+ throw SparkException.internalError("A file scan can only apply a
runtime filter " +
+ s"over its partition columns
${partitionSchema.fieldNames.mkString("[", ", ", "]")}" +
+ s", got $e"))
+ BoundReference(i, partitionSchema(i).dataType, nullable = true)
+ })
+ }
+ val all = listing()
+ val kept = all.filter { dir =>
+ predicates.forall { predicate =>
+ try predicate.eval(dir.values) catch { case NonFatal(_) => true }
Review Comment:
**38. [Nit] A runtime filter that fails on a directory leaves no log**
Keeping the directory is right, but nothing records that the filter failed,
so a user who sees no pruning has nothing to go on.
`PartitionPredicateImpl.eval` keeps a partition for the same reason and logs a
warning with the expression and the exception
(PartitionPredicateImpl.scala:86-95). Could this log one too, e.g. once per
call rather than per directory? This is also another binder of directory values
next to `PartitioningAwareFileIndex.prunePartitions`,
`FileSourceScanExec.dynamicallySelectedPartitions` and
`PartitionPredicateImpl`. None of them fits as is, so a shared one could wait
for a follow-up.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/RuntimePartitionPruningFileScan.scala:
##########
@@ -0,0 +1,108 @@
+/*
+ * 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.datasources.v2
+
+import java.util.Locale
+
+import scala.util.control.NonFatal
+
+import org.apache.spark.SparkException
+import org.apache.spark.sql.catalyst.expressions.{Attribute, BoundReference,
Expression, Predicate}
+import org.apache.spark.sql.connector.expressions.{FieldReference,
NamedReference}
+import org.apache.spark.sql.connector.read.InputPartition
+import org.apache.spark.sql.execution.datasources.FilePartition
+import org.apache.spark.sql.internal.connector.SupportsRuntimeCatalystFiltering
+import org.apache.spark.util.TransientBestEffortLazyVal
+
+/**
+ * A [[FileScan]] that prunes its partition directories with the runtime
filters Spark derives over
+ * its partition columns (SPARK-30628). The built-in file formats mix this in.
+ *
+ * Both read paths start from one listing of the file index and split and pack
the files they
+ * keep through `filePartitions`, so `partitions` and `planInputPartitions()`
are final here. A
+ * scan that reports a partitioning or an ordering should not mix this in.
+ */
+trait RuntimePartitionPruningFileScan extends FileScan with
SupportsRuntimeCatalystFiltering {
+
+ /**
+ * The file index listing and its packing, each taken once per scan so
planning and execution
+ * read one snapshot. A `lazy val` would compute them while holding the
scan's monitor.
+ */
+ private val listing = new TransientBestEffortLazyVal(() =>
+ fileIndex.listFiles(partitionFilters, dataFilters))
+
+ private val allPartitions = new TransientBestEffortLazyVal(() =>
filePartitions(listing()))
+
+ // Final so that a subclass of a built-in scan cannot change one read path
without the other.
+ final override protected def partitions: Seq[FilePartition] = allPartitions()
+
+ final override def planInputPartitions(): Array[InputPartition] =
allPartitions().toArray
+
+ /**
+ * The read partition columns that `readSchema()` still exposes; a
pushed-down aggregate keeps
+ * only the ones it groups by. None is declared fully pushed, so a scalar
subquery filter over
+ * them is evaluated again above the scan.
+ */
+ override def filterAttributes(): Array[NamedReference] = {
+ val readFields = readSchema().fieldNames.map(normalizePartitionName).toSet
+ readPartitionSchema.fieldNames
+ .filter(name => readFields.contains(normalizePartitionName(name)))
+ .map(FieldReference.column)
+ }
+
+ /**
+ * Keeps the listed partition directories whose values satisfy
`expressions`, then splits and
+ * packs their files for their own size, as V1 does for
`dynamicallySelectedPartitions`.
+ *
+ * A filter that fails to evaluate on a directory keeps it, since the join
or the `FilterExec`
+ * above the scan applies the filter again.
+ */
+ override def planInputPartitionsWithRuntimeFilters(
+ expressions: Array[Expression]): Array[InputPartition] = {
+ if (expressions.isEmpty) return allPartitions().toArray
+ // Directory values follow the table's partition schema, not the read one,
and bind as V1
+ // binds them: with that schema's types, nullable.
+ val partitionSchema = fileIndex.partitionSchema
+ val fieldIndex = partitionSchema.fieldNames.zipWithIndex
+ .map { case (name, i) => normalizePartitionName(name) -> i }.toMap
+ val predicates = expressions.toSeq.map { e =>
+ Predicate.createInterpreted(e.transform {
+ case a: Attribute =>
+ // PushDownUtils passes only filters over `filterAttributes()`.
+ val i = fieldIndex.getOrElse(normalizePartitionName(a.name),
+ throw SparkException.internalError("A file scan can only apply a
runtime filter " +
+ s"over its partition columns
${partitionSchema.fieldNames.mkString("[", ", ", "]")}" +
+ s", got $e"))
+ BoundReference(i, partitionSchema(i).dataType, nullable = true)
+ })
+ }
+ val all = listing()
+ val kept = all.filter { dir =>
+ predicates.forall { predicate =>
+ try predicate.eval(dir.values) catch { case NonFatal(_) => true }
+ }
+ }
+ if (kept.length == all.length) allPartitions().toArray else
filePartitions(kept).toArray
+ }
+
+ // Read when the scan is built, as `FileScan` does.
+ private val isCaseSensitive = conf.caseSensitiveAnalysis
+
+ private def normalizePartitionName(name: String): String = {
Review Comment:
**37. [Nit] Following up on item 6: a helper under a new name avoids the
clash with Kyuubi**
Your reply in item 6 is right that a `protected` or package-private
`normalizeName` would clash with Kyuubi's `private def normalizeName`, but that
comes from reusing the name. A helper under a new name has no such clash, e.g.
in `FileScan`:
```scala
private[v2] def normalizePartitionColumnName(name: String): String =
normalizeName(name)
```
A subclass outside the package neither sees nor overrides it, and as a `def`
it needs no field in a subclass built against 4.3.0. `filePartitions` and this
trait would then share one rule again, instead of two copies of the build-time
`conf.caseSensitiveAnalysis` (L102-107 and FileScan.scala:217-225).
--
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]
