peter-toth commented on code in PR #57727:
URL: https://github.com/apache/spark/pull/57727#discussion_r3726895511
##########
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:
**Finding 9.** Every other dispatch resolves `SupportsRuntimeV2Filtering`
first — `runtimeFilterAttrs` right above (`:205` before `:206`),
`pushRuntimeFilters` (`PushDownUtils.scala:190` before `:226`),
`PartitionPruning.getFilterableTableScan` (`:82` before `:90`),
`RowLevelOperationRuntimeGroupFiltering` (`:56` before `:61`). This one
doesn't: it matches only the new interface, so for a scan mixing in both it
reports non-empty fully-pushed attributes while every push decision goes down
the V2 path.
Concretely, with such a scan and `fullyPushedFilterAttributes() = [part]`:
```sql
SELECT * FROM t WHERE part > (SELECT max(val) FROM dim) + 1
```
`scalarSubqueryFilters` routes it (the V2 `filterAttributes` covers `part`),
`fullyPushedRuntimeFilters` drops it from `postScanFilters`, and then whether
it is pushed at all depends on `translateFilterV2` succeeding and, failing
that, on `supportsIterativePushdown()` plus `getPartitionPredicateSchema`
returning a schema. Any of those falling through leaves the predicate with no
evaluator and extra rows in the answer.
The Javadoc does say only one runtime filtering interface should be
implemented, but this is the one place where Spark's own behaviour depends on
that and doesn't check it, and the failure is silent. Making the precedence
uniform is enough:
```suggestion
// Matched after SupportsRuntimeV2Filtering, as `runtimeFilterAttrs`
and
// `pushRuntimeFilters` do: a scan taking the V2 push path must keep
its post-scan filters.
case _: SupportsRuntimeV2Filtering => Array.empty[NamedReference]
case s: SupportsRuntimeCatalystFiltering =>
s.fullyPushedFilterAttributes()
```
##########
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:
**Finding 11.** The V2 partition-predicate path flattens nested partition
references before pushing; this branch doesn't, so the two interfaces hand a
source different expression shapes for the same table.
`createRuntimePartitionPredicates` (`:465-467`) runs
`flattenNestedPartitionFilters`, which rewrites `GetStructField(s#1, "tz")`
into `attr("s.tz")` — its own scaladoc uses exactly that example (`:500-509`) —
and `getPartitionPredicateSchema` supports identity transforms on nested fields
via `resolveIdentityPartitionField` / `findNestedField` (`:394-409`). Here the
unwrapped expression goes out as-is.
It's reachable. `filterAttributes()` now documents that references must be
top-level, so a table partitioned by `s.tz` can only declare `s`.
`GetStructField(s, "tz").references` is `{s}`, so
`resExp.references.subsetOf(filterAttrs)` in `PartitionPruning` and
`f.references.subsetOf(runtimeFilterAttrs)` in `DataSourceV2Strategy` both
pass, and the source receives `GetStructField(s#1, "tz") = 3` while its
partition-key row is laid out as a flat `s.tz` field.
A line on `filter()` is the smaller fix: say the pushed expression may
contain nested field accesses over a declared top-level attribute, and the scan
is responsible for matching them against its own partition layout. Flattening
here is also possible — `pushRuntimeFilters` already has `table` and `output`,
so `getPartitionPredicateSchema(table,
output).map(flattenNestedPartitionFilters(catalystFilters, _).keys.toSeq)`
mirrors the V2 path — but that also rewrites attribute references in the flat
case, which is a bigger change than this branch needs.
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryRowLevelOperationTable.scala:
##########
@@ -267,6 +270,94 @@ class InMemoryRowLevelOperationTable private (
override def abort(messages: Array[WriterCommitMessage]): Unit = {}
}
+
+ /**
+ * Builds a scan for row-level operations. When
+ * `use-catalyst-runtime-filtering` is set, the scan implements
+ * [[SupportsRuntimeCatalystFiltering]] so group filtering goes through the
Catalyst path.
+ */
+ private def newRowLevelScanBuilder(
+ options: CaseInsensitiveStringMap)(
+ onBuild: BatchScanBaseClass => Unit): ScanBuilder = {
+ new InMemoryScanBuilder(schema, options) {
+ override protected def createScan(
+ partitions: Seq[InputPartition],
+ readSchema: StructType,
+ tableSchema: StructType,
+ options: CaseInsensitiveStringMap): BatchScanBaseClass = {
+ if (useCatalystRuntimeFiltering) {
+ InMemoryCatalystRowLevelBatchScan(partitions, readSchema,
tableSchema, options)
+ } else {
+ super.createScan(partitions, readSchema, tableSchema, options)
+ }
+ }
+
+ override def build: Scan = {
+ val scan = super.build().asInstanceOf[BatchScanBaseClass]
+ onBuild(scan)
+ scan
+ }
+ }
+ }
+
+ /**
+ * Row-level batch scan that receives runtime filters as Catalyst
expressions.
+ * Evaluates partition-column predicates against each partition key so group
filtering
+ * actually prunes partitions (needed for `replacedPartitions` assertions).
+ */
+ case class InMemoryCatalystRowLevelBatchScan(
+ var _data: Seq[InputPartition],
+ readSchema: StructType,
+ tableSchema: StructType,
+ options: CaseInsensitiveStringMap)
+ extends BatchScanBaseClass(_data, readSchema, tableSchema)
+ with SupportsRuntimeCatalystFiltering {
+
+ private val _catalystPredicates = ArrayBuffer.empty[Expression]
+
+ override def filterAttributes(): Array[NamedReference] = {
+ val scanFields = readSchema.fields.map(_.name).toSet
+ partitioning.flatMap(_.references())
+ .filter(ref => scanFields.contains(ref.fieldNames.mkString(".")))
+ }
+
+ override def filter(expressions: Array[Expression]): Unit = {
Review Comment:
**Finding 12.** This `filter` / `partitionAttributes` /
`pushedCatalystPredicates` block and `InMemoryCatalystRuntimeFilterBatchScan`'s
(`InMemoryCatalystRuntimeFilterTable.scala:105-141`) are the same ~40 lines
twice: the reference remap, the all-refs-are-partition-columns test,
`BindReferences.bindReference` + `Predicate.createInterpreted`, the
`data.filter` loop, the swallow-and-keep `catch`, and the accessor. Only
`filterAttributes()` genuinely differs — the other fixture also honours a
`filter-attributes` property.
Both are inner classes of tables extending `InMemoryBaseTable`, so a trait
next to `BatchScanBaseClass` carrying `filter`, `partitionAttributes` and
`pushedCatalystPredicates` would leave each scan with just its
`filterAttributes()` override — the same kind of sharing the new `createScan`
hook does for scan construction. Worth doing at two copies rather than three,
since this is the bind-and-interpret reference implementation an adopter will
read.
Also, the `catch { case _: Exception => true }` here lost the comment its
twin has (`InMemoryCatalystRuntimeFilterTable.scala:125`: "Keep the partition
on eval failure, matching PartitionPredicateImpl"), which is what tells a
reader it's deliberate.
##########
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.
Review Comment:
**Finding 13.** Two claims around this guard no longer hold. The guard
itself is fine — defence-in-depth is the right call — it's the wording I'd fix.
*"Keeps non-deterministic filters from being the sole evaluator"* — after
the rebase, `scalarSubqueryFilters` requires `f.deterministic`
(`DataSourceV2Strategy.scala:183`), and `dynamicFilters` can't carry
non-determinism either (`CleanupDynamicPruningFilters` rewrites those to
`TrueLiteral`), so no non-deterministic filter reaches this line. Your commit
message says exactly that; the comment doesn't. What's left for the screen to
reject is a `PythonUDF` — also unreachable, since `ExtractPythonUDFs` runs in
`SparkOptimizer`'s "Extract Python UDFs" batch (`SparkOptimizer.scala:92`) and
lifts the UDF into its own eval node below the `Filter`, so a post-scan filter
can't hold one — and a residual `SubqueryExpression`.
*"so both sides apply the same test"* (`DataSourceV2Strategy.scala:193`) —
not literally. `DataSourceV2Strategy` tests the logical filter with
`includeSubquery = true`; this line tests the *unwrapped* expression with
`includeSubquery = false`. Screening before the unwrap with the same flag makes
them identical, which is what the comment promises and what stops them drifting
again:
```scala
val catalystFilters = runtimeFilters
.filter(isPushablePartitionFilter(_, includeSubquery = true))
.flatMap(unwrapRuntimeFilterExpression)
.filterNot(_ == Literal.TrueLiteral)
```
No behaviour change today — `DynamicPruningExpression` is deterministic and
`SubqueryExpression.hasSubquery` doesn't match `InSubqueryExec` — just one
predicate applied to one form on both sides.
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2CatalystRuntimeFilterSuite.scala:
##########
@@ -0,0 +1,337 @@
+/*
+ * 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')")
+ // Matching and nonmatching partitions: the scan must prune nonmatching
ones itself
+ // because Spark drops the post-scan FilterExec for fully pushed
attributes.
+ 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)))
+ assertScalarSubqueryEvaluatedAfterScan(df, expected = false)
+ }
+ }
+
+ test("non-deterministic predicate on fully pushed attributes -> evaluated
after the scan") {
+ val tbl = s"$catalogName.tbl_nondeterministic"
+ val dim = s"$catalogName.dim_nondeterministic"
+ 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, $i)")
+ }
+ sql(s"CREATE TABLE $dim (val INT) USING $v2Source")
+ sql(s"INSERT INTO $dim VALUES (3)")
+
+ // A non-deterministic filter is never pushed, so it must keep its
post-scan FilterExec even
+ // though it only references a fully pushed attribute. Dropping it there
would leave nothing
+ // to evaluate it and the scan would return the nonmatching partitions
too.
+ val df = sql(
+ s"SELECT * FROM $tbl WHERE part = (SELECT max(val) FROM $dim) OR
rand() < 0.5")
+ // The row in the matching partition always qualifies, the others
qualify at random.
+ assert(df.collect().contains(Row(3, 3)))
+
+ assertScalarSubqueryEvaluatedAfterScan(df, expected = true)
+ assertScalarSubqueryRuntimeFilters(df, expectedCount = 0)
+ assertPushedCatalystPredicates(df, 0)
+ }
+ }
+
+ 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
Review Comment:
**Finding 15.** This filter does translate, so the test isn't covering the
case it names — and nothing else in the suite covers it either.
The V2 branch routes a non-DPP runtime filter through
`DataSourceV2Strategy.translateScalarSubqueryFilterV2`
(`PushDownUtils.scala:195`), which literalizes `ExecScalarSubquery` **first**
and only then calls `translateFilterV2`
(`DataSourceV2Strategy.scala:997-1003`). So the V2 interfaces see `part > (2 +
1)`, not `part > (scalar-subquery) + 1`. And that translates:
- `Literal.contextIndependentFoldable` is `true` (`literals.scala:503`) and
`BinaryArithmetic.contextIndependentFoldable` is `left && right`
(`arithmetic.scala:210-211`), so `Add(Literal(2), Literal(1))` is
context-independent foldable;
- `V2ExpressionBuilder.generateExpression` folds exactly that case to a
`LiteralValue` (`V2ExpressionBuilder.scala:97-100`), and
`spark.sql.optimizer.datasourceV2ExprFolding` defaults to `true` (asserted in
`SQLConfSuite.scala:248`);
- the outer `GreaterThan` is a `BinaryComparison`, which `canTranslate`
accepts unconditionally (`V2ExpressionBuilder.scala:78`).
So `translateFilterV2` returns `V2Predicate(">", [FieldReference("part"),
LiteralValue(3)])`. Even with folding turned off it still translates —
`canTranslate` accepts `Add` when `evalMode == ANSI`, which is the default
(`SQLConf.scala:5752`).
Three things say otherwise: this comment, the test name, and the "Why are
the changes needed?" paragraph, which leads with `part > (subquery) + 1` as a
filter that is "silently dropped and never reach[es] the data source". More
importantly, the capability this PR exists for — a runtime filter with *no* V2
translation reaching the source — is not tested.
The description's second example does hold: `V2ExpressionBuilder` has no
`RLike` case (only `STARTS_WITH` / `ENDS_WITH` / `CONTAINS`, `:138-140`), so
```sql
SELECT * FROM t WHERE CAST(part AS STRING) RLIKE CAST((SELECT max(val) FROM
dim) AS STRING)
```
references only `part`, carries a `SCALAR_SUBQUERY`, and has no translation
at all. With `dim = (3)` and `part = 0..4` it should push `RLIKE(cast(part as
string), 3)` and answer `Row(3, 3)`; the pattern is still a subquery at
optimization time, so nothing rewrites the `RLIKE` into a `Contains` beforehand.
Worth keeping the current test as well — it's a good check that the
surrounding arithmetic survives and the subquery is literalized — just renamed
to say that instead of "untranslatable".
##########
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:
**Finding 10.** Spark calls this more than once on the same scan instance,
and nothing here says so.
A group-based UPDATE is rewritten into a union of two branches sharing one
`Scan`; each `BatchScanExec` has its own `filteredPartitions`, so each one runs
`replanWithRuntimeFilters` -> `pushRuntimeFilters` -> `filter()`. Your own test
pins that down: `RowLevelOperationCatalystRuntimeFilterSuiteBase.scala:1540`,
`pushed.size === batchScans.size`, "expected each of the N scan node(s) to push
the filter once", with the scaladoc above it explaining the union.
`SupportsRuntimeV2Filtering` documents its multi-call case explicitly (the
"Iterative filtering" paragraph, gated on `supportsIterativePushdown()`). Here
there's no gate and no note, and `pushRuntimeFilters`' own scaladoc says the
opposite — *"Note: Do not call multiple times for the same `scan` instance"*
(`PushDownUtils.scala:170`, repeated at `:257`). An implementor who reads
`filter()` as "here is the complete predicate set" and replaces its state per
call un-prunes on the second call. Today both calls happen to carry equivalent
predicates (same subquery, different expr IDs), so it works by luck — which is
the reason to write it down rather than leave it to be discovered.
```scala
* 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.
*
* Spark may call this method more than once for the same scan instance: a
plan can hold several
* scan nodes sharing one scan (e.g. the two branches of a group-based
UPDATE), and each pushes
* its own copy of the runtime filters. Implementations must treat
successive calls as additive,
* ANDing the new expressions with those already pushed rather than
replacing them.
```
The stale note in `pushRuntimeFilters` / `replanWithRuntimeFilters` is worth
correcting in the same pass, since it's the note that would stop a future
reader from relying on this.
##########
sql/catalyst/src/main/java/org/apache/spark/sql/connector/read/SupportsRuntimeFiltering.java:
##########
@@ -26,6 +26,7 @@
/**
* A mix-in interface for {@link Scan}. Data sources can implement this
interface if they can
* filter initially planned {@link InputPartition}s using predicates Spark
infers at runtime.
+ * Only one runtime filtering interface should be implemented by a data source.
Review Comment:
**Finding 14.** On this interface the sentence can't be followed:
`SupportsRuntimeFiltering extends SupportsRuntimeV2Filtering` (`:36`), so
implementing it always implements two runtime filtering interfaces. It isn't
actionable either — the third interface it's warning about,
`SupportsRuntimeCatalystFiltering`, is internal and doesn't appear in the
published Javadoc, so a reader takes "only one" to mean the two names on this
page, which is the one pair the `extends` makes inseparable.
`SupportsRuntimeV2Filtering` already carries the version that works
(*"`SupportsRuntimeV2Filtering` is preferred over `SupportsRuntimeFiltering`"*
plus the same "only one" line, where it at least reads as a choice between that
interface and something else). I'd just drop the line here. If you want
something on this page, make it about Spark's behaviour rather than a count —
e.g. "Spark takes exactly one runtime filtering path per scan" — so it doesn't
read as advice to not implement the supertype.
--
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]
