peter-toth commented on code in PR #57727:
URL: https://github.com/apache/spark/pull/57727#discussion_r3734842337
##########
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:
**Finding 16.** This is the one line I'd change in the extracted trait, and
it's about the justification rather than the behaviour.
`PartitionPredicateImpl` is fail-open for a specific reason it states in both
branches — "Including partition in scan result to avoid incorrect filtering"
(`PartitionPredicateImpl.scala:48-68`) — and that reason is that Spark keeps
the post-scan `FilterExec` on that path, so failing open costs a pruning
opportunity and nothing else. `fullyPushedFilterAttributes` is the first thing
in Spark to remove that `FilterExec`, which inverts the trade: for a fully
pushed attribute, keeping a partition you couldn't evaluate silently returns
nonmatching rows.
`InMemoryCatalystRuntimeFilterBatchScan` mixes this trait in, so this is the
fixture behind `predicate on fully pushed filter attributes -> not evaluated
after the scan`. Nothing throws there today and the suite is green — but the
scaladoc above the trait calls it the reference implementation, so this comment
is what tells an adopter that fail-open is sanctioned in a context where it
isn't.
Smallest fix is to stop citing the precedent that doesn't carry over:
```suggestion
} catch {
// Keep the partition on eval failure. Safe here only because
none of the fixture's
// own predicates can throw: a scan that declares an attribute
in
// fullyPushedFilterAttributes() has no post-scan FilterExec
to fall back on, so
// keeping an unevaluated partition would return nonmatching
rows.
case _: Exception => true
```
If you'd rather make the fixture enforce it, having the trait rethrow when
the failing predicate references an attribute the scan reports as fully pushed
would do it, and would fail loudly instead of through `checkAnswer`.
Optional second half, and I'll drop it if you read it as already covered:
one sentence on `fullyPushedFilterAttributes()`
(`SupportsRuntimeCatalystFiltering.scala:53`) saying a scan must not keep a
partition it cannot prove satisfies a predicate over these attributes, and must
not declare the attribute if it may fail to evaluate one. You argued on finding
2 that "fully evaluates" already carries the shape requirement and I accepted
that; the same argument applies here, so this is a judgment call rather than a
gap I'd push on. The reason I think it's worth a line anyway is what the
predicate can be: it is an arbitrary deterministic expression Spark neither
translated nor checked against the source's capabilities, so an ANSI overflow
or cast failure, an arity mismatch against the partition-key row, or an
unresolvable nested access can all end up here — and this commit's own new
sentence on `filter()` explicitly hands sources nested accesses to "match
against its own partition layo
ut".
##########
sql/catalyst/src/main/java/org/apache/spark/sql/connector/read/SupportsRuntimeV2Filtering.java:
##########
@@ -28,8 +28,8 @@
* filter initially planned {@link InputPartition}s using predicates Spark
infers at runtime.
* This interface is very similar to {@link SupportsRuntimeFiltering} except
it uses
* data source V2 {@link Predicate} instead of data source V1 {@link Filter}.
- * {@link SupportsRuntimeV2Filtering} is preferred over {@link
SupportsRuntimeFiltering}
- * and only one of them should be implemented by the data sources.
+ * {@link SupportsRuntimeV2Filtering} is preferred over {@link
SupportsRuntimeFiltering}.
+ * Only one runtime filtering interface should be implemented by a data source.
Review Comment:
**Finding 18.** Dropping this line from `SupportsRuntimeFiltering` resolves
finding 14, thanks — but the argument was about the sentence, not about which
page it sat on, and it applies unchanged to the copy that stayed. A source
implementing `SupportsRuntimeFiltering` implements two runtime filtering
interfaces by inheritance, so read from here the sentence is still not
followable.
It also now understates the code:
`DataSourceV2ScanRelation.checkRuntimeFilteringInterfaces` and the first arm of
`PushDownUtils.pushRuntimeFilters` throw `SparkException.internalError`, so
this is not advice any more. Naming the actual rule fixes both, and keeps it
true for a `SupportsRuntimeFiltering` implementer:
```suggestion
* {@link SupportsRuntimeV2Filtering} is preferred over {@link
SupportsRuntimeFiltering}.
* A scan must not implement SupportsRuntimeCatalystFiltering together with
this interface;
* Spark rejects such a scan.
```
(Plain text rather than a `{@link}` for the same reason as finding 14 —
`SupportsRuntimeCatalystFiltering` is internal and absent from the published
Javadoc, which is how the new trait's own scaladoc refers to it too.)
##########
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:
**Finding 17.** This closes finding 10, and correcting the two internal
scaladocs as well was the right call. The one place the old story survives is
the public interface, and it's the one Iceberg reads:
`SupportsRuntimeV2Filtering.filter`'s Javadoc still says
> This method may be called multiple times with additional predicates (e.g.
{@link PartitionPredicate}) when {@link #supportsIterativePushdown()} returns
true.
(`SupportsRuntimeV2Filtering.java:74-77`, and the class-level *Iterative
filtering* paragraph at `:34-40` says the same).
But nothing about the shared-scan case is specific to the Catalyst path.
`RowLevelOperationRuntimeGroupFiltering.injectGroupFilters` wraps every
`DataSourceV2ScanRelation` whose `r.scan eq scan`, `DataSourceV2Strategy`
builds one `BatchScanExec` per relation, and each one's `filteredPartitions`
calls `PushDownUtils.pushRuntimeFilters` on the same `Scan` instance — which is
what your own `assertCatalystGroupFilter` pins down for this interface
(`pushed.size === batchScans.size`, 2 for a group-based UPDATE). A V2 scan with
`supportsIterativePushdown() == false` gets two `filter(Predicate[])` calls on
the same UPDATE.
Pre-existing behaviour, so not this PR's bug — but the PR already edits this
file, and after finding 10 the internal note and the public one now say
different things. One sentence next to the iterative-pushdown paragraph is
enough:
```java
* Independently of {@link #supportsIterativePushdown()}, this method may
also be called once
* per scan node when a plan holds several scan nodes sharing one {@link
Scan} instance (e.g.
* the two branches of a group-based UPDATE). Implementations must
accumulate state across
* those calls as well.
```
--
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]
