peter-toth commented on code in PR #57727:
URL: https://github.com/apache/spark/pull/57727#discussion_r3711189546
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/PushDownUtils.scala:
##########
@@ -213,6 +217,22 @@ 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.
+ val catalystFilters = runtimeFilters
+ .flatMap(unwrapRuntimeFilterExpression)
+ .filterNot(_ == Literal.TrueLiteral)
Review Comment:
**Finding 1.** This branch pushes every unwrapped runtime filter with no
determinism guard. Every sibling screens for it: the V2 second pass runs its
runtime filters through `isPushablePartitionFilter` (`PushDownUtils.scala:471`
— `deterministic && !hasSubquery && no PythonUDF`), both `pushFilters` branches
partition out non-deterministic filters (SPARK-58112, SPARK-58207), and the
Catalyst pushdown path this interface is modelled on does the same —
`FileScanBuilder.pushFilters` (`FileScanBuilder.scala:73-79`) splits on
`_.deterministic` and drops subquery/`PythonUDF` filters from the partition
filters it keeps. Here nothing does.
Run on this head against your own fixture:
SELECT * FROM tbl WHERE part = (SELECT max(val) FROM dim) OR rand() < 0.5
pushes `((part#270 = 3) OR (rand(6694523947714441432) < 0.5))` —
`deterministic = false` — while the same predicate stays in the post-scan
`FilterExec`. Any partition the source prunes on its own roll of `rand()` is
gone for good, and Spark re-rolls for the rows that survive, so rows that
should have passed are dropped.
Add `TBLPROPERTIES('fully-pushed-filter-attributes' = 'part')` and it gets
worse — the post-scan filter disappears entirely and the connector is the only
evaluator of `rand()`, once per partition rather than once per row:
*(1) Project [id#37, part#38]
+- BatchScan ...tbl_f[id#37, part#38] ... RuntimeFilters: [((part#38 =
Subquery subquery#36, [id=#100]) OR (rand(-656537516222552506) < 0.5))]
The V2 first pass shares the missing guard
(`translateScalarSubqueryFilterV2` translates `Rand` fine —
`V2ExpressionBuilder.scala:155`), so that half is pre-existing and deserves its
own ticket. But the V2 path can never drop a post-scan filter, so the
"evaluated nowhere" case is new here.
Screening with the guard the sibling pass already applies keeps the two
consistent:
```suggestion
.filterNot(_ == Literal.TrueLiteral)
.filter(isPushablePartitionFilter)
```
DPP filters still go through: `isPushablePartitionFilter`'s subquery check
is on catalyst `SubqueryExpression`, and `InSubqueryExec`/`ScalarSubquery` are
`ExecSubqueryExpression`.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/connector/SupportsRuntimeCatalystFiltering.scala:
##########
@@ -0,0 +1,77 @@
+/*
+ * 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.
+ */
+ 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]].
+ */
+ def fullyPushedFilterAttributes(): Array[NamedReference] = Array.empty
Review Comment:
**Finding 2.** No objection to the attribute-level shape — v2 file sources
already work this way. `FileScanBuilder.pushFilters`
(`FileScanBuilder.scala:72-95`) keeps every deterministic partition filter for
itself and returns only `dataFilters ++ nonDeterministicFilters` as post-scan
filters, so "any predicate over these attributes is fully evaluated by the
source" is established practice. What I'd like is for the Javadoc to say what
makes it sound, since nothing in the tree implements this trait yet and two
things are easy to get wrong, both silently:
1. **Exactness, not just reachability.** The file-source precedent holds
because partition pruning is exact — every row of a surviving file carries that
partition value. Nothing restricts `filterAttributes` to partition columns:
`SupportsRuntimeV2Filtering` documents it as "attributes this scan can be
filtered by at runtime", and a scan may prune files or row groups by min/max
statistics on a data column. Statistics-based pruning is not exact, so
declaring such an attribute here returns extra rows with no error.
2. **Any shape, not the shapes you recognize.** The source can't refuse an
individual predicate — by the time `filter()` runs, `DataSourceV2Strategy` has
already removed the `FilterExec`. On this head, with
`fully-pushed-filter-attributes='part'`:
SELECT * FROM t WHERE part > (SELECT max(val) FROM dim) + 1 AND
CAST(part AS STRING) RLIKE '4'
leaves only `Filter (isnotnull(part#341) AND RLIKE(cast(part#341 as
string), 4))` above the scan; `part > (2 + 1)` is gone, pushed as `(part#341 >
(2 + 1))`. A source that hand-matches operators and ignores the rest —
`InMemoryTableWithV2Filter.filter` handles only `=` and `IN` — drops it on the
floor. `InMemoryEnhancedRuntimePartitionFilterTable` gets it right by
delegating to `PartitionPredicate.eval`, i.e. bind and interpret
(`PartitionPredicateImpl.boundPredicate`), which is what the file index does
too.
Something along these lines:
```scala
/**
* 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]].
*
* Only declare an attribute here if this scan evaluates an arbitrary
deterministic Catalyst
* predicate over it exactly, for every row it returns -- e.g. an identity
partition column,
* whose value is known for every row of a surviving partition. Do not
declare an attribute
* whose predicates only guide approximate pruning, such as file or
row-group statistics.
* Spark may push any expression that references only these attributes, so
do not assume a
* fixed set of operators: bind and evaluate the expression (see
* [[PartitionPredicateImpl]]) instead of pattern matching it.
*/
```
While you're here, it would help to name the intended implementor in the PR
description — it makes the contract judgeable and tells a reader why the
interface is internal.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/connector/SupportsRuntimeCatalystFiltering.scala:
##########
@@ -0,0 +1,77 @@
+/*
+ * 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.
+ */
+ 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]].
+ */
+ 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.
+ *
+ * Note that Spark will call [[Scan.toBatch]] again after filtering the scan
at runtime.
+ */
+ def filter(expressions: Array[Expression]): Unit
Review Comment:
**Finding 4.** `SupportsRuntimeV2Filtering.filter` documents the
partitioning-preservation contract — *"If the scan also implements
SupportsReportPartitioning, it must preserve the originally reported
partitioning ... The scan must not report new partition values that were not
present in the original partitioning"* — and
`PushDownUtils.replanWithRuntimeFilters` enforces it for whatever scan it was
handed, this interface included: it calls `pushRuntimeFilters`, then
`scan.toBatch.planInputPartitions()`, then the `KeyedPartitioning` checks that
throw `SparkException` on a missing `HasPartitionKey`, a new partition key, or
a grown per-key partition count. An SPJ-active adopter reading only this
Javadoc finds out from `"Data source must have preserved the original
partitioning during runtime filtering"`. Please carry that paragraph over.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/connector/SupportsRuntimeCatalystFiltering.scala:
##########
@@ -0,0 +1,77 @@
+/*
+ * 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.
+ */
+ 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]].
+ */
+ 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.
+ *
+ * Note that Spark will call [[Scan.toBatch]] again after filtering the scan
at runtime.
+ */
+ def filter(expressions: Array[Expression]): Unit
+
+ /**
+ * Returns the predicates that are pushed to the data source via [[filter]].
+ *
+ * This method does not indicate whether a predicate is fully pushed. Spark
infers that from
+ * [[fullyPushedFilterAttributes]]. The returned predicates may fully or
partially help the data
+ * source prune initially planned
+ * [[org.apache.spark.sql.connector.read.InputPartition]]s.
+ *
+ * It's possible that there are no runtime predicates and [[filter]] is
never called;
+ * an empty array should be returned for this case.
+ */
+ def pushedPredicates(): Array[Expression] = Array.empty
Review Comment:
**Finding 5.** Nothing in Spark reads this.
`SupportsRuntimeV2Filtering.pushedPredicates()` earns its place —
`PushDownUtils.scala:133,205` uses it to avoid pushing the same predicate twice
across the two iterative passes — but this path pushes once and never consults
the result; grepping `sql/core/src/main` and `sql/catalyst/src/main` for
`pushedPredicates` finds no call on this trait, only the new suite's assertions.
Either drop it and let the fixture expose its own accessor
(`InMemoryEnhancedRuntimePartitionFilterTable.pushedPartitionPredicates` is the
precedent), or keep it and say in the Javadoc that it exists for
inspection/testing and Spark does not consult it — as written the doc reads
like part of a contract Spark relies on.
##########
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()
+ case _ => Array.empty[NamedReference]
+ }
+ AttributeSet(V2ExpressionUtils.resolveRefs[Attribute](
Review Comment:
**Finding 6.** `resolveRefs` → `V2ExpressionUtils.resolveRef` throws
`cannotResolveAttributeError` when a reference doesn't resolve against the
plan's output, and casts the result to `Attribute` — so a nested reference,
which `LogicalPlan.resolve` hands back as an `Alias(GetStructField(...))`,
throws a `ClassCastException`. `fullyPushedFilterAttributes()` therefore has an
unwritten requirement: top-level attributes only, and only ones that survived
column pruning into the scan's `readSchema`. Break it and the query fails at
planning time.
`filterAttributes` carries the same requirement and is equally undocumented,
but it's forced for every scan relation, so an adopter trips it on the first
query. This one is only forced when a scalar-subquery runtime filter is present
(`scalarSubqueryFilters.filter` doesn't evaluate its closure on an empty Seq),
which makes it a query-shape-dependent failure. Worth a line on the trait
alongside finding 2; the new fixture quietly depends on it via the
`scanFields.contains(name)` guard at
`InMemoryCatalystRuntimeFilterTable.scala:267`.
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2CatalystRuntimeFilterSuite.scala:
##########
@@ -0,0 +1,308 @@
+/*
+ * 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 org.apache.spark.SparkConf
+import org.apache.spark.sql.{DataFrame, Row}
+import org.apache.spark.sql.catalyst.expressions.{Add, AttributeReference,
DynamicPruning, DynamicPruningExpression, EqualTo, Expression, GreaterThan,
Literal}
+import
org.apache.spark.sql.connector.catalog.{InMemoryCatalystRuntimeFilterTable,
InMemoryTableCatalystRuntimeFilterCatalog}
+import org.apache.spark.sql.execution.{FilterExec, ScalarSubquery =>
ExecScalarSubquery}
+import org.apache.spark.sql.execution.ExplainUtils.stripAQEPlan
+import org.apache.spark.sql.execution.datasources.v2.BatchScanExec
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.test.SharedSparkSession
+import org.apache.spark.sql.types.IntegerType
+
+/**
+ * Tests for scans that implement
+ *
[[org.apache.spark.sql.internal.connector.SupportsRuntimeCatalystFiltering]],
+ * where runtime filters are pushed once as Catalyst expressions instead of
connector
+ * predicates.
+ */
+class DataSourceV2CatalystRuntimeFilterSuite extends SharedSparkSession {
+
+ protected val v2Source = classOf[FakeV2ProviderWithCustomSchema].getName
+ protected val catalogName = "testcatalystruntimefilter"
+
+ override def sparkConf: SparkConf = super.sparkConf
+ .set(s"spark.sql.catalog.$catalogName",
+ classOf[InMemoryTableCatalystRuntimeFilterCatalog].getName)
+
+ private def withDPPConf(f: => Unit): Unit = {
+ withSQLConf(
+ SQLConf.DYNAMIC_PARTITION_PRUNING_ENABLED.key -> "true",
+ SQLConf.DYNAMIC_PARTITION_PRUNING_REUSE_BROADCAST_ONLY.key -> "false",
+ SQLConf.DYNAMIC_PARTITION_PRUNING_FALLBACK_FILTER_RATIO.key -> "10")(f)
+ }
+
+ test("scalar subquery on partition column -> pushed as Catalyst expression")
{
+ val tbl = s"$catalogName.tbl1"
+ val dim = s"$catalogName.dim1"
+ withTable(tbl, dim) {
+ sql(s"CREATE TABLE $tbl (id INT, part INT) USING $v2Source PARTITIONED
BY (part)")
+ for (i <- 0 until 5) {
+ sql(s"INSERT INTO $tbl VALUES ($i, $i)")
+ }
+ sql(s"CREATE TABLE $dim (val INT) USING $v2Source")
+ sql(s"INSERT INTO $dim VALUES (3)")
+
+ val df = sql(s"SELECT * FROM $tbl WHERE part = (SELECT max(val) FROM
$dim)")
+ checkAnswer(df, Row(3, 3))
+
+ assertScalarSubqueryRuntimeFilters(df)
+ val part = AttributeReference("part", IntegerType, nullable = false)()
+ assertPushedCatalystPredicatesEqual(df, EqualTo(part, Literal(3)))
+ // `part` is not declared fully pushed, so Spark still evaluates the
filter after the scan.
+ assertScalarSubqueryEvaluatedAfterScan(df, expected = true)
+ }
+ }
+
+ test("predicate on fully pushed filter attributes -> not evaluated after the
scan") {
+ val tbl = s"$catalogName.tbl_fully_pushed"
+ val dim = s"$catalogName.dim_fully_pushed"
+ withTable(tbl, dim) {
+ sql(s"CREATE TABLE $tbl (id INT, part INT) USING $v2Source PARTITIONED
BY (part) " +
+ "TBLPROPERTIES('fully-pushed-filter-attributes' = 'part')")
+ for (i <- 0 until 5) {
+ sql(s"INSERT INTO $tbl VALUES ($i, 3)")
+ }
+ sql(s"CREATE TABLE $dim (val INT) USING $v2Source")
+ sql(s"INSERT INTO $dim VALUES (3)")
+
+ val df = sql(s"SELECT * FROM $tbl WHERE part = (SELECT max(val) FROM
$dim)")
+ checkAnswer(df, (0 until 5).map(i => Row(i, 3)))
+
+ assertScalarSubqueryRuntimeFilters(df)
+ val part = AttributeReference("part", IntegerType, nullable = false)()
+ assertPushedCatalystPredicatesEqual(df, EqualTo(part, Literal(3)))
+ assertScalarSubqueryEvaluatedAfterScan(df, expected = false)
+ }
+ }
+
+ test("predicate on partly fully pushed filter attributes -> evaluated after
the scan") {
+ val tbl = s"$catalogName.tbl_partly_pushed"
+ val dim = s"$catalogName.dim_partly_pushed"
+ withTable(tbl, dim) {
+ sql(s"CREATE TABLE $tbl (id INT, p1 INT, p2 INT) USING $v2Source " +
+ "PARTITIONED BY (p1, p2) " +
+ "TBLPROPERTIES('fully-pushed-filter-attributes' = 'p1')")
+ for (i <- 0 until 5) {
+ sql(s"INSERT INTO $tbl VALUES ($i, 1, 2)")
+ }
+ sql(s"CREATE TABLE $dim (val INT) USING $v2Source")
+ sql(s"INSERT INTO $dim VALUES (3)")
+
+ // The predicate also references p2, which is not declared fully pushed,
so it is not
+ // considered fully pushed and Spark keeps evaluating it after the scan.
+ val df = sql(s"SELECT * FROM $tbl WHERE p1 + p2 = (SELECT max(val) FROM
$dim)")
+ checkAnswer(df, (0 until 5).map(i => Row(i, 1, 2)))
+
+ assertScalarSubqueryRuntimeFilters(df)
+ val p1 = AttributeReference("p1", IntegerType, nullable = false)()
+ val p2 = AttributeReference("p2", IntegerType, nullable = false)()
+ assertPushedCatalystPredicatesEqual(df, EqualTo(Add(p1, p2), Literal(3)))
+ assertScalarSubqueryEvaluatedAfterScan(df, expected = true)
+ }
+ }
+
+ test("untranslatable filter -> pushed instead of dropped") {
+ val tbl = s"$catalogName.tbl2"
+ val dim = s"$catalogName.dim2"
+ withTable(tbl, dim) {
+ sql(s"CREATE TABLE $tbl (id INT, part INT) USING $v2Source PARTITIONED
BY (part)")
+ for (i <- 0 until 5) {
+ sql(s"INSERT INTO $tbl VALUES ($i, $i)")
+ }
+ sql(s"CREATE TABLE $dim (val INT) USING $v2Source")
+ sql(s"INSERT INTO $dim VALUES (2)")
+
+ // `part > sub + 1` has no data source V2 translation, so the V2
interfaces would never
+ // see it. The scalar subquery is literalized but the surrounding
expression is kept.
+ val df = sql(s"SELECT * FROM $tbl WHERE part > (SELECT max(val) FROM
$dim) + 1")
+ checkAnswer(df, Row(4, 4))
+
+ assertScalarSubqueryRuntimeFilters(df)
+ val part = AttributeReference("part", IntegerType, nullable = false)()
+ assertPushedCatalystPredicatesEqual(
+ df, GreaterThan(part, Add(Literal(2), Literal(1))))
+ }
+ }
+
+ test("DPP filter -> pushed as InSubqueryExec expression") {
+ val fact = s"$catalogName.fact3"
+ val dim = s"$catalogName.dim3"
+ withTable(fact, dim) {
+ sql(s"CREATE TABLE $fact (id INT, part INT) USING $v2Source PARTITIONED
BY (part)")
+ for (i <- 0 until 5) {
+ sql(s"INSERT INTO $fact VALUES ($i, $i)")
+ }
+ sql(s"CREATE TABLE $dim (dim_id INT, dim_val STRING) USING $v2Source")
+ sql(s"INSERT INTO $dim VALUES (2, 'two')")
+
+ withDPPConf {
+ val df = sql(
+ s"""SELECT f.id, f.part FROM $fact f JOIN $dim d
+ |ON f.part = d.dim_id WHERE d.dim_val = 'two'""".stripMargin)
+ checkAnswer(df, Row(2, 2))
+
+ assertDPPRuntimeFilters(df)
+ val dppPredicate = collectBatchScan(df).runtimeFilters.collectFirst {
+ case DynamicPruningExpression(e) => e
+ }.get
+ assertPushedCatalystPredicatesEqual(df, dppPredicate)
+ }
+ }
+ }
+
+ test("filter on column outside filterAttributes -> not pushed") {
+ val tbl = s"$catalogName.tbl4"
+ val dim = s"$catalogName.dim4"
+ withTable(tbl, dim) {
+ sql(s"CREATE TABLE $tbl (id INT, p1 INT, p2 INT) USING $v2Source " +
+ "PARTITIONED BY (p1, p2) " +
+ "TBLPROPERTIES('filter-attributes' = 'p1')")
+ for (i <- 0 until 5) {
+ sql(s"INSERT INTO $tbl VALUES ($i, $i, 10)")
+ }
+ sql(s"CREATE TABLE $dim (val INT) USING $v2Source")
+ sql(s"INSERT INTO $dim VALUES (10)")
+
+ // p2 is a partition column but is not declared filterable, so no
runtime filter is derived.
+ val df = sql(s"SELECT * FROM $tbl WHERE p2 = (SELECT max(val) FROM
$dim)")
+ checkAnswer(df, (0 until 5).map(i => Row(i, i, 10)))
+
+ assert(collectBatchScan(df).runtimeFilters.isEmpty,
+ "Expected no runtime filters for a column outside filterAttributes")
+ assertPushedCatalystPredicates(df, 0)
+ }
+ }
+
+ test("no runtime filter -> filter() is never called") {
+ val tbl = s"$catalogName.tbl5"
+ withTable(tbl) {
+ sql(s"CREATE TABLE $tbl (id INT, part INT) USING $v2Source PARTITIONED
BY (part)")
+ for (i <- 0 until 5) {
+ sql(s"INSERT INTO $tbl VALUES ($i, $i)")
+ }
+
+ val df = sql(s"SELECT * FROM $tbl WHERE part = 3")
+ checkAnswer(df, Row(3, 3))
+
+ assert(collectBatchScan(df).runtimeFilters.isEmpty)
+ assertPushedCatalystPredicates(df, 0)
+ }
+ }
+
+ //
---------------------------------------------------------------------------
+ // Helper methods
+ //
---------------------------------------------------------------------------
+
+ private def assertDPPRuntimeFilters(
+ df: DataFrame, expectedCount: Int = 1): Unit = {
+ val batchScan = collectBatchScan(df)
+ val dppFilters = batchScan.runtimeFilters.collect {
+ case d: DynamicPruningExpression => d
+ }
+ assert(dppFilters.size === expectedCount,
+ s"Expected $expectedCount DynamicPruningExpression(s) " +
+ s"in runtimeFilters, got ${dppFilters.size}")
+ }
+
+ private def assertScalarSubqueryRuntimeFilters(
+ df: DataFrame, expectedCount: Int = 1): Unit = {
+ val batchScan = collectBatchScan(df)
+ val scalarFilters = batchScan.runtimeFilters.collect {
+ case f if !f.isInstanceOf[DynamicPruning] => f
+ }
+ val dppFilters = batchScan.runtimeFilters.collect {
+ case d: DynamicPruning => d
+ }
+ assert(scalarFilters.size === expectedCount,
+ s"Expected $expectedCount scalar subquery runtime filter(s), " +
+ s"got ${scalarFilters.size}")
+ assert(dppFilters.isEmpty,
+ "Expected non-DPP runtime filters (scalar subquery)")
+ }
+
+ /**
+ * Checks whether a scalar subquery runtime filter is still evaluated by a
[[FilterExec]] above
+ * the scan. Filters that only reference `fullyPushedFilterAttributes` are
dropped from it.
+ */
+ private def assertScalarSubqueryEvaluatedAfterScan(
+ df: DataFrame,
+ expected: Boolean): Unit = {
+ val postScanConditions =
stripAQEPlan(df.queryExecution.executedPlan).collect {
+ case f: FilterExec => f.condition
+ }
+ val evaluated =
postScanConditions.exists(_.exists(_.isInstanceOf[ExecScalarSubquery]))
+ assert(evaluated === expected,
+ s"Expected scalar subquery evaluated after scan to be $expected, " +
+ s"post-scan filter conditions: $postScanConditions")
+ }
+
+ private def collectBatchScan(df: DataFrame): BatchScanExec = {
+ stripAQEPlan(df.queryExecution.executedPlan).collectFirst {
+ case b: BatchScanExec => b
+ }.getOrElse(fail("Expected BatchScanExec in plan"))
+ }
+
+ private def getPushedCatalystPredicates(df: DataFrame): Seq[Expression] = {
+ collectBatchScan(df).scan match {
+ case s:
InMemoryCatalystRuntimeFilterTable#InMemoryCatalystRuntimeFilterBatchScan =>
+ s.pushedPredicates().toSeq
+ case _ => Seq.empty
Review Comment:
**Finding 7.** This swallows a wrong-scan-type case, and the two
`assertPushedCatalystPredicates(df, 0)` assertions (`:191`, `:207`) are exactly
the ones that then pass for the wrong reason — "nothing was pushed" and "we
didn't find the scan we expected" become indistinguishable.
```suggestion
case other => fail(s"Expected InMemoryCatalystRuntimeFilterBatchScan,
got $other")
```
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/dynamicpruning/PartitionPruning.scala:
##########
@@ -86,6 +87,14 @@ object PartitionPruning extends Rule[LogicalPlan] with
PredicateHelper with Join
} else {
None
}
+ case (resExp, r @ ExtractV2Scan(scan: SupportsRuntimeCatalystFiltering))
=>
Review Comment:
**Finding 3.** This is one of three places that dispatch on
`SupportsRuntimeV2Filtering`; the third isn't updated.
`RowLevelOperationRuntimeGroupFiltering.scala:54,58` matches
`ExtractV2Scan(scan: SupportsRuntimeV2Filtering)` for group-based and
delta-based row-level operations, and `canInjectGroupFilters` /
`injectGroupFilters` are typed on that interface. Since the Javadoc you added
says only one runtime filtering interface should be implemented, a source that
adopts `SupportsRuntimeCatalystFiltering` silently loses runtime group
filtering for MERGE/UPDATE/DELETE — no error, just a rule that stops firing and
whole unmodified groups getting read.
The rule needs nothing but `filterAttributes` and injects a plain
`DynamicPruningExpression(InSubquery(...))`, which the new branch in
`pushRuntimeFilters` already handles, so adding the two cases looks mechanical.
If you'd rather keep it out of scope, please say so in the trait's Javadoc so
an adopter isn't surprised.
--
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]
