szehon-ho commented on code in PR #57727:
URL: https://github.com/apache/spark/pull/57727#discussion_r3731864450
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Relation.scala:
##########
@@ -196,14 +196,30 @@ case class DataSourceV2ScanRelation(
/**
* Resolved attributes that the scan declares for runtime filtering via
- * [[SupportsRuntimeV2Filtering.filterAttributes]]. Empty when the scan
- * does not implement [[SupportsRuntimeV2Filtering]] or exposes no
attributes.
+ * [[SupportsRuntimeV2Filtering.filterAttributes]] or
+ * [[SupportsRuntimeCatalystFiltering.filterAttributes]]. Empty when the scan
+ * implements neither interface or exposes no attributes.
*/
- lazy val runtimeFilterAttrs: AttributeSet = scan match {
- case s: SupportsRuntimeV2Filtering =>
- AttributeSet(V2ExpressionUtils.resolveRefs[Attribute](
- s.filterAttributes.toImmutableArraySeq, this))
- case _ => AttributeSet.empty
+ lazy val runtimeFilterAttrs: AttributeSet = {
+ val filterAttrs = scan match {
+ case s: SupportsRuntimeV2Filtering => s.filterAttributes
+ case s: SupportsRuntimeCatalystFiltering => s.filterAttributes()
+ case _ => Array.empty[NamedReference]
+ }
+ AttributeSet(V2ExpressionUtils.resolveRefs[Attribute](
+ filterAttrs.toImmutableArraySeq, this))
+ }
+
+ /**
+ * Resolved attributes for which a Catalyst runtime-filtering scan fully
evaluates predicates.
+ */
+ lazy val fullyPushedRuntimeFilterAttrs: AttributeSet = {
+ val filterAttrs = scan match {
+ case s: SupportsRuntimeCatalystFiltering =>
s.fullyPushedFilterAttributes()
Review Comment:
Went further than the suggestion: rather than making the precedence uniform,
a scan implementing both interfaces is now rejected outright.
`DataSourceV2ScanRelation` checks it from both `runtimeFilterAttrs` and
`fullyPushedRuntimeFilterAttrs`, and `pushRuntimeFilters` carries the same
rejection as its first arm, so all three dispatch sites are covered.
Uniform precedence would make the plan safe, but it would also mean silently
ignoring `fullyPushedFilterAttributes()` for such a scan. Since the contract
already says only one runtime filtering interface may be implemented, and this
is the place where Spark's behaviour depends on it, failing is more useful than
picking one and moving on.
New test: "scan implementing both runtime filtering interfaces -> rejected".
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/connector/SupportsRuntimeCatalystFiltering.scala:
##########
@@ -0,0 +1,83 @@
+/*
+ * 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.internal.connector
+
+import org.apache.spark.sql.catalyst.expressions.Expression
+import org.apache.spark.sql.connector.expressions.NamedReference
+import org.apache.spark.sql.connector.read.Scan
+
+/**
+ * A mix-in interface for [[Scan]]. Data sources can implement this interface
if they can
+ * filter initially planned
[[org.apache.spark.sql.connector.read.InputPartition]]s using
+ * Catalyst [[Expression]]s Spark infers at runtime.
+ * Only one runtime filtering interface should be implemented by a data source.
+ *
+ * Spark considers a runtime predicate fully pushed when all attributes
referenced by the
+ * predicate are returned by [[fullyPushedFilterAttributes]]. Fully pushed
predicates are not
+ * evaluated again after the scan.
+ *
+ * Note that Spark will push runtime filters only if they are beneficial.
+ */
+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 be a top-level attribute present in
[[Scan.readSchema]]. Nested
+ * references and attributes pruned out of the read schema fail to resolve
when Spark builds
+ * the scan relation.
+ */
+ def filterAttributes(): Array[NamedReference]
+
+ /**
+ * Returns attributes for which this scan fully evaluates runtime predicates.
+ *
+ * Any runtime predicate that references only attributes in this set is
considered fully pushed
+ * and will not be evaluated again after the scan. These attributes must
also be returned by
+ * [[filterAttributes]].
+ *
+ * Each reference must be a top-level attribute present in
[[Scan.readSchema]]. Nested
+ * references and attributes pruned out of the read schema fail to resolve
when Spark builds
+ * the scan relation.
+ */
+ def fullyPushedFilterAttributes(): Array[NamedReference] = Array.empty
+
+ /**
+ * Filters this scan using runtime Catalyst expressions.
+ *
+ * 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.
+ *
+ * If the scan also implements
+ * [[org.apache.spark.sql.connector.read.SupportsReportPartitioning]], it
must preserve
+ * the originally reported partitioning during runtime filtering. While
applying runtime
+ * predicates, the scan may detect that some
+ * [[org.apache.spark.sql.connector.read.InputPartition]]s have no matching
data, in which
+ * case it can either replace the initially planned
+ * [[org.apache.spark.sql.connector.read.InputPartition]]s that have no
matching data with
+ * empty [[org.apache.spark.sql.connector.read.InputPartition]]s, or report
only a subset of
+ * the original partition values (omitting those with no data) via
+ * [[org.apache.spark.sql.connector.read.Batch#planInputPartitions]]. The
scan must not
+ * report new partition values that were not present in the original
partitioning.
+ *
+ * Note that Spark will call [[Scan.toBatch]] again after filtering the scan
at runtime.
+ */
+ def filter(expressions: Array[Expression]): Unit
Review Comment:
Applied verbatim, thanks -- the multi-call paragraph is on `filter()` now.
Also corrected the stale notes you pointed at, since they are what would
have stopped a reader from relying on it: `pushRuntimeFilters`' scaladoc now
says successive calls are additive, and the "Do not call multiple times for the
same `scan` instance" bullet on `replanWithRuntimeFilters` says the same
instead of the opposite.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/PushDownUtils.scala:
##########
@@ -218,6 +222,26 @@ object PushDownUtils extends Logging {
}
translatedFiltersPushed || partPredicatesPushed
+
+ case catalystScan: SupportsRuntimeCatalystFiltering if
runtimeFilters.nonEmpty =>
+ // A DPP filter degrades to TrueLiteral when its subquery is pruned
away; it carries no
+ // information for the source. The V2 path above drops these
implicitly because
+ // translateRuntimeFilterV2 returns None; here we push Catalyst
expressions directly,
+ // so filter them out explicitly.
+ // Screen with the same pushability guard as the V2 PartitionPredicate
path
+ // (deterministic, no subquery, no Python UDF). Keeps
non-deterministic filters
+ // from being the sole evaluator when fullyPushedFilterAttributes
drops FilterExec.
+ val catalystFilters = runtimeFilters
+ .flatMap(unwrapRuntimeFilterExpression)
Review Comment:
Took the smaller fix. `filter()` now says an expression may access nested
fields of an attribute returned by `filterAttributes()`, as that attribute is
required to be top-level, and that the scan is responsible for matching such
accesses against its own partition layout.
Flattening here would also rewrite attribute references in the flat case,
which is more than this branch needs, and the whole point of the Catalyst path
is that the expression reaches the source unchanged -- a source on this
interface is already binding and interpreting expressions, so `GetStructField`
is not a shape it needs Spark to normalize away.
--
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]
