szehon-ho commented on code in PR #57727:
URL: https://github.com/apache/spark/pull/57727#discussion_r3752496622
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryBaseTable.scala:
##########
@@ -691,6 +703,64 @@ abstract class InMemoryBaseTable(
new InMemoryMicroBatchStream(readSchema, tableSchema)
}
+ /**
+ * Reference implementation of [[SupportsRuntimeCatalystFiltering.filter]]
for the in-memory
+ * fixtures: records what was pushed, and for expressions referencing only
partition columns
+ * binds them against the partition key and drops partitions that do not
match. Binding and
+ * interpreting rather than pattern matching a fixed set of operators is
what lets the fixture
+ * honor an arbitrary pushed expression, the same way
`PartitionPredicateImpl` does. Mixing
+ * classes supply their own `filterAttributes()`.
+ */
+ trait CatalystRuntimeFilteringScan extends SupportsRuntimeCatalystFiltering {
+ self: BatchScanBaseClass =>
+
+ /** The full table schema, used to locate partition columns pruned out of
`readSchema`. */
+ protected def tableSchema: StructType
+
+ private val catalystPredicates = ArrayBuffer.empty[CatalystExpression]
+
+ override def filter(expressions: Array[CatalystExpression]): Unit = {
+ catalystPredicates ++= expressions
+ val partAttrs = partitionAttributes
+ if (partAttrs.isEmpty) return
+
+ val resolver = SQLConf.get.resolver
+ expressions.foreach { expr =>
+ val remapped = expr.transform {
+ case a: AttributeReference =>
+ partAttrs.find(p => resolver(p.name, a.name)).getOrElse(a)
+ }
+ // Only evaluate expressions whose refs are all partition columns, so
we can bind
+ // against the partition key InternalRow (same approach as
PartitionPredicateImpl).
+ if (remapped.references.forall(r => partAttrs.exists(_.exprId ==
r.exprId))) {
+ val bound = BindReferences.bindReference(remapped, partAttrs)
+ val pred = CatalystPredicate.createInterpreted(bound)
+ self.data = self.data.filter { p =>
+ try {
+ pred.eval(p.asInstanceOf[BufferedRows].partitionKey())
+ } catch {
+ // Keep the partition on eval failure, matching
PartitionPredicateImpl.
Review Comment:
Applied, with the reason spelled out rather than the bare precedent. The
comment now says why `PartitionPredicateImpl`'s fail-open does not carry over:
Spark keeps the post-scan `FilterExec` on that path, so failing open costs a
pruning opportunity, while a scan declaring an attribute in
`fullyPushedFilterAttributes()` stands alone as the evaluator.
I took the optional second half too. Your finding 2 argument was that "fully
evaluates" carries the shape requirement, and I still think it does, but it
says nothing about what happens when evaluation *fails*, and as you note the
predicate is an arbitrary expression Spark neither translated nor
capability-checked, so failure is reachable. `fullyPushedFilterAttributes()`
now says the scan must return only partitions it has proven satisfy the
predicate, names the ways evaluating one can fail (an ANSI cast or overflow
error, or a nested access the scan matches differently against its partition
layout), and says to declare an attribute only when the scan can evaluate every
predicate over it.
I left the fixture failing open rather than rethrowing. Nothing in it can
throw today, so the enforcement would be untested code in a test helper; the
comment now explains why fail-open is safe *here* instead of implying it is
sanctioned generally.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/connector/SupportsRuntimeCatalystFiltering.scala:
##########
@@ -0,0 +1,94 @@
+/*
+ * 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 attribute's value must therefore be fixed
within every
+ * [[org.apache.spark.sql.connector.read.InputPartition]] the scan returns,
since pruning
+ * partitions cannot fully evaluate a predicate on a column that varies
within a partition.
+ *
+ * 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.
+ *
+ * An expression may access nested fields of an attribute returned by
[[filterAttributes]], as
+ * that attribute is required to be top-level. The scan is responsible for
matching such
+ * accesses against its own partition layout.
+ *
+ * Spark may call this method more than once for the same scan instance: a
plan can hold several
Review Comment:
Applied verbatim on `SupportsRuntimeV2Filtering.filter`.
Agreed it is pre-existing rather than this PR's bug, but this PR is what
made the internal and public notes disagree: after finding 10 the two
`PushDownUtils` scaladocs say successive calls are additive while the published
interface still gates the multi-call case on `supportsIterativePushdown()`.
Leaving the public one as the stale copy would be the worse outcome, since that
is the page an implementor actually reads.
--
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]
