szehon-ho commented on code in PR #57727:
URL: https://github.com/apache/spark/pull/57727#discussion_r3731866199
##########
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:
Right, `extends SupportsRuntimeV2Filtering` makes it unfollowable on this
page. Dropped the line.
The "only one runtime filtering interface" statement stays on
`SupportsRuntimeCatalystFiltering`, which is the interface that can actually
conflict and the one an author would be adopting when they create the conflict.
Spark now enforces it rather than only documenting it (finding 9).
##########
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:
You are right, and the "Why are the changes needed?" paragraph was wrong for
the same reason. Fixed all three.
Added "filter with no V2 translation -> pushed instead of dropped". I used a
`STRING` partition column so the filter is `part RLIKE (SELECT max(val) FROM
dim)` with no casts around it, and asserted the untranslatability directly
rather than by reasoning:
`DataSourceV2Strategy.translateScalarSubqueryFilterV2(runtimeFilter)` returns
`None`, and `RLike(part, 3)` is what reaches the scan. With `dim = ('3')` and
`part = '0'..'4'` the answer is `Row(3, "3")`.
Kept the old test, renamed to "arithmetic around a scalar subquery ->
subquery literalized, expression pushed intact", with the comment corrected to
say it would translate and to point at the new test for one that would not. The
PR description now leads with `RLIKE` instead.
##########
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:
Fair, and applied your sentence verbatim. You are right that the
neighbouring `filter()` wording invites the reading you describe -- pruning
partitions is the only action it sanctions, so "fully evaluates" collapses into
"prunes partitions" -- and min/max pruning over a data column is exactly the
case where that is wrong and silent.
--
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]
