dongjoon-hyun commented on code in PR #59715:
URL: https://github.com/apache/spark/pull/59715#discussion_r4379088695
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/DataSourceScanExec.scala:
##########
@@ -817,8 +817,8 @@ case class FileSourceScanExec(
protected lazy val preparedStorageFilters: Seq[Expression] = {
val requestedDataAttrs = output.take(requiredSchema.length)
storageFilters.map { expr =>
- val subqueryReplaced = expr.transform { case s: execution.ScalarSubquery
=> s.toLiteral }
- BindReferences.bindReference(subqueryReplaced, requestedDataAttrs)
+ BindReferences.bindReference(
+ StorageFilterPlanning.withSubqueryResults(expr), requestedDataAttrs)
Review Comment:
**3. [Low] No V1 test covers the per-bloom literal in
`preparedStorageFilters`**
This changes the V1 Parquet path too. A bloom that `MergeSubplans` merged
with another used to ship the whole merged struct to every task, and now ships
only its own bytes. But `ParquetStorageFilterSuite` changed only an import and
one call, and none of its tests runs a two-key bloom join. The helper's unit
test ("a bloom read out of a merged subquery's struct ships as a literal of its
own" in `StorageFilterPushDownSuite`) does not go through this call. So if this
line went back to `s.toLiteral`, every test would still pass. Could
`ParquetStorageFilterSuite` get a two-key counterpart of the new V2 test, one
that checks that each bloom reaches the reader once?
##########
sql/catalyst/src/main/java/org/apache/spark/sql/connector/read/Scan.java:
##########
@@ -70,9 +70,10 @@ default String description() {
* {@link Table} that creates this scan returns {@link
TableCapability#BATCH_READ} support in its
* {@link Table#capabilities()}.
* <p>
- * If the scan supports runtime filtering and implements {@link
SupportsRuntimeFiltering},
- * this method may be called multiple times. Therefore, implementations can
cache some state
- * to avoid planning the job twice.
+ * If the scan supports runtime filtering and implements {@link
SupportsRuntimeFiltering}, or
+ * Spark offers it a storage filter through {@link SupportsStorageFilters},
this method may be
Review Comment:
**6. [Nit] `toBatch` can be called more than once for every scan, not only
these two kinds**
As your
[reply](https://github.com/apache/spark/pull/59715#discussion_r4377264760) to
item 7 notes, `planInputPartitions` is already called twice with the conf off.
In the default `PARTITION_DEFINED` mode, `DataSourceV2Strategy` forces `batch`
through `supportsColumnar` (DataSourceV2Strategy.scala:240). The executed plan
is then built from `sparkPlan.clone()` (QueryExecution.scala:393), and that
copy calls `toBatch` again. `Batch.planInputPartitions` now says so without a
condition. This doc still reads as if a scan with neither runtime filtering nor
a storage filter is planned once, so a connector may skip caching it needs.
Maybe use the same wording as in `Batch`: "Spark may call this method more than
once for the same scan, for instance when it plans the scan again for runtime
filtering or for a storage filter."
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/StorageFilterPushDownSuite.scala:
##########
@@ -0,0 +1,491 @@
+/*
+ * 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.{SparkException, SparkThrowable}
+import org.apache.spark.sql.{DataFrame, QueryTest}
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions.{Attribute, AttributeMap,
AttributeReference, AttributeSet, BloomFilterMightContain, Cast, Divide,
EvalMode, GetStructField, GreaterThan, If, Literal, NumericEvalContext,
PredicateHelper, Rand, XxHash64}
+import org.apache.spark.sql.catalyst.plans.logical.ReturnAnswer
+import org.apache.spark.sql.execution.{BaseLimitExec, CoalesceExec,
ExecSubqueryExpression, 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,
DataSourceV2ScanRelation}
+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.{BinaryType, DoubleType, LongType,
StructType}
+
+/**
+ * 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 with
PredicateHelper {
+
+ 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")
Review Comment:
**2. [Low] The new multi-bloom and multi-reader tests run only with AQE off**
`withBloomJoin` turns AQE off by default, and only the first test loop
(L127-131) turns it on, over one bloom and one partition. So the two-key join
(L329), the task with two readers (L370), the runtime-filtering tests and the
missing-column tests all run without AQE, while AQE is the default.
Under AQE the scan that runs is a different node. `PlanAdaptiveSubqueries`
gives each occurrence of the bloom's subquery its own `SubqueryExec`, so
`ReuseAdaptiveSubquery` turns one of the two into a `ReusedSubqueryExec`
(ReuseAdaptiveSubquery.scala:38). For the scan that means a rebuilt
`BatchScanExec`, as in item 7. For the two-key join, its `storageFilters` then
read the blooms through
`GetStructField(ScalarSubquery(ReusedSubqueryExec(...)), i)`. No test checks
that `withSubqueryResults` still gives each bloom a literal of its own there,
that both conjuncts are still skipped on checked rows, or that the readers of a
task still share one filter. Could the two-key and the two-reader tests also
run with `aqe <- Seq(false, true)`?
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceRDD.scala:
##########
@@ -254,6 +279,7 @@ private class PartitionIterator[T](
}
valuePrepared = false
val result = reader.get()
+ check.foreach(_(result))
Review Comment:
**5. [Nit] The width check now runs a capturing lambda per row**
Since `cb01a6f`, this runs for every row of a row-based read. `_(result)`
captures `result`, so each call allocates a lambda unless the JIT removes it,
and then calls the check function before it reads `numFields`. That is cheap
next to reading a row, but it is on the per-row path of every row-based scan
that has the checked column. A `match` avoids the capture:
```scala
check match {
case Some(c) => c(result)
case None =>
}
```
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/StorageFilterPlanning.scala:
##########
@@ -0,0 +1,103 @@
+/*
+ * 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, BloomFilterMightContain, Expression,
ExpressionSet, If, Literal}
+import org.apache.spark.sql.execution.{ScalarSubquery => ExecScalarSubquery,
WholeStageCodegenExec}
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.types.{BinaryType, BooleanType}
+
+/**
+ * The 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 {
+
+ /**
+ * 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, when a filter is
`offered`. The caller
+ * places it in the scan's output. Only where whole-stage codegen is on and
`output` with the
+ * column still fits it, so the column keeps the scan in whole-stage
codegen. Elsewhere the
+ * Project that drops the column could copy every surviving row. A post-scan
conjunct that cannot
+ * be generated still puts that Project in a stage of its own, the trade-off
SPARK-60034 accepted.
+ */
+ def checkedColumn(
+ conf: SQLConf,
+ offered: Seq[Expression],
+ output: Seq[Attribute]): Option[AttributeReference] = {
+ if (offered.isEmpty) return None
+ val checked =
AttributeReference(FileFormat.STORAGE_FILTER_CHECKED_COLUMN_NAME, BooleanType)()
+ Option.when(conf.wholeStageEnabled &&
+ !WholeStageCodegenExec.isTooManyFields(conf, (output :+
checked).toStructType))(checked)
+ }
+
+ /**
+ * `filters` with each of `offered` skipped on the rows `checked` marks, or
as they are without
+ * the column. Matched by `ExpressionSet`, as `offered` was built. An `If`
keeps two things an
+ * `Or` would lose:
+ * - a null mark leaves the conjunct as it was, an `And` around it included;
+ * - subexpression elimination cannot hoist what the conjunct shares with
another one ahead of
+ * the skip.
+ */
+ def skipCheckedRows(
+ filters: Seq[Expression],
+ offered: Seq[Expression],
+ checked: Option[Attribute]): Seq[Expression] = checked match {
+ case Some(c) =>
+ val offeredSet = ExpressionSet(offered)
+ filters.map(f => if (offeredSet.contains(f)) If(c, Literal.TrueLiteral,
f) else f)
+ case None => filters
+ }
+
+ /**
+ * `filter` with each scalar subquery replaced by its result, for the
readers. Each bloom filter
+ * becomes a literal of its own bytes. A subquery `MergeSubplans` merged
returns the blooms of
+ * several filters in one struct, so a bloom read out of it would otherwise
carry the whole
+ * struct.
+ */
+ def withSubqueryResults(filter: Expression): Expression = filter.transformUp
{
+ case s: ExecScalarSubquery => s.toLiteral
+ case m @ BloomFilterMightContain(bloom, _) if bloom.foldable &&
!bloom.isInstanceOf[Literal] =>
Review Comment:
**4. [Nit] `withSubqueryResults` folds only a bloom's argument**
The fix for item 6 of my previous review
([thread](https://github.com/apache/spark/pull/59715#discussion_r4372009034))
special-cases `BloomFilterMightContain`. Any other conjunct that reads a field
of a merged subquery still gets the whole struct, and that struct holds every
bloom of the merge. Only blooms are offered today, so this costs nothing yet.
But the description names the key-column conjuncts of SPARK-60448 as the next
step. If those come from a subquery that `MergeSubplans` merges with the
blooms, a conjunct such as `k <= GetStructField(merged, 2)` would ship all the
blooms again, once per conjunct. Folding the field access covers both cases:
```scala
def withSubqueryResults(filter: Expression): Expression = filter.transformUp
{
case s: ExecScalarSubquery => s.toLiteral
case g @ GetStructField(_: Literal, _, _) => Literal(g.eval(), g.dataType)
}
```
Unlike folding any foldable expression here, a `GetStructField` over a
literal cannot fail.
##########
sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/filter/StorageFilter.java:
##########
@@ -0,0 +1,105 @@
+/*
+ * 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.io.Serializable;
+
+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, an aggregate or a join.
+ * <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. A reader must not copy or
serialize it, since an
+ * evaluated instance may have released what a copy would be built from.
+ *
+ * @since 5.0.0
+ */
+@Evolving
+public abstract class StorageFilter implements Serializable {
Review Comment:
**8. [Nit] Should `StorageFilter` move to `connector.read`?**
Since the change for item 4
([thread](https://github.com/apache/spark/pull/59715#discussion_r4372009027)),
`StorageFilter` is a standalone class, not a `Predicate`. Yet it stays in
`connector.expressions.filter`, with the V2 expressions `Predicate`,
`AlwaysTrue` and `PartitionPredicate`. A connector gets it only through
`StorageFilterRequest`, which sits in `connector.read` next to
`SupportsStorageFilters`, and never inside an expression tree. Both classes are
new, so moving them, with `StorageFilterNotAppliedException`, costs nothing
before a release and would be hard after one. Keeping them where they are is
fine too if you prefer the filter types together.
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/StorageFilterTestSource.scala:
##########
@@ -0,0 +1,444 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.connector
+
+import java.io.{ByteArrayOutputStream, ObjectOutputStream}
+import java.util
+import java.util.OptionalLong
+import java.util.concurrent.atomic.{AtomicInteger, AtomicLong}
+
+import scala.collection.mutable
+import scala.jdk.CollectionConverters._
+
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions.{Expression,
GenericInternalRow}
+import org.apache.spark.sql.connector.catalog.{MetadataColumn,
SupportsMetadataColumns, SupportsRead, Table, TableCapability, TableProvider}
+import org.apache.spark.sql.connector.expressions.{Expressions,
FieldReference, NamedReference, SortOrder, Transform}
+import org.apache.spark.sql.connector.expressions.filter.{Predicate,
StorageFilter, StorageFilterNotAppliedException}
+import org.apache.spark.sql.connector.read._
+import org.apache.spark.sql.execution.vectorized.OnHeapColumnVector
+import org.apache.spark.sql.internal.connector.SupportsRuntimeCatalystFiltering
+import org.apache.spark.sql.types.{BooleanType, DataType, 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 longs that implements
[[SupportsStorageFilters]]:
+ * the key `k`, the value `v` holding `k * 10`, and with `wide` also `w`
holding `k * 100`. Its
+ * reader gets the filter in the task and drops the rows the filter rejects.
It reads the columns
+ * Spark requires, and builds the filter's row from the filter's
`references()`. 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 partition;
+ * - `markAllChecked`: reads without the filter but marks every row checked,
which breaks the
+ * contract on purpose;
+ * - `dropChecked`: `true` never appends the checked column, and `later`
appends it to the first
+ * row or batch only. Either 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;
+ * - `wide`: adds the column `w`;
+ * - `numPartitions`: splits the keys into that many input partitions, 1 by
default;
+ * - `metadataColumn`: adds the metadata column `_pos`, which holds the key;
+ * - `runtimeFiltering`: `v2` makes the scan a `SupportsRuntimeV2Filtering`
one over every column,
+ * with iterative pushdown on a table partitioned by `k`. `catalyst` makes
it a
+ * `SupportsRuntimeCatalystFiltering` one that evaluates the filters on `k`
fully. Neither
+ * applies the filters it takes, so they serve planning tests only;
+ * - `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 = {
+ val schema = new StructType().add("k",
LongType).add(options.getOrDefault("valueName", "v"),
+ LongType)
+ if (options.getBoolean("wide", false)) schema.add("w", LongType) else
schema
+ }
+
+ override def getTable(
+ schema: StructType,
+ partitioning: Array[Transform],
+ properties: util.Map[String, String]): Table = {
+ val options = new CaseInsensitiveStringMap(properties)
+ if (options.getBoolean("metadataColumn", false)) {
+ new StorageFilterTestTable(schema, options) with SupportsMetadataColumns
{
+ override def metadataColumns(): Array[MetadataColumn] = Array(new
MetadataColumn {
+ override def name(): String = StorageFilterTestSource.posName
+ override def dataType(): DataType = LongType
+ })
+ }
+ } else {
+ new StorageFilterTestTable(schema, options)
+ }
+ }
+}
+
+object StorageFilterTestSource {
+
+ /** Rows the source serves, `k` from 0 until numRows. */
+ val numRows: Int = 200
+
+ /** Rows a batch holds at most in a columnar read. */
+ val batchSize: Int = 64
+
+ /** The name of the metadata column the `metadataColumn` option adds. */
+ val posName: String = "_pos"
+
+ /** 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)
+
+ /** The serialized size of the last filter a reader got. */
+ val filterBytes = new AtomicLong(0)
+
+ /** The references of the filters the readers got, and the identities of
those filters. */
+ val filterReferences = mutable.Set.empty[Seq[String]]
+ val filterIdentities = mutable.Set.empty[Int]
+
+ /** How many times a runtime-filtering scan was handed filters. */
+ val runtimeFilterCalls = new AtomicInteger(0)
+
+ def reset(): Unit = {
+ filteredReaders.set(0)
+ checkedColumnRequests.set(0)
+ rowsEvaluated.set(0)
+ rowsReturned.set(0)
+ filterBytes.set(0)
+ filterReferences.synchronized(filterReferences.clear())
+ filterIdentities.synchronized(filterIdentities.clear())
+ runtimeFilterCalls.set(0)
+ }
+
+ /** The value of column `name` in the row of `key`. Any column but `k`, `w`
and `_pos` is `v`. */
+ def valueOf(name: String, key: Long): Long = name match {
+ case "k" | `posName` => key
+ case "w" => key * 100
+ case _ => key * 10
+ }
+
+ /** The columns of the row the filter evaluates, in that order. */
+ def columnsOf(filter: StorageFilter): Seq[String] =
filter.references.map(_.fieldNames.head).toSeq
+
+ // Records the filter a reader got, before the reader evaluates it.
+ private[connector] def record(filter: StorageFilter): Unit = {
+ val bytes = new ByteArrayOutputStream()
+ val out = new ObjectOutputStream(bytes)
+ try out.writeObject(filter) finally out.close()
Review Comment:
**7. [Nit] The test source serializes the shared filter, against the new
doc**
`filterOf` calls `record(request.filter())` for every reader (L284), and
this line serializes the filter. The `StorageFilter` doc now says "A reader
must not copy or serialize it" (StorageFilter.java:52). In a task with two
readers, the second one records the filter after the first has evaluated it and
released its expression. `filterBytes`, "The serialized size of the last filter
a reader got" (L117), then drops to a size without the bloom. The two-key test
reads one partition, so its assertion at StorageFilterPushDownSuite.scala:361
holds today. With `numPartitions -> 2` and a `coalesce(1)`, it would pass even
if each bloom shipped twice. Recording the size only for the first reader, e.g.
with `filterBytes.compareAndSet(0, bytes.size())`, keeps it right.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceRDD.scala:
##########
@@ -204,6 +221,13 @@ class DataSourceRDD(
}
}
+ private def checkWidth(expected: Int, actual: Int): Unit = {
+ if (actual != expected) {
Review Comment:
**1. [Low] The width check passes a misplaced checked column and fails a
wider batch**
This compares only the number of columns, with `!=`. Two cases come out
differently from what the `StorageFilterRequest.appendsCheckedColumn` doc and
the error message say:
- A reader that adds the checked column in front of the partition or
metadata columns it appends last returns `n + 1` columns, so the check passes.
`ColumnarToRowExec` binds `output` to `batch.column(i)` by position
(Columnar.scala:149-152 and :175). The first of those columns then reads the
boolean vector, and the checked attribute reads the last of them. That fails
with an NPE, or, when the types happen to match, returns shifted values and
skips the bloom on whatever rows that last column marks, with no error.
- A reader that appends the column at position `n`, as documented, and
returns more columns after it now fails with
`DATA_SOURCE_STORAGE_FILTER_CHECKED_COLUMN_MISSING`, although the column is
there. Without a request, the plan above the scan reads only the first
`output.length` columns of such a batch or row.
For a columnar read, also checking `batch.column(expected - 1).dataType() ==
BooleanType` would catch the first case at one call per batch. For the second,
either `actual < expected`, or a message that says the width is wrong rather
than that the column is missing, would do. This is the check from item 2 of my
previous review
([thread](https://github.com/apache/spark/pull/59715#discussion_r4372009016)).
--
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]
