dongjoon-hyun commented on code in PR #59715:
URL: https://github.com/apache/spark/pull/59715#discussion_r4372009016
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceRDD.scala:
##########
@@ -187,10 +191,16 @@ class DataSourceRDD(
castPartition(split).inputPartition.iterator.flatMap { inputPartition =>
val iter = if (columnarReads) {
- val batchReader =
partitionReaderFactory.createColumnarReader(inputPartition)
+ val batchReader = storageFilterRequest match {
+ case Some(request) =>
partitionReaderFactory.createColumnarReader(inputPartition, request)
+ case None =>
partitionReaderFactory.createColumnarReader(inputPartition)
+ }
new PartitionIterator[ColumnarBatch](batchReader, _.numRows, taskState)
} else {
- val rowReader = partitionReaderFactory.createReader(inputPartition)
+ val rowReader = storageFilterRequest match {
+ case Some(request) =>
partitionReaderFactory.createReader(inputPartition, request)
Review Comment:
**2. [Low] Nothing checks that a reader appends the checked column it was
asked for**
When `appendsCheckedColumn()` is true, the scan's `output` ends with the
checked column. But nothing checks that the reader returned here actually
appends it. An easy mistake is a stub that takes the request and delegates to
`createReader(partition)`. Then:
- For an `UnsafeRow` of `n` fields, the Filter's generated code reads
ordinal `n`. `assertIndexIsValid` is a Java `assert`, which is off in
production, so `isNullAt(n)` and `getBoolean(n)` read the bytes after the
fixed-length region, and the bloom is skipped on arbitrary rows.
- A `GenericInternalRow`, or `batch.column(n)` in `ColumnarToRowExec`'s
generated code, throws an `ArrayIndexOutOfBoundsException` that names neither
the connector nor the request.
Today the join still filters, so only the contract breaks, but the failure
is hard to trace back. Could the first row or batch of each such reader be
checked against `output.length`, the way `StorageFilterImpl.eval` checks its
own row width, and fail with an error that names the factory class?
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/connector/StorageFilterImpl.scala:
##########
@@ -0,0 +1,132 @@
+/*
+ * 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 scala.util.control.NonFatal
+
+import org.apache.spark.{SparkEnv, SparkException, TaskContext}
+import org.apache.spark.executor.Executor
+import org.apache.spark.internal.{Logging, LogKeys}
+import org.apache.spark.internal.config.KILL_ON_FATAL_ERROR_DEPTH
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions.{Attribute, BindReferences,
Expression => CatalystExpression, Predicate => CatalystPredicate}
+import org.apache.spark.sql.connector.expressions.{FieldReference,
NamedReference}
+import org.apache.spark.sql.connector.expressions.filter.{StorageFilter,
StorageFilterNotAppliedException}
+
+/**
+ * A [[StorageFilter]] wrapping a Catalyst expression, so a scan can apply a
predicate it does not
+ * have to understand. It mirrors [[PartitionPredicateImpl]], which evaluates
a partition key
+ * rather than a data row.
+ *
+ * `attrs` is the layout of the row [[eval]] expects, in that order, and the
expression is bound to
+ * it. The binding is built on first use, so an instance created on the driver
can be evaluated on
+ * an executor.
+ */
+class StorageFilterImpl private (
+ private val catalystExpr: CatalystExpression,
+ private val attrs: Seq[Attribute])
+ extends StorageFilter with Logging {
+
+ // The width of the row `eval` takes.
+ private val numReferences = attrs.length
+
+ // Whether this instance has logged a failure, so a task logs only its first.
+ @transient private var loggedFailure = false
+
+ @transient private lazy val boundPredicate: InternalRow => Boolean = {
+ val bound = BindReferences.bindReference(catalystExpr, attrs)
+ CatalystPredicate.create(bound).eval
+ }
+
+ /**
+ * Wraps a non-fatal error in [[StorageFilterNotAppliedException]], unless
the task is being
+ * killed or a fatal error is in its cause chain. Each task logs the first
one. A row of another
+ * width than [[references]] is a reader bug, which fails the task, since
the predicate would read
+ * the wrong values.
+ */
+ @throws[StorageFilterNotAppliedException]
+ override def eval(row: InternalRow): Boolean = {
+ // Outside the `try`, so a failure to build the predicate, or a row of
another width, fails the
+ // task.
+ if (row.numFields != numReferences) {
+ throw SparkException.internalError(
+ s"A storage filter over ${references.mkString("[", ", ", "]")} takes
rows of " +
+ s"$numReferences values, but got ${row.numFields}")
+ }
+ val predicate = boundPredicate
+ try {
+ predicate(row)
+ } catch {
+ case NonFatal(e) if !StorageFilterImpl.taskKilled &&
!StorageFilterImpl.hasFatalCause(e) =>
Review Comment:
**1. [Medium] `eval` does not follow the V1 error rules it says it mirrors**
The comment at L120-121 and the PR description say a fatal error and a task
kill go out "as from the V1 reader", but this `catch` only wraps a `NonFatal`
error and lets everything else out as it was thrown.
`ParquetStorageFilter.rethrowIfMustPropagate(e, fromRead = false)` differs in
four cases:
| Error from the predicate | V1 reader | `StorageFilterImpl.eval` |
|---|---|---|
| `NonFatal` wrapper around a fatal error | throws the fatal error itself |
throws the wrapper |
| bare `InternalError` | throws it in an `UnknownError` | throws it as is |
| `NonFatal` error while a kill is pending | throws a checked
`SparkException` around it | throws it as is |
| `LinkageError`, `SparkOutOfMemoryError`, `InterruptedException` without a
kill | gives the filter up | fails the task |
V1 converts the first three on purpose (ParquetStorageFilter.scala:228-230
and 276-280). Spark's own `FilePartitionReader` takes any `RuntimeException |
IOException | InternalError` from a reader for a corrupt file under
`spark.sql.files.ignoreCorruptFiles=true` (FilePartitionReader.scala:74-77,
DataSourceUtils.scala:273). So take a `FileScan`-based source that adopts this
API, with a per-file reader that catches only
`StorageFilterNotAppliedException` as the Javadoc asks. If the bloom raises
`IllegalStateException(cause = OutOfMemoryError)` or an `InternalError`, that
reader logs "Skipped the rest of the content in the corrupted file" and
silently drops the rest of the file, and the executor keeps running after the
OOM. The same evaluation in the post-scan `Filter` fails the task.
No built-in scan implements the mix-in yet, so this is latent. But the two
suites already pin opposite outcomes for the same input:
StorageFilterImplSuite.scala:112-125 asserts `e eq error`, and a raw
`SparkArithmeticException` under a kill, while
ParquetStorageFilterSuite.scala:645-659 asserts a checked `SparkException` and
the wrapped `OutOfMemoryError` itself.
Could `eval` catch `Throwable` and apply the `fromRead = false` half of
`rethrowIfMustPropagate` (with `throwIfKilled` and `fatalErrorDepth`) before it
wraps the error, so that both paths share one rule? If the narrower rule is
intended, the comment at L120-121, the PR description and the `@throws` text of
`StorageFilter.eval` should say so.
##########
sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/filter/StorageFilterNotAppliedException.java:
##########
@@ -0,0 +1,52 @@
+/*
+ * 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.expressions.filter;
+
+import java.util.Map;
+
+import org.apache.spark.SparkThrowable;
+import org.apache.spark.SparkThrowableHelper;
+import org.apache.spark.annotation.Evolving;
+
+/**
+ * Thrown by a {@link StorageFilter} that cannot be applied to a row, for
instance because
+ * evaluating it raised an error. The reader catches it and reads the rest of
the unit it was
+ * filtering, such as a file or a row group, without the filter. The cause is
the error the filter
+ * met.
+ * <p>
+ * It is checked, so a reader handles it where it can fall back to reading
without the filter. A
Review Comment:
**5. [Nit] "It is checked" does not help a Scala reader**
Scala does not enforce checked exceptions. So a Scala `PartitionReader` that
calls `filter.eval` or `select` without a `catch` compiles without a warning.
On the first evaluation error it then fails the task with
`STORAGE_FILTER_NOT_APPLIED`, instead of reading the unit without the filter.
The test readers have to remember the `catch` by hand
(StorageFilterTestSource.scala:222 and :275). Maybe add a sentence here and in
`docs/sql-v2-data-sources.md` that a reader written in Scala must catch it
itself.
##########
sql/catalyst/src/main/java/org/apache/spark/sql/connector/read/PartitionReaderFactory.java:
##########
@@ -56,10 +56,40 @@ default PartitionReader<ColumnarBatch>
createColumnarReader(InputPartition parti
"DATA_SOURCE_COLUMNAR_READER_NOT_SUPPORTED", Map.of("class",
getClass().getName()));
}
+ /**
+ * Returns a row-based partition reader that may apply the storage filter of
{@code request} while
+ * it reads the given {@link InputPartition}. Spark calls this instead of
+ * {@link #createReader(InputPartition)} for a {@link
SupportsStorageFilters} scan it offered a
+ * filter, so the factory of such a scan must implement it for row-based
reads.
+ *
+ * @since 5.0.0
+ */
+ default PartitionReader<InternalRow> createReader(
Review Comment:
**3. [Nit] A factory without the new overload is found only inside the
tasks**
Spark calls this overload because the `Scan` declares
`storageFilterAttributes()`, while the factory comes from the `Batch`. Take a
factory that does not override the overload for the scan's read mode, e.g. a
delegating or metrics-wrapping one. It fails here in every task, after the
bloom's build side has run, and only after `spark.task.maxFailures` retries.
Small test tables rarely trigger `InjectRuntimeFilter`, so this can show up
first in production.
`BatchScanExec` knows `readerFactory` and `supportsColumnar` on the driver.
So `inputRDD` could check once, e.g. through `getDeclaringClass` of the
overload, and fail with the same error before any task starts, while EXPLAIN
keeps working. The existing `createColumnarReader(partition)` default behaves
the same way, so this is only a suggestion.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Strategy.scala:
##########
@@ -200,10 +223,20 @@ class DataSourceV2Strategy(session: SparkSession) extends
Strategy with Predicat
// rewritten it to TrueLiteral by the time we get here.
val runtimeFilters = dynamicFilters ++ scalarSubqueryFilters
- val batchExec = BatchScanExec(relation.output, relation.scan,
runtimeFilters,
- relation.ordering, relation.relation.table,
relation.keyGroupedPartitioning)
- DataSourceV2Strategy.withProjectAndFilter(
- project, postScanFilters.diff(fullyPushedRuntimeFilters),
+ // Whether the scan gets the checked column does not depend on whether
it is columnar. Asking
+ // that can plan the scan's partitions, which the node built afterwards
would plan again. A
+ // row-based scan gets a Project for the UnsafeRow conversion anyway, so
dropping the column
+ // there copies nothing more.
+ val filtersAbove = postScanFilters.diff(fullyPushedRuntimeFilters)
+ val storageFilters = storageFiltersFor(relation, filtersAbove)
+ val checked = StorageFilterPlanning.checkedColumn(
+ session.sessionState.conf, storageFilters, relation.output)
+ val batchExec = BatchScanExec(relation.output ++ checked, relation.scan,
runtimeFilters,
Review Comment:
**7. [Low] The subquery rules now re-plan every storage-filtered scan**
With the bloom's `ScalarSubquery` in `storageFilters`, the node built here
no longer survives planning:
- `PlanSubqueries` or `PlanAdaptiveSubqueries` rewrites it.
- Under AQE, `ReuseAdaptiveSubquery` rewrites it again at stage creation,
after `EnsureRequirements` (AdaptiveSparkPlanExec.scala:195,
ReuseAdaptiveSubquery.scala:38).
Each copy computes `batch`, `inputPartitions` and `readerFactory` again.
`fastEquals` forces `batch` through `equals`, and `RemoveRedundantProjects`,
`ApplyColumnarRulesAndInsertTransitions` or `EnsureRequirements` force the
partitions. With AQE on and the default `PARTITION_DEFINED` mode, a scan
offered a bloom runs `toBatch()` 4 times and `planInputPartitions()` /
`createReaderFactory()` 3 times. With the conf off, each runs 2 times. Every
AQE re-plan of such a scan pays the extra copy again. The comment at L226-229
avoids planning a probe node, but the node it keeps is replaced right away.
Two consequences:
- `Batch.planInputPartitions` still promises a single call (Batch.java:40),
and only the doc of `Scan.toBatch` was updated.
- Under a storage-partitioned join, the `ReuseAdaptiveSubquery` copy lands
under a `GroupPartitionsExec` that was planned over the previous node.
`checkChildStillMatches` then compares the re-planned keys by value
(GroupPartitionsExec.scala:127 and :352). A connector whose second planning
returns other keys fails with INTERNAL_ERROR, and that doc names only
`PlanAdaptiveDynamicPruningFilters` as a rule that rebuilds `BatchScanExec`.
The test source returns `this` from `toBatch` and a fixed partition, so no
test can see the extra calls. Could the planned `batch`, partitions and factory
carry over to copies of the same `scan`, e.g. through a holder passed in
`otherCopyArgs`? If not, the two docs above could name the new rebuilders.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Strategy.scala:
##########
@@ -200,10 +223,20 @@ class DataSourceV2Strategy(session: SparkSession) extends
Strategy with Predicat
// rewritten it to TrueLiteral by the time we get here.
val runtimeFilters = dynamicFilters ++ scalarSubqueryFilters
- val batchExec = BatchScanExec(relation.output, relation.scan,
runtimeFilters,
- relation.ordering, relation.relation.table,
relation.keyGroupedPartitioning)
- DataSourceV2Strategy.withProjectAndFilter(
- project, postScanFilters.diff(fullyPushedRuntimeFilters),
+ // Whether the scan gets the checked column does not depend on whether
it is columnar. Asking
+ // that can plan the scan's partitions, which the node built afterwards
would plan again. A
+ // row-based scan gets a Project for the UnsafeRow conversion anyway, so
dropping the column
+ // there copies nothing more.
+ val filtersAbove = postScanFilters.diff(fullyPushedRuntimeFilters)
+ val storageFilters = storageFiltersFor(relation, filtersAbove)
+ val checked = StorageFilterPlanning.checkedColumn(
+ session.sessionState.conf, storageFilters, relation.output)
+ val batchExec = BatchScanExec(relation.output ++ checked, relation.scan,
runtimeFilters,
+ relation.ordering, relation.relation.table,
relation.keyGroupedPartitioning,
+ storageFilters, checked)
+ // With the checked column the scan's output differs from `project`, so
a Project drops it.
+ DataSourceV2Strategy.withProjectAndFilter(project,
+ StorageFilterPlanning.skipCheckedRows(filtersAbove, storageFilters,
checked),
Review Comment:
**10. [Low] The bloom's subquery now occurs twice in the plan**
The same `ScalarSubquery` now sits in the scan's `storageFilters` and,
through this call, in the post-scan `Filter`.
- Without AQE, `PlanSubqueries` calls `QueryExecution.prepareExecutedPlan`
once per occurrence (subquery.scala:456). So the bloom's creation side is
physically planned twice, including the partition planning of its own scans.
- With `spark.sql.execution.reuseSubquery=false`, nothing dedups the two
`SubqueryExec`s, so the bloom aggregation runs twice. With AQE, its final stage
runs twice.
Running it twice needs a non-default conf, so this may be acceptable. If so,
could the PR description say so next to its note that the task binary carries
the bloom twice?
##########
docs/sql-v2-data-sources.md:
##########
@@ -319,7 +319,10 @@ for data locality.
is serialized to executors and creates a
[`PartitionReader`](api/java/org/apache/spark/sql/connector/read/PartitionReader.html)
for each
input partition. The reader iterates over rows (or `ColumnarBatch` instances
for columnar sources)
-and is closed after consumption.
+and is closed after consumption. For a `SupportsStorageFilters` scan that
Spark offered a storage
+filter, `createReader(partition, request)` or `createColumnarReader(partition,
request)` creates the
+reader instead. The `StorageFilterRequest` carries the filter, and whether the
reader appends a
+column that marks the rows it applied the filter to.
Review Comment:
**15. [Low] The guide says the column marks the rows the filter was applied
to**
The API docs say the mark is true only on a row the filter was evaluated on
and kept, and false on a row kept although the filter rejected it:
StorageFilterRequest.java:40-42, `StorageFilter.eval` and
`SupportsStorageFilters`. "the rows it applied the filter to" invites a reader
that filters per page or per batch to mark every row it evaluated, including
returned rows the bloom rejected. The post-scan `If(checked, true, bloom)` then
skips the bloom on exactly those rows. Today only the join, which applies the
bloom again, hides that.
Maybe: "... and whether the reader appends a boolean column that is true
only on the rows the filter was evaluated on and kept."
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/StorageFilterPushDownSuite.scala:
##########
@@ -0,0 +1,314 @@
+/*
+ * 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 scala.collection.mutable
+
+import org.apache.spark.sql.{DataFrame, QueryTest}
+import org.apache.spark.sql.catalyst.expressions.{Attribute, AttributeMap,
AttributeReference, AttributeSet, BloomFilterMightContain, Cast, Divide,
EvalMode, GreaterThan, If, Literal, NumericEvalContext, Rand}
+import org.apache.spark.sql.execution.{BaseLimitExec, FilterExec, SampleExec,
SparkPlan, TakeOrderedAndProjectExec}
+import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanHelper
+import org.apache.spark.sql.execution.datasources.{FileFormat,
StorageFilterPlanning}
+import org.apache.spark.sql.execution.datasources.v2.BatchScanExec
+import org.apache.spark.sql.execution.joins.SortMergeJoinExec
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.internal.connector.StorageFilterImpl
+import org.apache.spark.sql.test.SharedSparkSession
+import org.apache.spark.sql.types.{DoubleType, LongType}
+
+/**
+ * Covers the storage-filter contract for a data source v2 scan. With
+ * `V2_STORAGE_FILTER_PUSHDOWN_ENABLED` on, a scan that declares columns
through
+ * `SupportsStorageFilters` gets the runtime bloom filter over those columns
in every reader. The
+ * bloom stays in the Filter above the scan, skipped on the rows the reader
marks as checked.
+ */
+class StorageFilterPushDownSuite
+ extends QueryTest with SharedSparkSession with AdaptiveSparkPlanHelper {
+
+ private def sourceName: String = classOf[StorageFilterTestSource].getName
+
+ // Runs `body` with runtime blooms and V2 storage filters on, `conf` on top,
and the build table.
+ private def withBloomJoin(conf: (String, String)*)(body: => Unit): Unit = {
+ val bloomJoinConf = Seq(
+ SQLConf.V2_STORAGE_FILTER_PUSHDOWN_ENABLED.key -> "true",
+ SQLConf.RUNTIME_BLOOM_FILTER_ENABLED.key -> "true",
+ SQLConf.RUNTIME_BLOOM_FILTER_APPLICATION_SIDE_SCAN_SIZE_THRESHOLD.key ->
"1000",
+ SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1",
+ SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false")
+ withSQLConf(bloomJoinConf ++ conf: _*) {
+ withTable("bf_build") {
+ spark.range(0, 30).selectExpr("id AS bk", "id AS bv")
+ .write.format("parquet").saveAsTable("bf_build")
+ body
+ }
+ }
+ }
+
+ // A join whose build side is small and carries a selective predicate, which
is the shape
+ // InjectRuntimeFilter builds a bloom filter for. The bloom lands on
`probe`'s `k`.
+ private def bloomJoin(probe: DataFrame): DataFrame = {
+ val build = spark.table("bf_build").where("bv = 5")
+ probe.join(build, probe("k") ===
build("bk")).select(probe.columns.map(probe(_)).toSeq: _*)
+ }
+
+ private def v2Probe(options: (String, String)*): DataFrame =
+ spark.read.format(sourceName).options(options.toMap).load()
+
+ private val checkedName = FileFormat.STORAGE_FILTER_CHECKED_COLUMN_NAME
+
+ private def scanOf(plan: SparkPlan): BatchScanExec =
+ collect(plan) { case b: BatchScanExec => b }.headOption
+ .getOrElse(fail(s"no BatchScanExec in plan:\n$plan"))
+
+ // The blooms the Filters above the scan evaluate on every row, and those
they skip on the rows
+ // the reader marks as checked.
+ private def bloomsAbove(plan: SparkPlan): (Int, Int) = {
+ val conditions = collect(plan) { case f: FilterExec => f.condition }
+ val all = conditions.map(_.collect { case _: BloomFilterMightContain => 1
}.sum).sum
+ val skipped = conditions.map(_.collect {
+ case If(a: Attribute, Literal.TrueLiteral, _: BloomFilterMightContain)
+ if a.name == checkedName => 1
+ }.sum).sum
+ (all - skipped, skipped)
+ }
+
+ // The rows the Filter on the scan returned, counted before the join.
+ private def rowsOutOfFilter(plan: SparkPlan): Long = {
+ val filters = collect(plan) {
+ case f: FilterExec if f.exists(_.isInstanceOf[BatchScanExec]) => f
+ }
+ assert(filters.size == 1, s"expected one Filter on the scan:\n$plan")
+ filters.head.metrics("numOutputRows").value
+ }
+
+ // Runs the bloom join over the v2 source read with `options`, and returns
its rows and plan.
+ private def runBloomJoin(options: (String, String)*): (Set[(Long, Long)],
SparkPlan) = {
+ StorageFilterTestSource.reset()
+ val df = bloomJoin(v2Probe(options: _*))
+ (df.collect().map(r => (r.getLong(0), r.getLong(1))).toSet,
df.queryExecution.executedPlan)
+ }
+
+ // The build side keeps `bv = 5`, so the join returns that one row.
+ private val joinResult = Set((5L, 50L))
+
+ for (columnar <- Seq(false, true); aqe <- Seq(false, true)) {
+ val kind = if (columnar) "columnar" else "row-based"
+ test(s"a $kind scan gets the checked column, and the Filter skips the
bloom on checked rows " +
+ s"(AQE $aqe)") {
+ withBloomJoin(SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> aqe.toString) {
+ val (result, plan) = runBloomJoin("columnar" -> columnar.toString)
+ val scan = scanOf(plan)
+ assert(scan.supportsColumnar == columnar, s"the scan must read
$kind:\n$plan")
+ assert(scan.storageFilters.size == 1,
+ s"the bloom filter must be on the scan; got
${scan.storageFilters}\n$plan")
+ assert(scan.checkedColumn.isDefined && scan.checkedColumn ==
scan.output.lastOption,
+ s"the scan must get the checked column, last:\n$plan")
+ assert(bloomsAbove(plan) == (0, 1),
+ s"the Filter must skip the bloom on checked rows:\n$plan")
+ assert(rowsOutOfFilter(plan) ==
StorageFilterTestSource.rowsReturned.get(),
+ s"the Filter must keep every checked row; got
${rowsOutOfFilter(plan)} of " +
+ s"${StorageFilterTestSource.rowsReturned.get()}")
+ val probeSide = collectFirst(plan) { case j: SortMergeJoinExec =>
j.left.output }
+ assert(probeSide.exists(!_.exists(_.name == checkedName)),
+ s"the Project above the scan must drop the checked column; got
$probeSide")
+ assert(StorageFilterTestSource.filteredReaders.get() == 1,
+ s"the one partition's reader must get the filter; " +
+ s"got ${StorageFilterTestSource.filteredReaders.get()}")
+ assert(StorageFilterTestSource.checkedColumnRequests.get() == 1,
+ "the reader must be asked for the checked column")
+ assert(result == joinResult, s"got $result")
+ assert(StorageFilterTestSource.rowsEvaluated.get() ==
StorageFilterTestSource.numRows,
+ s"the reader must evaluate every row; got
${StorageFilterTestSource.rowsEvaluated.get()}")
+ assert(StorageFilterTestSource.rowsReturned.get() <
StorageFilterTestSource.numRows,
+ "the reader must have dropped rows, otherwise the filter did
nothing")
+ }
+ }
+ }
+
+ Seq(false, true).foreach { columnar =>
+ val kind = if (columnar) "columnar" else "row-based"
+ test(s"a $kind reader that gives the filter up marks its rows unchecked,
and the Filter " +
+ "filters them") {
+ withBloomJoin() {
+ val (result, plan) = runBloomJoin("columnar" -> columnar.toString,
"giveUp" -> "true")
+ assert(bloomsAbove(plan) == (0, 1), s"the plan is that of a filtering
reader:\n$plan")
+ assert(StorageFilterTestSource.rowsEvaluated.get() == 0,
+ s"the reader must evaluate nothing; got
${StorageFilterTestSource.rowsEvaluated.get()}")
+ assert(StorageFilterTestSource.rowsReturned.get() ==
StorageFilterTestSource.numRows,
+ s"the reader must return every row; got
${StorageFilterTestSource.rowsReturned.get()}")
+ assert(rowsOutOfFilter(plan) < StorageFilterTestSource.numRows,
+ s"the Filter must apply the bloom to unchecked rows; got
${rowsOutOfFilter(plan)}")
+ assert(result == joinResult, s"got $result")
+ }
+ }
+
+ test(s"the Filter takes a $kind reader's word on the rows it marks
checked") {
+ // This reader breaks the contract on purpose: it applies no filter but
marks every row
+ // checked. Only skipping the bloom on those rows keeps all of them.
+ withBloomJoin() {
+ val (result, plan) =
+ runBloomJoin("columnar" -> columnar.toString, "markAllChecked" ->
"true")
+ assert(rowsOutOfFilter(plan) == StorageFilterTestSource.numRows,
+ s"the Filter must keep every row marked checked; got
${rowsOutOfFilter(plan)}")
+ assert(result == joinResult, s"the join must still filter; got
$result")
+ }
+ }
+ }
+
+ test("a filter that cannot be applied is given up, and its rows come back
unchecked") {
Review Comment:
**14. [Nit] This test exercises only the fixture**
It builds the same ANSI `1 / k > 0` filter as
`StorageFilterImplSuite.positiveInverse`, and drives
`StorageFilterTestColumnarReader` by hand. StorageFilterImplSuite's "an error
evaluating the filter answers that it is not applied" already covers the
filter. So the only new code this test covers is the fixture's own `catch`, and
it would pass even if `DataSourceRDD` never called the new overload. The row
reader's identical `catch` (StorageFilterTestSource.scala:222-226) never runs.
Yet the PR description counts this test as coverage of "a filter that throws
`StorageFilterNotAppliedException` is given up".
Smaller things in this suite:
- Every caller of `runBloomJoin` asserts `result == joinResult` right after
it.
- The "nothing offered" assertion pair is copied into two loops.
- `checkedName` (L72) uses the V1 alias
`FileFormat.STORAGE_FILTER_CHECKED_COLUMN_NAME` rather than
`SupportsStorageFilters.CHECKED_COLUMN_NAME`.
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/StorageFilterPushDownSuite.scala:
##########
@@ -0,0 +1,314 @@
+/*
+ * 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 scala.collection.mutable
+
+import org.apache.spark.sql.{DataFrame, QueryTest}
+import org.apache.spark.sql.catalyst.expressions.{Attribute, AttributeMap,
AttributeReference, AttributeSet, BloomFilterMightContain, Cast, Divide,
EvalMode, GreaterThan, If, Literal, NumericEvalContext, Rand}
+import org.apache.spark.sql.execution.{BaseLimitExec, FilterExec, SampleExec,
SparkPlan, TakeOrderedAndProjectExec}
+import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanHelper
+import org.apache.spark.sql.execution.datasources.{FileFormat,
StorageFilterPlanning}
+import org.apache.spark.sql.execution.datasources.v2.BatchScanExec
+import org.apache.spark.sql.execution.joins.SortMergeJoinExec
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.internal.connector.StorageFilterImpl
+import org.apache.spark.sql.test.SharedSparkSession
+import org.apache.spark.sql.types.{DoubleType, LongType}
+
+/**
+ * Covers the storage-filter contract for a data source v2 scan. With
+ * `V2_STORAGE_FILTER_PUSHDOWN_ENABLED` on, a scan that declares columns
through
+ * `SupportsStorageFilters` gets the runtime bloom filter over those columns
in every reader. The
+ * bloom stays in the Filter above the scan, skipped on the rows the reader
marks as checked.
+ */
+class StorageFilterPushDownSuite
+ extends QueryTest with SharedSparkSession with AdaptiveSparkPlanHelper {
+
+ private def sourceName: String = classOf[StorageFilterTestSource].getName
+
+ // Runs `body` with runtime blooms and V2 storage filters on, `conf` on top,
and the build table.
+ private def withBloomJoin(conf: (String, String)*)(body: => Unit): Unit = {
+ val bloomJoinConf = Seq(
+ SQLConf.V2_STORAGE_FILTER_PUSHDOWN_ENABLED.key -> "true",
+ SQLConf.RUNTIME_BLOOM_FILTER_ENABLED.key -> "true",
+ SQLConf.RUNTIME_BLOOM_FILTER_APPLICATION_SIDE_SCAN_SIZE_THRESHOLD.key ->
"1000",
+ SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1",
+ SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false")
+ withSQLConf(bloomJoinConf ++ conf: _*) {
+ withTable("bf_build") {
+ spark.range(0, 30).selectExpr("id AS bk", "id AS bv")
+ .write.format("parquet").saveAsTable("bf_build")
+ body
+ }
+ }
+ }
+
+ // A join whose build side is small and carries a selective predicate, which
is the shape
+ // InjectRuntimeFilter builds a bloom filter for. The bloom lands on
`probe`'s `k`.
+ private def bloomJoin(probe: DataFrame): DataFrame = {
+ val build = spark.table("bf_build").where("bv = 5")
+ probe.join(build, probe("k") ===
build("bk")).select(probe.columns.map(probe(_)).toSeq: _*)
+ }
+
+ private def v2Probe(options: (String, String)*): DataFrame =
+ spark.read.format(sourceName).options(options.toMap).load()
+
+ private val checkedName = FileFormat.STORAGE_FILTER_CHECKED_COLUMN_NAME
+
+ private def scanOf(plan: SparkPlan): BatchScanExec =
+ collect(plan) { case b: BatchScanExec => b }.headOption
+ .getOrElse(fail(s"no BatchScanExec in plan:\n$plan"))
+
+ // The blooms the Filters above the scan evaluate on every row, and those
they skip on the rows
+ // the reader marks as checked.
+ private def bloomsAbove(plan: SparkPlan): (Int, Int) = {
+ val conditions = collect(plan) { case f: FilterExec => f.condition }
+ val all = conditions.map(_.collect { case _: BloomFilterMightContain => 1
}.sum).sum
+ val skipped = conditions.map(_.collect {
+ case If(a: Attribute, Literal.TrueLiteral, _: BloomFilterMightContain)
+ if a.name == checkedName => 1
+ }.sum).sum
+ (all - skipped, skipped)
+ }
+
+ // The rows the Filter on the scan returned, counted before the join.
+ private def rowsOutOfFilter(plan: SparkPlan): Long = {
+ val filters = collect(plan) {
+ case f: FilterExec if f.exists(_.isInstanceOf[BatchScanExec]) => f
+ }
+ assert(filters.size == 1, s"expected one Filter on the scan:\n$plan")
+ filters.head.metrics("numOutputRows").value
+ }
+
+ // Runs the bloom join over the v2 source read with `options`, and returns
its rows and plan.
+ private def runBloomJoin(options: (String, String)*): (Set[(Long, Long)],
SparkPlan) = {
+ StorageFilterTestSource.reset()
+ val df = bloomJoin(v2Probe(options: _*))
+ (df.collect().map(r => (r.getLong(0), r.getLong(1))).toSet,
df.queryExecution.executedPlan)
+ }
+
+ // The build side keeps `bv = 5`, so the join returns that one row.
+ private val joinResult = Set((5L, 50L))
+
+ for (columnar <- Seq(false, true); aqe <- Seq(false, true)) {
+ val kind = if (columnar) "columnar" else "row-based"
+ test(s"a $kind scan gets the checked column, and the Filter skips the
bloom on checked rows " +
+ s"(AQE $aqe)") {
+ withBloomJoin(SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> aqe.toString) {
+ val (result, plan) = runBloomJoin("columnar" -> columnar.toString)
+ val scan = scanOf(plan)
+ assert(scan.supportsColumnar == columnar, s"the scan must read
$kind:\n$plan")
+ assert(scan.storageFilters.size == 1,
+ s"the bloom filter must be on the scan; got
${scan.storageFilters}\n$plan")
+ assert(scan.checkedColumn.isDefined && scan.checkedColumn ==
scan.output.lastOption,
+ s"the scan must get the checked column, last:\n$plan")
+ assert(bloomsAbove(plan) == (0, 1),
+ s"the Filter must skip the bloom on checked rows:\n$plan")
+ assert(rowsOutOfFilter(plan) ==
StorageFilterTestSource.rowsReturned.get(),
+ s"the Filter must keep every checked row; got
${rowsOutOfFilter(plan)} of " +
+ s"${StorageFilterTestSource.rowsReturned.get()}")
+ val probeSide = collectFirst(plan) { case j: SortMergeJoinExec =>
j.left.output }
+ assert(probeSide.exists(!_.exists(_.name == checkedName)),
+ s"the Project above the scan must drop the checked column; got
$probeSide")
+ assert(StorageFilterTestSource.filteredReaders.get() == 1,
+ s"the one partition's reader must get the filter; " +
+ s"got ${StorageFilterTestSource.filteredReaders.get()}")
+ assert(StorageFilterTestSource.checkedColumnRequests.get() == 1,
+ "the reader must be asked for the checked column")
+ assert(result == joinResult, s"got $result")
+ assert(StorageFilterTestSource.rowsEvaluated.get() ==
StorageFilterTestSource.numRows,
+ s"the reader must evaluate every row; got
${StorageFilterTestSource.rowsEvaluated.get()}")
+ assert(StorageFilterTestSource.rowsReturned.get() <
StorageFilterTestSource.numRows,
+ "the reader must have dropped rows, otherwise the filter did
nothing")
+ }
+ }
+ }
+
+ Seq(false, true).foreach { columnar =>
+ val kind = if (columnar) "columnar" else "row-based"
+ test(s"a $kind reader that gives the filter up marks its rows unchecked,
and the Filter " +
+ "filters them") {
+ withBloomJoin() {
+ val (result, plan) = runBloomJoin("columnar" -> columnar.toString,
"giveUp" -> "true")
+ assert(bloomsAbove(plan) == (0, 1), s"the plan is that of a filtering
reader:\n$plan")
+ assert(StorageFilterTestSource.rowsEvaluated.get() == 0,
+ s"the reader must evaluate nothing; got
${StorageFilterTestSource.rowsEvaluated.get()}")
+ assert(StorageFilterTestSource.rowsReturned.get() ==
StorageFilterTestSource.numRows,
+ s"the reader must return every row; got
${StorageFilterTestSource.rowsReturned.get()}")
+ assert(rowsOutOfFilter(plan) < StorageFilterTestSource.numRows,
+ s"the Filter must apply the bloom to unchecked rows; got
${rowsOutOfFilter(plan)}")
+ assert(result == joinResult, s"got $result")
+ }
+ }
+
+ test(s"the Filter takes a $kind reader's word on the rows it marks
checked") {
+ // This reader breaks the contract on purpose: it applies no filter but
marks every row
+ // checked. Only skipping the bloom on those rows keeps all of them.
+ withBloomJoin() {
+ val (result, plan) =
+ runBloomJoin("columnar" -> columnar.toString, "markAllChecked" ->
"true")
+ assert(rowsOutOfFilter(plan) == StorageFilterTestSource.numRows,
+ s"the Filter must keep every row marked checked; got
${rowsOutOfFilter(plan)}")
+ assert(result == joinResult, s"the join must still filter; got
$result")
+ }
+ }
+ }
+
+ test("a filter that cannot be applied is given up, and its rows come back
unchecked") {
+ // `1 / k` under ANSI throws at k = 0, the partition's first row, so the
reader gives the
+ // filter up before it keeps or drops anything.
+ val k = AttributeReference("k", LongType)()
+ val inverse = Divide(Literal(1.0), Cast(k, DoubleType),
NumericEvalContext(EvalMode.ANSI))
+ val filter = StorageFilterImpl(GreaterThan(inverse, Literal(0.0)), Seq(k))
+ val reader = new StorageFilterTestColumnarReader(Some(filter),
appendsChecked = true)
+ val rows = mutable.ArrayBuffer.empty[(Long, Boolean)]
+ try {
+ while (reader.next()) {
+ val batch = reader.get()
+ (0 until batch.numRows()).foreach { i =>
+ rows += ((batch.column(0).getLong(i), batch.column(2).getBoolean(i)))
+ }
+ }
+ } finally {
+ reader.close()
+ }
+ assert(rows == (0L until StorageFilterTestSource.numRows.toLong).map(_ ->
false),
+ s"expected every row, unchecked; got ${rows.take(5)} and ${rows.size}
rows in all")
+ }
+
+ // Under a limit of two fields, the checked column would make three.
+ Seq(
+ ("too wide for whole-stage codegen", SQLConf.WHOLESTAGE_MAX_NUM_FIELDS.key
-> "2"),
+ ("without whole-stage codegen", SQLConf.WHOLESTAGE_CODEGEN_ENABLED.key ->
"false")
+ ).foreach { case (when, conf) =>
+ test(s"a scan $when gets the bloom without the checked column") {
+ withBloomJoin(conf) {
+ val (result, plan) = runBloomJoin("columnar" -> "true")
+ val scan = scanOf(plan)
+ assert(scan.storageFilters.size == 1,
+ s"the bloom filter must be on the scan; got
${scan.storageFilters}\n$plan")
+ assert(scan.checkedColumn.isEmpty, s"the scan must not get the checked
column:\n$plan")
+ assert(StorageFilterTestSource.checkedColumnRequests.get() == 0,
+ "the reader must not be asked for the checked column")
+ assert(bloomsAbove(plan) == (1, 0), s"the Filter must keep the bloom
as it is:\n$plan")
+ assert(result == joinResult, s"got $result")
+ assert(StorageFilterTestSource.rowsReturned.get() <
StorageFilterTestSource.numRows,
+ "the reader must have dropped rows, otherwise the filter did
nothing")
+ }
+ }
+ }
+
+ Seq(
+ ("with the conf off", Seq(SQLConf.V2_STORAGE_FILTER_PUSHDOWN_ENABLED.key
-> "false"), Nil),
+ ("to a scan that declares no column", Nil, Seq("declared" -> "")),
+ ("to a scan that declares only another column", Nil, Seq("declared" ->
"v")),
+ ("to a scan that declares only a nested field", Nil, Seq("declared" ->
"k.x")),
+ ("to a scan with a column named like the checked column", Nil,
+ Seq("valueName" -> checkedName.toUpperCase(java.util.Locale.ROOT)))
+ ).foreach { case (when, conf, options) =>
+ test(s"nothing is offered $when") {
+ withBloomJoin(conf: _*) {
+ val (result, plan) = runBloomJoin(options :+ ("columnar" -> "true"):
_*)
+ val scan = scanOf(plan)
+ assert(scan.storageFilters.isEmpty && scan.checkedColumn.isEmpty,
+ s"the scan must get no storage filter:\n$plan")
+ assert(bloomsAbove(plan) == (1, 0), s"the Filter must keep the bloom
as it is:\n$plan")
+ assert(StorageFilterTestSource.filteredReaders.get() == 0,
+ "no reader may get a storage filter")
+ assert(result == joinResult, s"got $result")
+ }
+ }
+ }
+
+ test("the storage filter shows in the scan description and separates
canonicalized plans") {
+ withBloomJoin() {
+ val plan = bloomJoin(v2Probe("columnar" ->
"true")).queryExecution.executedPlan
+ val scan = scanOf(plan)
+ assert(scan.simpleString(100).contains("StorageFilters:"),
+ s"the scan description must name the filters:
${scan.simpleString(100)}")
+ assert(scan.verboseStringWithOperatorId().contains("StorageFilters:"),
+ s"EXPLAIN FORMATTED must name the filters:
${scan.verboseStringWithOperatorId()}")
+
+ intercept[IllegalArgumentException](scan.copy(storageFilters = Nil))
+ val plain = scan.copy(output = scan.output.init, storageFilters = Nil,
checkedColumn = None)
+ assert(scan != plain, "equals must see storageFilters")
Review Comment:
**13. [Low] These asserts do not show that `equals` and `doCanonicalize` see
`storageFilters`**
`plain` also drops `checkedColumn`, and `equals` compares
`checkedColumn.isDefined` on its own. So both asserts hold even with
`storageFilters` removed from `equals` or `doCanonicalize`. The case the
message names, a scan with storage filters and no checked column (too wide, or
codegen off), is never compared, and `hashCode` is not checked.
Other tests would still catch such a removal today. Still, could this
compare two scans that differ only in `storageFilters`, both with
`checkedColumn = None` and the same output, and also assert on `hashCode`?
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Strategy.scala:
##########
@@ -127,6 +127,29 @@ class DataSourceV2Strategy(session: SparkSession) extends
Strategy with Predicat
tableSpec.withNewLocation(newLoc)
}
+ /**
+ * The conjuncts of `filters` offered to the scan as storage filters, see
+ * `SupportsStorageFilters` and `StorageFilterPlanning.offeredConjuncts`.
Only runtime blooms for
+ * now, as on the file-source side.
+ */
+ private def storageFiltersFor(
+ relation: DataSourceV2ScanRelation,
+ filters: Seq[Expression]): Seq[Expression] = {
+ if (!session.sessionState.conf.v2StorageFilterPushdownEnabled) return Nil
+ if (relation.storageFilterAttrs.isEmpty) return Nil
+ // A scan that took a pushed limit, offset, top-N or sample, or an
aggregate or join, would have
+ // its readers filter ahead of it. `V2ScanRelationPushDown` leaves every
such scan, and any scan
+ // a later pushdown builds, not mergeable.
+ if (!relation.mergeableScan) return Nil
Review Comment:
**18. [Low] Gating on `mergeableScan` couples storage filters to scan
merging**
`mergeableScan` means "a rebuilt scan can reproduce this scan's pushdowns"
(`hasBlockingPushdown`, read by `PlanMerger`). This check uses it as "no
pushdown decides which rows the scan returns". The two sets agree today, but
not by construction:
- A scan with pushed variant extraction keeps its rows. Yet
`buildScanWithPushedVariants` leaves it not mergeable
(V2ScanRelationPushDown.scala:959), so `SELECT f.k, variant_get(f.v, '$.a',
'int') FROM fact f JOIN dim d ON f.k = d.k WHERE d.x = 5` gets no storage
filter.
- `hasBlockingPushdown`'s maintenance rule sorts a pushdown by whether a
merge can reproduce it. Suppose a later change makes a row-selecting pushdown
reproducible, e.g. an identical pushed top-N or a seeded sample. The scan then
becomes mergeable and gets the bloom ahead of that pushdown, which then picks
other rows. Only doc sentences and the four limit / offset / top-N / sample
tests guard against that.
A dedicated flag on `DataSourceV2ScanRelation` would decouple the two. It
would default to false, and the same build sites would set it from the
pushdown's row semantics.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/BatchScanExec.scala:
##########
@@ -29,22 +29,37 @@ import org.apache.spark.sql.connector.catalog.Table
import org.apache.spark.sql.connector.read._
import org.apache.spark.sql.connector.write.RowLevelOperation.Command.DELETE
import org.apache.spark.sql.connector.write.RowLevelOperationTable
-import org.apache.spark.sql.execution.EmptyRDDWithPartitions
+import org.apache.spark.sql.execution.{EmptyRDDWithPartitions, ScalarSubquery
=> ExecScalarSubquery}
import org.apache.spark.sql.execution.metric.{SQLLastAttemptMetrics,
SQLMetric, SQLMetrics}
+import org.apache.spark.sql.internal.connector.{StorageFilterImpl,
StorageFilterRequestImpl}
import org.apache.spark.util.ArrayImplicits._
/**
* Physical plan node for scanning a batch of data from a data source v2.
+ *
+ * @param storageFilters Predicates the scan's readers may evaluate, to skip
reading the other
+ * columns of the rows they reject, see
[[SupportsStorageFilters]]. Non-empty
+ * only for a scan that declared their columns. Each
task's readers get them
+ * ANDed into one filter, in a [[StorageFilterRequest]].
They stay in the
+ * post-scan `Filter` too.
+ * @param checkedColumn The column the readers mark the rows they checked in,
last in `output`, when
+ * the planner asked for one, see
`StorageFilterPlanning.checkedColumn`.
*/
case class BatchScanExec(
output: Seq[AttributeReference],
@transient scan: Scan,
runtimeFilters: Seq[Expression],
ordering: Option[Seq[SortOrder]] = None,
@transient table: Table,
- keyGroupedPartitioning: Option[Seq[Expression]] = None
+ keyGroupedPartitioning: Option[Seq[Expression]] = None,
+ storageFilters: Seq[Expression] = Nil,
+ checkedColumn: Option[AttributeReference] = None
Review Comment:
**19. [Nit] `checkedColumn` duplicates `output.last`**
Main code reads only `checkedColumn.isDefined` (L86, L92, L125), and the
strategy uses its local `checked` for `skipCheckedRows`. Three things exist
only to keep the two copies equal: the `require` (L60), the extra normalization
in `doCanonicalize` (L151) and the test twin's remapping.
`appendsCheckedColumn: Boolean = false` with `require(!appendsCheckedColumn ||
storageFilters.nonEmpty)` would carry the same information, with fewer places
to keep in sync. If wanted, it could also check that the last output column is
a `BooleanType`.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/connector/StorageFilterImpl.scala:
##########
@@ -0,0 +1,132 @@
+/*
+ * 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 scala.util.control.NonFatal
+
+import org.apache.spark.{SparkEnv, SparkException, TaskContext}
+import org.apache.spark.executor.Executor
+import org.apache.spark.internal.{Logging, LogKeys}
+import org.apache.spark.internal.config.KILL_ON_FATAL_ERROR_DEPTH
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions.{Attribute, BindReferences,
Expression => CatalystExpression, Predicate => CatalystPredicate}
+import org.apache.spark.sql.connector.expressions.{FieldReference,
NamedReference}
+import org.apache.spark.sql.connector.expressions.filter.{StorageFilter,
StorageFilterNotAppliedException}
+
+/**
+ * A [[StorageFilter]] wrapping a Catalyst expression, so a scan can apply a
predicate it does not
+ * have to understand. It mirrors [[PartitionPredicateImpl]], which evaluates
a partition key
+ * rather than a data row.
+ *
+ * `attrs` is the layout of the row [[eval]] expects, in that order, and the
expression is bound to
+ * it. The binding is built on first use, so an instance created on the driver
can be evaluated on
+ * an executor.
+ */
+class StorageFilterImpl private (
+ private val catalystExpr: CatalystExpression,
+ private val attrs: Seq[Attribute])
+ extends StorageFilter with Logging {
+
+ // The width of the row `eval` takes.
+ private val numReferences = attrs.length
Review Comment:
**20. [Nit] Small cleanups in `StorageFilterImpl`**
The column list is rendered three ways (L68, L79 and L99), and the comment
"a runtime bloom's literal would print every byte of the bloom" appears twice
(L78 and L98). `toString` already is the safe rendering, so the internal error
and the warning could interpolate `$this`, and the comment could stay once, on
`toString`. Also, `numReferences` is just `attrs.length`.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/BatchScanExec.scala:
##########
@@ -98,13 +116,22 @@ case class BatchScanExec(
override lazy val readerFactory: PartitionReaderFactory =
batch.createReaderFactory()
+ // What every task's readers get, the storage filters ANDed into one. A
runtime bloom arrives as
+ // a ScalarSubquery over the build side, which SparkPlan has computed by the
time this is forced.
+ // The readers get its literal, so the task does not carry the subquery's
plan.
+ @transient private lazy val storageFilterRequest:
Option[StorageFilterRequest] =
+ storageFilters.map(_.transform { case s: ExecScalarSubquery => s.toLiteral
})
Review Comment:
**6. [Medium] A merged bloom ships the whole merged struct once per
conjunct**
Only the `ScalarSubquery` node is replaced here. Take a join on two declared
keys:
1. `InjectRuntimeFilter` injects one bloom per key
(InjectRuntimeFilter.scala:418).
2. `MergeSubplans` runs in the next batch whenever subquery reuse is on,
which is the default. It merges the two into
`might_contain(GetStructField(ScalarSubquery(merged), i), xxhash64(k_i))`, and
both conjuncts still pass `_.isInstanceOf[BloomFilterMightContain]`.
3. Each `ScalarSubquery`'s `toLiteral` (subquery.scala:112-114) is then a
separate `UnsafeRow` view of the whole `struct<bloom_1, bloom_2>`, and Java
serialization writes each one in full.
So the request carries `2 x (b1 + b2)` bytes instead of `b1 + b2`. That is
up to 32 MiB at the default
`spark.sql.optimizer.runtime.bloomFilter.maxNumBits`. With `n` merged blooms,
every task gets `n` times their total. The post-scan `Filter` ships only the
deserialized bloom it uses.
Could the substitution also fold each bloom argument to its own binary
literal on the driver, e.g. `Literal(bloomFilterExpression.eval(), BinaryType)`
once the subquery is replaced? A test with a two-key join would cover this
shape too (item 12).
##########
sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/filter/StorageFilter.java:
##########
@@ -0,0 +1,111 @@
+/*
+ * 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.expressions.filter;
+
+import org.apache.spark.annotation.Evolving;
+import org.apache.spark.sql.catalyst.InternalRow;
+import org.apache.spark.sql.connector.expressions.NamedReference;
+import org.apache.spark.sql.connector.read.Scan;
+import org.apache.spark.sql.connector.read.StorageFilterRequest;
+import org.apache.spark.sql.connector.read.SupportsStorageFilters;
+import org.apache.spark.sql.vectorized.ColumnarBatch;
+
+/**
+ * A row-level predicate a scan may evaluate inside its readers, to skip
reading the other columns
+ * of the rows it rejects. The reader reads the columns of {@link
#references()} first, evaluates
+ * the filter, and reads the remaining columns only for the rows it keeps.
+ * <p>
+ * The filter is a predicate on the rows the scan returns. Spark offers none
to a scan that took a
+ * pushed operation that decides which rows it returns, such as a limit, an
offset, a top-N, a
+ * sample or an aggregate.
+ * <p>
+ * Spark implements the predicate, and the reader only supplies the values.
For example, a scan
+ * does not have to reproduce the Spark sketch and hash of a runtime bloom
filter. The values must
+ * be the ones the scan returns for those columns, in the types of {@link
Scan#readSchema()}, nulls
+ * included.
+ * <p>
+ * Applying it is optional, see {@link SupportsStorageFilters}.
+ * <p>
+ * A row's result depends only on its own values. So with one reference, a
reader may evaluate each
+ * distinct value once, such as each entry of a dictionary, and reuse the
result for every row
+ * holding that value. A dictionary entry is evaluated as the value the scan
returns for it, decoded
+ * and converted.
+ * <p>
+ * Spark hands the filter to each task in a {@link StorageFilterRequest}. The
readers of a task
+ * share one instance, which is not thread-safe.
+ *
+ * @since 5.0.0
+ */
+@Evolving
+public abstract class StorageFilter extends Predicate {
Review Comment:
**4. [Nit] Does `StorageFilter` need to be a `Predicate`?**
A connector gets a `StorageFilter` only through
`StorageFilterRequest.filter()`, never in a `Predicate[]`. `PartitionPredicate`
is different: it travels through `SupportsRuntimeV2Filtering.filter` and
`pushPredicates`. Extending `Predicate` brings two things here:
- `name() == "STORAGE_FILTER"` with no children, which a generic V2
predicate translator cannot handle;
- `GeneralScalarExpression`'s name-and-children equality, under which any
two filters are equal. That is why `StorageFilterImpl` overrides `equals` and
`hashCode`, which nothing calls.
A standalone `public abstract class StorageFilter implements Serializable`
with `references()`, `eval` and `select` would avoid both. This is new
`@Evolving` API, and removing a superclass later breaks binary compatibility,
so it may be worth settling now.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Strategy.scala:
##########
@@ -200,10 +223,20 @@ class DataSourceV2Strategy(session: SparkSession) extends
Strategy with Predicat
// rewritten it to TrueLiteral by the time we get here.
val runtimeFilters = dynamicFilters ++ scalarSubqueryFilters
- val batchExec = BatchScanExec(relation.output, relation.scan,
runtimeFilters,
- relation.ordering, relation.relation.table,
relation.keyGroupedPartitioning)
- DataSourceV2Strategy.withProjectAndFilter(
- project, postScanFilters.diff(fullyPushedRuntimeFilters),
+ // Whether the scan gets the checked column does not depend on whether
it is columnar. Asking
+ // that can plan the scan's partitions, which the node built afterwards
would plan again. A
+ // row-based scan gets a Project for the UnsafeRow conversion anyway, so
dropping the column
+ // there copies nothing more.
+ val filtersAbove = postScanFilters.diff(fullyPushedRuntimeFilters)
+ val storageFilters = storageFiltersFor(relation, filtersAbove)
+ val checked = StorageFilterPlanning.checkedColumn(
Review Comment:
**8. [Low] A row-based scan loses the checked column where it would cost
nothing**
The gate in `StorageFilterPlanning.checkedColumn` is meant for columnar
scans: whole-stage codegen on, and the output plus the column within
`spark.sql.codegen.maxFields`. As the comment above says, a row-based scan
always gets the `ProjectExec` for the UnsafeRow conversion
(`needsUnsafeConversion`, kept by RemoveRedundantProjects.scala:108).
Take a row-based scan whose reader drops rejected rows, with
`spark.sql.codegen.wholeStage=false` or with an output already over
`maxFields`. It gets no column, so the post-scan `Filter` evaluates `xxhash64`
and the bloom probes again on every row the reader kept. The same Project would
have dropped the column at no extra cost. At exactly `maxFields` the column
does push the Filter and the Project out of codegen, so the gate is right there.
`scan.columnarSupportMode()` answers without planning partitions for
`SUPPORTED` and `UNSUPPORTED`. Could those two decide by read mode, and
`PARTITION_DEFINED` keep today's gate? The doc of
`StorageFilterRequest.appendsCheckedColumn` would then need to say so.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Strategy.scala:
##########
@@ -127,6 +127,29 @@ class DataSourceV2Strategy(session: SparkSession) extends
Strategy with Predicat
tableSpec.withNewLocation(newLoc)
}
+ /**
+ * The conjuncts of `filters` offered to the scan as storage filters, see
+ * `SupportsStorageFilters` and `StorageFilterPlanning.offeredConjuncts`.
Only runtime blooms for
+ * now, as on the file-source side.
+ */
+ private def storageFiltersFor(
+ relation: DataSourceV2ScanRelation,
+ filters: Seq[Expression]): Seq[Expression] = {
+ if (!session.sessionState.conf.v2StorageFilterPushdownEnabled) return Nil
+ if (relation.storageFilterAttrs.isEmpty) return Nil
+ // A scan that took a pushed limit, offset, top-N or sample, or an
aggregate or join, would have
+ // its readers filter ahead of it. `V2ScanRelationPushDown` leaves every
such scan, and any scan
+ // a later pushdown builds, not mergeable.
+ if (!relation.mergeableScan) return Nil
+ // Over the whole table, as on the file-source side, since
`SupportsStorageFilters` reserves
+ // the name.
+ if
(StorageFilterPlanning.clashesWithCheckedColumn(relation.relation.output.map(_.name)))
{
+ return Nil
+ }
+ StorageFilterPlanning.offeredConjuncts(filters,
relation.storageFilterAttrs,
+ AttributeSet(relation.output), _.isInstanceOf[BloomFilterMightContain])
Review Comment:
**11. [Nit] Metadata columns count as columns left to skip**
`readColumns` is the whole `relation.output`, metadata columns included,
while the file-source side passes only the data columns (`readDataColumns`).
Take a scan that reads only the key plus, say, `_file` or `_pos`. It passes the
"something left to skip" rule, and pays for the checked column, the request and
the extra planning with no IO saved. V1 would offer nothing there.
Whether that is a waste depends on the connector's metadata columns. So
either of these would do: leave the relation's metadata attributes out, matched
by exprId against `relation.relation.output`, or add a sentence to the
`offeredConjuncts` doc that the rule is an approximation.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/connector/StorageFilterImpl.scala:
##########
@@ -0,0 +1,132 @@
+/*
+ * 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 scala.util.control.NonFatal
+
+import org.apache.spark.{SparkEnv, SparkException, TaskContext}
+import org.apache.spark.executor.Executor
+import org.apache.spark.internal.{Logging, LogKeys}
+import org.apache.spark.internal.config.KILL_ON_FATAL_ERROR_DEPTH
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions.{Attribute, BindReferences,
Expression => CatalystExpression, Predicate => CatalystPredicate}
+import org.apache.spark.sql.connector.expressions.{FieldReference,
NamedReference}
+import org.apache.spark.sql.connector.expressions.filter.{StorageFilter,
StorageFilterNotAppliedException}
+
+/**
+ * A [[StorageFilter]] wrapping a Catalyst expression, so a scan can apply a
predicate it does not
+ * have to understand. It mirrors [[PartitionPredicateImpl]], which evaluates
a partition key
+ * rather than a data row.
+ *
+ * `attrs` is the layout of the row [[eval]] expects, in that order, and the
expression is bound to
+ * it. The binding is built on first use, so an instance created on the driver
can be evaluated on
+ * an executor.
+ */
+class StorageFilterImpl private (
+ private val catalystExpr: CatalystExpression,
+ private val attrs: Seq[Attribute])
+ extends StorageFilter with Logging {
+
+ // The width of the row `eval` takes.
+ private val numReferences = attrs.length
+
+ // Whether this instance has logged a failure, so a task logs only its first.
+ @transient private var loggedFailure = false
+
+ @transient private lazy val boundPredicate: InternalRow => Boolean = {
+ val bound = BindReferences.bindReference(catalystExpr, attrs)
+ CatalystPredicate.create(bound).eval
Review Comment:
**9. [Low] The bloom's literal stays alive for the whole task**
Once `boundPredicate` is built, the generated predicate holds only the
deserialized bloom. But `catalystExpr` keeps the bloom's binary literal
reachable from the task's `DataSourceRDD` until the task ends. At the default 8
MiB bloom, that is another 8 MiB per running task, about 128 MiB on a 16-slot
executor, and more with item 6.
The PR description lists this under possible later work. V1 already handles
it in a few lines: `ParquetStorageFilter.preparedPredicate` drops its
expression once `Predicate.create` returns
(ParquetStorageFilter.scala:120-127). Only `equals` and `hashCode` read
`catalystExpr` afterwards, so they could use a hash computed up front, or the
fields that remain.
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/StorageFilterTestSource.scala:
##########
@@ -0,0 +1,303 @@
+/*
+ * 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.util
+import java.util.OptionalLong
+import java.util.concurrent.atomic.{AtomicInteger, AtomicLong}
+
+import scala.jdk.CollectionConverters._
+
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions.GenericInternalRow
+import org.apache.spark.sql.connector.catalog.{SupportsRead, Table,
TableCapability, TableProvider}
+import org.apache.spark.sql.connector.expressions.{FieldReference,
NamedReference, SortOrder, Transform}
+import org.apache.spark.sql.connector.expressions.filter.{StorageFilter,
StorageFilterNotAppliedException}
+import org.apache.spark.sql.connector.read._
+import org.apache.spark.sql.execution.vectorized.OnHeapColumnVector
+import org.apache.spark.sql.types.{BooleanType, LongType, StructType}
+import org.apache.spark.sql.util.CaseInsensitiveStringMap
+import org.apache.spark.sql.vectorized.{ColumnarBatch, ColumnVector}
+
+/**
+ * A read-only v2 source over a fixed table of `(k, v)` longs that implements
+ * [[SupportsStorageFilters]]. Its reader gets the filter in the task and
drops the rows the filter
+ * rejects. Its options:
+ * - `columnar`: reads in batches, through `StorageFilter.select`, rather
than row by row;
+ * - `giveUp`: reads without the filter, as a reader does for a unit it
cannot use the filter on.
+ * Here the unit is the one partition;
+ * - `markAllChecked`: reads without the filter but marks every row checked,
which breaks the
+ * contract on purpose;
+ * - `declared`: the comma-separated storage-filter columns, `k` by default,
a dot separating a
+ * nested field;
+ * - `valueName`: the name of the value column, `v` by default;
+ * - `push`: the comma-separated pushed operations it takes whole without
applying them, of
+ * `limit`, `offset`, `topN` and `sample`, for planning tests only.
+ *
+ * A filter that throws `StorageFilterNotAppliedException` is given up for the
rest of the
+ * partition too. The counters below record what the readers got, evaluated
and returned.
+ */
+class StorageFilterTestSource extends TableProvider {
+
+ override def inferSchema(options: CaseInsensitiveStringMap): StructType =
+ new StructType().add("k", LongType).add(options.getOrDefault("valueName",
"v"), LongType)
+
+ override def getTable(
+ schema: StructType,
+ partitioning: Array[Transform],
+ properties: util.Map[String, String]): Table = new
StorageFilterTestTable(schema)
+}
+
+object StorageFilterTestSource {
+
+ /** Rows the source serves, `(k, k * 10)` for k in 0 until numRows. */
+ val numRows: Int = 200
+
+ /** Rows a batch holds at most in a columnar read. */
+ val batchSize: Int = 64
+
+ /** How many readers were created with a storage-filter request. */
+ val filteredReaders = new AtomicInteger(0)
+
+ /** How many of those requests asked for the checked column. */
+ val checkedColumnRequests = new AtomicInteger(0)
+
+ /** Rows the reader evaluated the storage filter against. */
+ val rowsEvaluated = new AtomicLong(0)
+
+ /** Rows the reader returned. */
+ val rowsReturned = new AtomicLong(0)
+
+ def reset(): Unit = {
+ filteredReaders.set(0)
+ checkedColumnRequests.set(0)
+ rowsEvaluated.set(0)
+ rowsReturned.set(0)
+ }
+}
+
+class StorageFilterTestTable(tableSchema: StructType) extends Table with
SupportsRead {
+
+ override def name(): String = "storage_filter_test"
+
+ override def schema(): StructType = tableSchema
+
+ override def capabilities(): util.Set[TableCapability] =
+ Set(TableCapability.BATCH_READ).asJava
+
+ override def newScanBuilder(options: CaseInsensitiveStringMap): ScanBuilder
= {
+ val factory = StorageFilterTestReaderFactory(
+ columnar = options.getBoolean("columnar", false),
+ giveUp = options.getBoolean("giveUp", false),
+ marksAllChecked = options.getBoolean("markAllChecked", false))
+ val declared = options.getOrDefault("declared",
"k").split(",").filter(_.nonEmpty).toSeq
+ val scan = () => new StorageFilterTestScan(tableSchema, factory, declared)
+ val pushes = options.getOrDefault("push",
"").split(",").filter(_.nonEmpty).toSet
+ if (pushes.nonEmpty) new StorageFilterTestPushingScanBuilder(scan, pushes)
else () => scan()
+ }
+}
+
+/**
+ * Takes the pushed operations named in `pushes` whole. Its scan does not
apply them, so it serves
+ * planning tests only.
+ */
+class StorageFilterTestPushingScanBuilder(scan: () => Scan, pushes:
Set[String])
+ extends SupportsPushDownLimit
+ with SupportsPushDownOffset
+ with SupportsPushDownTopN
+ with SupportsPushDownTableSample {
+
+ override def build(): Scan = scan()
+ override def pushLimit(limit: Int): Boolean = pushes("limit")
+ override def pushOffset(offset: Int): Boolean = pushes("offset")
+ override def pushTopN(orders: Array[SortOrder], limit: Int): Boolean =
pushes("topN")
+ override def pushTableSample(
+ lowerBound: Double,
+ upperBound: Double,
+ withReplacement: Boolean,
+ seed: Long,
+ sampleMethod: SampleMethod): Boolean = pushes("sample")
+ override def isPartiallyPushed: Boolean = false
+}
+
+class StorageFilterTestScan(
+ tableSchema: StructType,
+ factory: StorageFilterTestReaderFactory,
+ declared: Seq[String])
+ extends Scan with Batch with SupportsReportStatistics with
SupportsStorageFilters {
+
+ override def readSchema(): StructType = tableSchema
+
+ override def toBatch: Batch = this
+
+ // Over the suite's application-side scan size threshold, so
InjectRuntimeFilter considers this
+ // side worth filtering.
+ override def estimateStatistics(): Statistics = new Statistics {
+ override def sizeInBytes(): OptionalLong = OptionalLong.of(10L * 1024 *
1024)
+ override def numRows(): OptionalLong =
OptionalLong.of(StorageFilterTestSource.numRows.toLong)
+ }
+
+ override def storageFilterAttributes(): Array[NamedReference] =
+ declared.map(name =>
FieldReference(name.split('.').toSeq)).toArray[NamedReference]
+
+ override def planInputPartitions(): Array[InputPartition] =
+ Array(StorageFilterTestPartition)
+
+ override def createReaderFactory(): PartitionReaderFactory = factory
+}
+
+case object StorageFilterTestPartition extends InputPartition
+
+case class StorageFilterTestReaderFactory(
+ columnar: Boolean,
+ giveUp: Boolean,
+ marksAllChecked: Boolean)
+ extends PartitionReaderFactory {
+
+ override def supportColumnarReads(partition: InputPartition): Boolean =
columnar
+
+ // The filter the reader applies, none when it gives the filter up or only
marks the rows.
+ private def filterOf(request: StorageFilterRequest): Option[StorageFilter] =
{
+ StorageFilterTestSource.filteredReaders.incrementAndGet()
+ if (request.appendsCheckedColumn()) {
+ StorageFilterTestSource.checkedColumnRequests.incrementAndGet()
+ }
+ if (giveUp || marksAllChecked) None else Some(request.filter())
+ }
+
+ override def createReader(partition: InputPartition):
PartitionReader[InternalRow] =
+ new StorageFilterTestReader(None, appendsChecked = false)
+
+ override def createReader(
+ partition: InputPartition,
+ request: StorageFilterRequest): PartitionReader[InternalRow] =
+ new StorageFilterTestReader(filterOf(request),
request.appendsCheckedColumn(), marksAllChecked)
+
+ override def createColumnarReader(partition: InputPartition):
PartitionReader[ColumnarBatch] =
+ new StorageFilterTestColumnarReader(None, appendsChecked = false)
+
+ override def createColumnarReader(
+ partition: InputPartition,
+ request: StorageFilterRequest): PartitionReader[ColumnarBatch] =
+ new StorageFilterTestColumnarReader(
+ filterOf(request), request.appendsCheckedColumn(), marksAllChecked)
+}
+
+/**
+ * Serves `(k, k * 10)` row by row and drops any row `filter` rejects, with
the checked column
+ * last when `appendsChecked`. The filter's row layout is its `references()`,
only `k` here.
+ * `filter` is cleared when it cannot be applied. `marksAllChecked` marks
every row checked.
+ */
+class StorageFilterTestReader(
+ private var filter: Option[StorageFilter],
+ appendsChecked: Boolean,
+ marksAllChecked: Boolean = false)
+ extends PartitionReader[InternalRow] {
+
+ private var current: Long = -1L
+ private var row: InternalRow = _
+
+ override def next(): Boolean = {
+ var found = false
+ while (!found && current + 1 < StorageFilterTestSource.numRows) {
+ current += 1
+ val key = current
+ val passes = filter.forall { f =>
+ StorageFilterTestSource.rowsEvaluated.incrementAndGet()
+ try f.eval(new GenericInternalRow(Array[Any](key))) catch {
Review Comment:
**12. [Medium] No test covers more than one bloom, more than one reader, or
runtime filtering**
Every planner test plans one bloom over `k` on a one-partition scan
(`storageFilters.size == 1`, `filteredReaders == 1`). This reader ignores
`references()` and builds a one-value key row, and the table has only `(k, v)`.
So these paths have no end-to-end test:
- Several offered conjuncts: the `reduceOption(And)` request, its layout
ordered by output position, and one `If(checked, true, _)` per conjunct in the
Filter. For example, keeping only the first storage filter would pass every
test, while the Filter would skip the second bloom on rows never evaluated
against it.
- The `MergeSubplans` form `GetStructField(ScalarSubquery(merged), i)` (item
6).
- A scan that also implements `SupportsRuntimeV2Filtering` or
`SupportsRuntimeCatalystFiltering` on the key, where a fully pushed bloom must
not be offered (`filtersAbove`).
- Several readers of one task sharing one `StorageFilterImpl`, which the
`StorageFilter` doc promises. This happens with coalesced partitions or the SPJ
k-way merge.
Most of this would be covered by three tests: a two-key join with a
probe-side conjunct that is not offered, a runtime-filtering variant of the
source, and a task that reads several partitions. The fixture would need a
third column, and would need to read the layout from `references()`.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Strategy.scala:
##########
@@ -127,6 +127,29 @@ class DataSourceV2Strategy(session: SparkSession) extends
Strategy with Predicat
tableSpec.withNewLocation(newLoc)
}
+ /**
+ * The conjuncts of `filters` offered to the scan as storage filters, see
+ * `SupportsStorageFilters` and `StorageFilterPlanning.offeredConjuncts`.
Only runtime blooms for
+ * now, as on the file-source side.
+ */
+ private def storageFiltersFor(
+ relation: DataSourceV2ScanRelation,
+ filters: Seq[Expression]): Seq[Expression] = {
+ if (!session.sessionState.conf.v2StorageFilterPushdownEnabled) return Nil
+ if (relation.storageFilterAttrs.isEmpty) return Nil
+ // A scan that took a pushed limit, offset, top-N or sample, or an
aggregate or join, would have
+ // its readers filter ahead of it. `V2ScanRelationPushDown` leaves every
such scan, and any scan
+ // a later pushdown builds, not mergeable.
+ if (!relation.mergeableScan) return Nil
+ // Over the whole table, as on the file-source side, since
`SupportsStorageFilters` reserves
Review Comment:
**16. [Low] The stated reason for this check is circular**
"since `SupportsStorageFilters` reserves the name" explains the check by the
reservation it enforces. The real reason is different. On the V2 path the
checked column is found by position, but `BatchScanExec.filteredPartitions`
passes the widened `output` to `PushDownUtils.replanWithRuntimeFilters`, which
resolves names over it: `resolveDataSourceRuntimeFilterRefs` at
PushDownUtils.scala:212 and `getPartitionPredicateSchema` at :222.
So without this check, take a table column of that name declared in
`filterAttributes()` of a `SupportsRuntimeV2Filtering` scan with iterative
pushdown. The query would fail on an ambiguous runtime-filter attribute, or
silently drop its iterative partition predicates. d82f19a48ab gave this reason
("since `output` is matched by name later"), and 2a855b5af65 replaced it.
There are two ways to fix it:
- Restore that reason here.
- Pass `output` without the checked column to `replanWithRuntimeFilters`.
The V2 path could then drop the check, the reservation and the public
`CHECKED_COLUMN_NAME`, and the name would stay internal to `FileFormat`, where
V1 needs it.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/StorageFilterPlanning.scala:
##########
@@ -0,0 +1,95 @@
+/*
+ * 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
+
+import org.apache.spark.sql.catalyst.expressions.{Attribute,
AttributeReference, AttributeSet, Expression, ExpressionSet, If, Literal}
+import org.apache.spark.sql.connector.read.SupportsStorageFilters
+import org.apache.spark.sql.execution.WholeStageCodegenExec
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.types.BooleanType
+
+/**
+ * The planning rules file-source and DSv2 scans share for storage filters, see
+ * `FileFormat.buildReaderWithStorageFilters` and `SupportsStorageFilters`.
Each strategy decides
+ * on its own whether its scan can take any.
+ */
+object StorageFilterPlanning {
+
+ /** Whether one of `names` is the checked column's, in which case no filter
is offered. */
+ def clashesWithCheckedColumn(names: Seq[String]): Boolean =
+
names.exists(_.equalsIgnoreCase(SupportsStorageFilters.CHECKED_COLUMN_NAME))
+
+ /**
+ * The conjuncts of `filters` offered to the scan, each once, as an
`ExpressionSet` counts them.
+ * Each one:
+ * - is deterministic, since the post-scan `Filter` takes the reader's word
on the rows it
+ * checked;
+ * - references at least one column, and only `keyCandidates`;
+ * - is taken by `accepts`.
+ *
+ * Nothing is offered when the offered conjuncts reference every one of
`readColumns`, since the
+ * reader then reads the same columns for the same rows either way.
+ */
+ def offeredConjuncts(
+ filters: Seq[Expression],
+ keyCandidates: AttributeSet,
+ readColumns: AttributeSet,
+ accepts: Expression => Boolean): Seq[Expression] = {
+ val offered = ExpressionSet(filters).toSeq.filter { expr =>
+ val refs = expr.references
+ expr.deterministic && refs.nonEmpty && refs.subsetOf(keyCandidates) &&
accepts(expr)
+ }
+ if (offered.isEmpty) return Nil
+ val storageKeyAttrs =
AttributeSet.fromAttributeSets(offered.map(_.references))
+ if (readColumns.subsetOf(storageKeyAttrs)) Nil else offered
+ }
+
+ /**
+ * The column the reader marks the rows it checked in, added after `output`
when a filter is
Review Comment:
**17. [Low] This doc claims more than the gate does**
Two statements here do not hold for both callers:
- "added after `output`": `FileSourceStrategy` passes `readerDataColumns ++
partitionColumns ++ constantMetadataColumns`, but places the column right after
`readerDataColumns` (FileSourceStrategy.scala:344 and :354).
`FileSourceScanExec` relies on that position
(`output.take(requiredSchema.length)`, DataSourceScanExec.scala:818). A
maintainer who trusts this line and appends the column last on the V1 path too
would break the binding. It would be safer to say that the caller places it.
- "the column never changes how the scan runs", and the PR description's
"requested only where dropping it copies no extra row": take a post-scan
conjunct that is `CodegenFallback`, e.g. `cardinality(map_filter(m, (a, b) -> b
> 0)) > 0`. `CollapseCodegenStages` keeps the Filter out of codegen
(WholeStageCodegenExec.scala:1206-1212). The Project that drops the column then
becomes its own stage, and copies every surviving row of a columnar scan.
Without the column, no Project is planned at all.
The trade-off itself was accepted in SPARK-60034, so only the wording needs
to change.
--
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]
