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. The file scans no longer declare
`fullyPushedFilterAttributes()`, so a scalar-subquery filter keeps its
`FilterExec`, and a node that replaces `BatchScanExec` without pruning still
returns the right rows. Since every runtime filter is now evaluated again above
the scan, a filter that fails to evaluate on a directory keeps it, whether it
is a DPP filter or a scalar-subquery filter. "a scalar subquery filter on a
partition column is evaluated again after the scan" checks the `FilterExec`.
`PushDownUtils.pushRuntimeFilters` now also returns false for these scans
instead of failing, so a node that pushes through it and then plans the scan
reads every partition and the join or the `FilterExec` drops the extra rows.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/RuntimePartitionPruningFileScan.scala:
##########
@@ -0,0 +1,136 @@
+/*
+ * 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.execution.datasources.v2
+
+import java.util.Locale
+import java.util.concurrent.atomic.AtomicReference
+
+import org.apache.spark.SparkException
+import org.apache.spark.sql.catalyst.expressions.{And, Attribute,
BoundReference, Expression, Predicate}
+import org.apache.spark.sql.connector.expressions.{FieldReference,
NamedReference}
+import org.apache.spark.sql.connector.read.InputPartition
+import org.apache.spark.sql.execution.datasources.FilePartition
+import org.apache.spark.sql.internal.connector.SupportsRuntimeCatalystFiltering
+
+/**
+ * A [[FileScan]] that prunes its partition directories with runtime filters,
the DPP filters and
+ * scalar subquery filters Spark derives over its partition columns
(SPARK-30628). The built-in
+ * file formats mix this in; a [[FileScan]] that does not extend one of their
scans keeps its
+ * previous behavior.
+ *
+ * The runtime-filter path returns plain [[FilePartition]]s packed from what
`partitions` planned,
+ * so a scan that overrides `planInputPartitions()` or reports a partitioning
or an ordering should
+ * not mix this in.
+ */
+trait RuntimePartitionPruningFileScan extends FileScan with
SupportsRuntimeCatalystFiltering {
+
+ @transient private lazy val plannedPartitionsRef = new
AtomicReference[Seq[FilePartition]]()
+
+ /**
+ * The partitions `partitions` planned, stored once per scan instance. Both
the plain read path
+ * and the runtime-filter path go through this, so a scan node that gets
runtime filters lists the
+ * files once and filters that listing, rather than reading the index twice
and risking two
+ * snapshots. A subclass still customizes `partitions`.
+ *
+ * `partitions` runs outside the scan's monitor, which a `lazy val` would
hold while it runs. Two
+ * threads asking first at once may both list; both get the result that was
stored.
+ */
+ private def plannedPartitions: Seq[FilePartition] = {
+ val ref = plannedPartitionsRef
+ Option(ref.get()).getOrElse {
+ ref.compareAndSet(null, partitions)
+ ref.get()
+ }
+ }
+
+ override def planInputPartitions(): Array[InputPartition] = {
+ plannedPartitions.toArray
+ }
+
+ /**
+ * The partition columns Spark can derive a runtime filter on, restricted to
the ones
+ * `readSchema()` still exposes: a reference missing from the scan relation
output fails to
+ * resolve, and a pushed-down aggregate keeps only the partition columns it
groups by.
+ *
+ * A filter over one of them is evaluated against each file's partition
values, so the scan
+ * evaluates it in full and Spark does not evaluate it again after the scan
--
+ * `FileScanBuilder.pushFilters` already keeps compile-time partition
filters out of the post-scan
+ * filters for the same reason.
+ */
+ override def filterAttributes(): Array[NamedReference] = {
+ val readFields = readSchema().fieldNames.map(normalizePartitionName).toSet
+ readPartitionSchema.fieldNames
+ .filter(name => readFields.contains(normalizePartitionName(name)))
+ .map(FieldReference.column)
+ }
+
+ override def fullyPushedFilterAttributes(): Array[NamedReference] =
filterAttributes()
Review Comment:
Done in e721ae7. `TextBasedFileScan` is back to what master has, and
`CSVScan`, `JsonScan` and `TextScan` mix in the trait themselves.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/TextBasedFileScan.scala:
##########
@@ -27,7 +27,7 @@ import org.apache.spark.util.Utils
abstract class TextBasedFileScan(
sparkSession: SparkSession,
- options: CaseInsensitiveStringMap) extends FileScan {
+ options: CaseInsensitiveStringMap) extends RuntimePartitionPruningFileScan
{
Review Comment:
This went away in 7158934. The runtime path now filters the listed
directories and passes them to `filePartitions`, which sums the length and open
cost per file, as V1 does.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/RuntimePartitionPruningFileScan.scala:
##########
@@ -0,0 +1,136 @@
+/*
+ * 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.execution.datasources.v2
+
+import java.util.Locale
+import java.util.concurrent.atomic.AtomicReference
+
+import org.apache.spark.SparkException
+import org.apache.spark.sql.catalyst.expressions.{And, Attribute,
BoundReference, Expression, Predicate}
+import org.apache.spark.sql.connector.expressions.{FieldReference,
NamedReference}
+import org.apache.spark.sql.connector.read.InputPartition
+import org.apache.spark.sql.execution.datasources.FilePartition
+import org.apache.spark.sql.internal.connector.SupportsRuntimeCatalystFiltering
+
+/**
+ * A [[FileScan]] that prunes its partition directories with runtime filters,
the DPP filters and
+ * scalar subquery filters Spark derives over its partition columns
(SPARK-30628). The built-in
+ * file formats mix this in; a [[FileScan]] that does not extend one of their
scans keeps its
+ * previous behavior.
+ *
+ * The runtime-filter path returns plain [[FilePartition]]s packed from what
`partitions` planned,
+ * so a scan that overrides `planInputPartitions()` or reports a partitioning
or an ordering should
+ * not mix this in.
+ */
+trait RuntimePartitionPruningFileScan extends FileScan with
SupportsRuntimeCatalystFiltering {
+
+ @transient private lazy val plannedPartitionsRef = new
AtomicReference[Seq[FilePartition]]()
+
+ /**
+ * The partitions `partitions` planned, stored once per scan instance. Both
the plain read path
+ * and the runtime-filter path go through this, so a scan node that gets
runtime filters lists the
+ * files once and filters that listing, rather than reading the index twice
and risking two
+ * snapshots. A subclass still customizes `partitions`.
+ *
+ * `partitions` runs outside the scan's monitor, which a `lazy val` would
hold while it runs. Two
+ * threads asking first at once may both list; both get the result that was
stored.
+ */
+ private def plannedPartitions: Seq[FilePartition] = {
+ val ref = plannedPartitionsRef
+ Option(ref.get()).getOrElse {
+ ref.compareAndSet(null, partitions)
+ ref.get()
+ }
+ }
+
+ override def planInputPartitions(): Array[InputPartition] = {
+ plannedPartitions.toArray
+ }
+
+ /**
+ * The partition columns Spark can derive a runtime filter on, restricted to
the ones
+ * `readSchema()` still exposes: a reference missing from the scan relation
output fails to
+ * resolve, and a pushed-down aggregate keeps only the partition columns it
groups by.
+ *
+ * A filter over one of them is evaluated against each file's partition
values, so the scan
+ * evaluates it in full and Spark does not evaluate it again after the scan
--
+ * `FileScanBuilder.pushFilters` already keeps compile-time partition
filters out of the post-scan
+ * filters for the same reason.
+ */
+ override def filterAttributes(): Array[NamedReference] = {
+ val readFields = readSchema().fieldNames.map(normalizePartitionName).toSet
+ readPartitionSchema.fieldNames
+ .filter(name => readFields.contains(normalizePartitionName(name)))
+ .map(FieldReference.column)
+ }
+
+ override def fullyPushedFilterAttributes(): Array[NamedReference] =
filterAttributes()
+
+ /**
+ * Plans the files whose partition values satisfy `expressions`, taken from
what `partitions`
+ * planned and bin-packed again for the surviving set.
+ *
+ * Filtering what was planned rather than listing the files again keeps one
listing per scan, and
+ * the files a subclass excluded in `partitions` stay excluded. How the
survivors group into
+ * partitions is decided here again, for their own size. Where each file was
cut into splits is
+ * not: `partitions` cut them for the whole set it listed, before these
filters existed.
+ *
+ * A partition value is fixed within a file, so every row of every partition
returned satisfies
+ * the expressions even though one partition can still hold files from
several partition
+ * directories.
+ */
+ override def planInputPartitionsWithRuntimeFilters(
+ expressions: Array[Expression]): Array[InputPartition] = {
+ val fieldIndex = readPartitionSchema.fieldNames.zipWithIndex
+ .map { case (name, i) => normalizePartitionName(name) -> i }.toMap
+ // Evaluating these is the scan's only chance to apply them, since the
attributes are declared
+ // fully pushed and Spark drops the post-scan filter. A reference outside
the read partition
+ // schema has no value to evaluate against, so fail loudly rather than
return wrong rows. Spark
+ // screens against `filterAttributes()` before it gets here, which is a
subset of that schema.
+ val notApplicable = expressions.filterNot(
+ _.references.forall(a =>
fieldIndex.contains(normalizePartitionName(a.name))))
+ if (notApplicable.nonEmpty) {
+ throw SparkException.internalError("A file scan can only apply a runtime
filter over its " +
+ s"read partition columns
${readPartitionSchema.fieldNames.mkString("[", ", ", "]")}, " +
+ s"got ${notApplicable.mkString(", ")}")
+ }
+ val bound = expressions.reduce(And).transform {
+ case a: Attribute =>
+ BoundReference(fieldIndex(normalizePartitionName(a.name)), a.dataType,
a.nullable)
+ }
+ val predicate = Predicate.createInterpreted(bound)
+ val keptFiles = plannedPartitions
+ .flatMap(_.files.filter(file => predicate.eval(file.partitionValues)))
+ .sortBy(_.length)(implicitly[Ordering[Long]].reverse)
+ // Pack the survivors instead of handing back the planned partitions with
files removed:
+ // `partitions` sized those bins for the whole file set, so a scan that
pruned most of its files
+ // would run the same few tasks over a fraction of the data. V1 packs the
pruned set the same
+ // way, from `dynamicallySelectedPartitions`.
+ val openCostInBytes = conf.filesOpenCostInBytes
+ val maxSplitBytes = FilePartition.maxSplitBytes(
+ sparkSession, keptFiles.map(_.length + openCostInBytes).sum)
Review Comment:
Done in e721ae7, with one change from your snippet. Since 7158934 the index
is into `fileIndex.partitionSchema`, the schema the directory values follow, so
the type comes from there rather than from `readPartitionSchema`.
--
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]
