LuciferYang commented on code in PR #55579:
URL: https://github.com/apache/spark/pull/55579#discussion_r4454062991
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/connector/SupportsRuntimeCatalystFiltering.scala:
##########
@@ -72,35 +80,38 @@ trait SupportsRuntimeCatalystFiltering extends Scan {
def fullyPushedFilterAttributes(): Array[NamedReference] = Array.empty
/**
- * Filters this scan using runtime Catalyst expressions.
+ * Plans this scan's input partitions with runtime Catalyst expressions
applied.
+ *
+ * The provided expressions must be interpreted as a set of predicates that
are ANDed together,
+ * and applied on top of the filters already pushed at query compilation.
Implementations may use
+ * them to plan fewer [[InputPartition]]s than
[[org.apache.spark.sql.connector.read.Batch]]
+ * would without them.
*
- * 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 calls this method at execution time, and only when at least one
runtime filter survives
+ * screening; the partitions it returns replace the ones that scan node
would otherwise have
+ * read. Spark may also call `Batch.planInputPartitions()` on the same scan,
so neither method may
+ * assume it is the only one used. One scan instance can back several scan
nodes (e.g. the two
+ * branches of a group-based UPDATE), each with its own runtime filters, so
an implementation
+ * must derive its result from the given expressions alone and must not
carry state over from a
+ * previous call. A group-based scan whose write replaces the groups that
were read has to record
+ * what each call returned for that write, and count a scan node that read
through
+ * `planInputPartitions()` as having read every group that method returns.
Spark may also call
+ * that method while planning a scan node, so such a call alone does not
show that the node read
+ * through it.
*
* Spark currently tracks runtime-filter eligibility by root attribute. If
[[filterAttributes]]
* returns a nested reference, an expression may access another nested field
under the same root.
* The scan must match each access against its own partition layout and use
only expressions it
* can apply.
*
- * 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.
- *
* 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.
+ * the originally reported partitioning. While applying runtime predicates,
the scan may detect
+ * that some [[InputPartition]]s have no matching data, in which case it can
either replace
+ * them with empty [[InputPartition]]s, or report only a subset of the
original partition
+ * values (omitting those with no data). The scan must not report new
partition values that
+ * were not present in the original partitioning.
*/
- def filter(expressions: Array[Expression]): Unit
+ def planInputPartitionsWithRuntimeFilters(
+ expressions: Array[Expression]): Array[InputPartition]
Review Comment:
Done in e721ae7. "DPP and a scalar subquery prune input partitions at
runtime for v2 text" uses its own single-column text fixture, and `AvroV2Suite`
has "SPARK-30628: Avro source v2: runtime filters prune partitions". Both check
that only partition 7 is read.
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2FileSourceDPPSuite.scala:
##########
@@ -0,0 +1,787 @@
+/*
+ * 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 java.io.File
+
+import scala.collection.mutable
+
+import org.apache.spark.{SparkException, SparkThrowable}
+import org.apache.spark.sql.{DataFrame, QueryTest, Row}
+import org.apache.spark.sql.catalyst.expressions.{AttributeReference,
DynamicPruningExpression, DynamicPruningSubquery, EqualTo, Expression, Literal}
+import org.apache.spark.sql.catalyst.plans.logical.{Filter, LogicalPlan}
+import org.apache.spark.sql.connector.read.InputPartition
+import org.apache.spark.sql.execution.FilterExec
+import org.apache.spark.sql.execution.adaptive.{AdaptiveSparkPlanHelper,
DisableAdaptiveExecutionSuite, EnableAdaptiveExecutionSuite}
+import org.apache.spark.sql.execution.datasources.FilePartition
+import org.apache.spark.sql.execution.datasources.v2.{BatchScanExec,
DataSourceV2ScanRelation, FileScan, RuntimePartitionPruningFileScan}
+import org.apache.spark.sql.execution.datasources.v2.parquet.ParquetScan
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.test.SharedSparkSession
+import org.apache.spark.sql.types.{IntegerType, LongType}
+
+/**
+ * Tests for SPARK-30628: subquery partition pruning and DPP for V2 file
sources.
+ *
+ * These tests exercise V2 file scans (e.g. ParquetScan) -- distinct from the
in-memory V2
+ * catalog used by [[org.apache.spark.sql.DynamicPartitionPruningV2Suite]] and
from the V2
+ * iterative-pushdown work covered by
[[DataSourceV2EnhancedRuntimePartitionFilterSuite]].
+ */
+abstract class DataSourceV2FileSourceDPPSuiteBase extends QueryTest
+ with SharedSparkSession
+ with AdaptiveSparkPlanHelper {
+
+ // Standard V2-only conf for DPP tests on file sources. AQE is set by the
concrete subclasses,
+ // which mix in DisableAdaptiveExecutionSuite /
EnableAdaptiveExecutionSuite: the DPP subquery is
+ // planned by PlanDynamicPruningFilters without AQE and by
PlanAdaptiveDynamicPruningFilters with
+ // it, and those are two separate rules.
+ private def withDppV2Conf[T](thunk: => T): T = {
+ withSQLConf(
+ SQLConf.USE_V1_SOURCE_LIST.key -> "",
+ SQLConf.DYNAMIC_PARTITION_PRUNING_ENABLED.key -> "true",
+ SQLConf.DYNAMIC_PARTITION_PRUNING_REUSE_BROADCAST_ONLY.key -> "false",
+ SQLConf.DYNAMIC_PARTITION_PRUNING_FALLBACK_FILTER_RATIO.key ->
"2")(thunk)
+ }
+
+ // Writes a partitioned `fact` and an unpartitioned `dim` table in `format`
and registers them
+ // as temp views. `fact` has 100 rows across 10 partitions (10 rows each);
`dim` has 10 rows
+ // with `dim_id` and `dim_val` both in [0, 10). The read schema is given
explicitly because csv
+ // infers strings, and a cast on the join key changes whether DPP fires at
all.
+ private def writeFactAndDim(dir: File, format: String = "parquet"): Unit = {
+ val factPath = new File(dir, "fact").getCanonicalPath
+ val dimPath = new File(dir, "dim").getCanonicalPath
+ spark.range(100)
+ .selectExpr("id", "id % 10 AS part")
+ .write.format(format).partitionBy("part").save(factPath)
+ spark.read.format(format).schema("id long, part int").load(factPath)
+ .createOrReplaceTempView("fact")
+ spark.range(10)
+ .selectExpr("id AS dim_id", "id AS dim_val")
+ .write.format(format).save(dimPath)
+ spark.read.format(format).schema("dim_id long, dim_val long").load(dimPath)
+ .createOrReplaceTempView("dim")
Review Comment:
Done in e721ae7. `DataSourceV2FileSourceDPPSuite` clears the temp views
after each test, and `withFactAndDim` in `DataSourceV2FileScanSubclassSuite`
uses `withTempView`.
--
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]
