peter-toth commented on code in PR #59486:
URL: https://github.com/apache/spark/pull/59486#discussion_r4358549563
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -2052,6 +2052,21 @@ object SQLConf {
.booleanConf
.createWithDefault(false)
+ val PARQUET_STORAGE_FILTER_PUSHDOWN_KEY_CONDITIONS_ENABLED =
+ buildConf("spark.sql.parquet.storageFilterPushdown.keyConditions.enabled")
+ .doc(s"If true, and ${PARQUET_STORAGE_FILTER_PUSHDOWN_ENABLED.key}
applies a bloom filter, " +
+ "the reader also applies the query's other filter conditions that
reference only the " +
+ "bloom filter's columns, on the rows the bloom filter keeps. They then
run before the " +
+ "query's conditions on other columns. Only scans that return columnar
batches with whole-" +
+ "stage codegen take them. The answer does not change, with the
exceptions of " +
Review Comment:
**Finding 13.** As rendered, this says the answer does not change "with the
exceptions of spark.sql.parquet.storageFilterPushdown.enabled and one more",
which names a conf as the exception. The exceptions meant are the two that
conf's doc lists:
```suggestion
"stage codegen take them. The answer does not change, with the same
exceptions as " +
```
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetStorageFilterSuite.scala:
##########
@@ -1066,6 +1073,41 @@ class ParquetStorageFilterSuite extends QueryTest with
SharedSparkSession
}
}
+ test("a LEFT OUTER JOIN applies a condition on the fact side's join key with
its runtime bloom") {
+ // The fact side of a LEFT OUTER JOIN filters on its join key in a derived
table. The
+ // optimizer does not infer that condition onto the side the bloom is
built from, so the bloom
+ // keeps keys 0 until 20 while the condition keeps only every fourth of
them. With both in the
+ // reader, it excludes more rows than the bloom alone can.
+ withBloomFilterTables {
+ val query =
+ """SELECT bf2.k, f.v
+ |FROM bf2 LEFT JOIN (SELECT * FROM bf1 WHERE k % 4 = 0) f ON f.k =
bf2.k
+ |WHERE bf2.v < 20
+ |""".stripMargin
+ def run(pushdown: Boolean): (SparkPlan, Seq[Row]) = withSQLConf(
+ SQLConf.PARQUET_STORAGE_FILTER_PUSHDOWN_ENABLED.key ->
pushdown.toString,
+ SQLConf.PARQUET_STORAGE_FILTER_PUSHDOWN_KEY_CONDITIONS_ENABLED.key
-> pushdown.toString,
+ SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false") {
+ val df = spark.sql(query)
+ (df.queryExecution.executedPlan,
df.collect().toSeq.sortBy(_.getLong(0)))
+ }
+ val (_, plain) = run(pushdown = false)
+ val (plan, rows) = run(pushdown = true)
+ assert(rows == plain && rows.size == 20, s"got $rows, a plain read got
$plain")
+ val offered = collect(plan) { case s: FileSourceScanExec =>
s.storageFilters }.flatten
+ assert(offered.exists(_.isInstanceOf[BloomFilterMightContain]) &&
+ offered.exists(_.exists(_.isInstanceOf[Remainder])) &&
+ offered.exists(_.isInstanceOf[IsNotNull]),
+ s"expected the bloom, k % 4 = 0 and isnotnull(k); got
$offered.\nPlan:\n$plan")
+ val bareNotNull = collect(plan) { case f: FilterExec => f.condition }
+ .flatMap(splitConjunctivePredicates).exists(_.isInstanceOf[IsNotNull])
Review Comment:
**Finding 12.** This passes without the `IsNotNull` exception it checks. The
`collect` takes every `FilterExec` in the plan, and `bf2`'s own Filter is
`isnotnull(v) AND (v < 20)`. With `&& !f.isInstanceOf[IsNotNull]` removed at
`FileSourceStrategy.scala:457`, this test still passes, and only the planner
test fails. Scoped to the Filter that reads the checked column, it fails there
and passes at the head:
```suggestion
// Only the fact side's Filter reads the checked column. bf2's own
Filter holds a bare
// `isnotnull(v)`, which would satisfy this on its own.
val bareNotNull = collect(plan) {
case f: FilterExec
if f.references.exists(_.name ==
FileFormat.STORAGE_FILTER_CHECKED_COLUMN_NAME) =>
f.condition
}.flatMap(splitConjunctivePredicates).exists(_.isInstanceOf[IsNotNull])
```
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetStorageFilterSuite.scala:
##########
@@ -2249,6 +2291,151 @@ class ParquetStorageFilterSuite extends QueryTest with
SharedSparkSession
}
}
+ test("FileSourceStrategy offers a conjunct over the bloom's key columns
along with the bloom") {
+ // It adds no column to the read, so it goes to the reader after the
bloom, an `IsNotNull`
+ // included, which the Filter keeps bare. A conjunct that also references
another column does
+ // not, nor one over the key with no bloom to go with. Nor any of them
with its own conf off, or
+ // where the scan cannot mark the rows the reader checked.
+ withTempDir { dir =>
+ val path = writeKeyParquetFile(dir, 0L until 50L, rowGroupSize = 64 *
1024L)
+ withSQLConf(
+ SQLConf.PARQUET_STORAGE_FILTER_PUSHDOWN_ENABLED.key -> "true",
+ SQLConf.PARQUET_STORAGE_FILTER_PUSHDOWN_KEY_CONDITIONS_ENABLED.key
-> "true") {
+ val relation = spark.read.parquet(path).queryExecution.optimizedPlan
+ val k = columnOf(relation, "k")
+ val v = columnOf(relation, "v")
+ val bloom = bloomOn(k)
+ val notNull = IsNotNull(k)
+ val onKey = EqualTo(Remainder(k, Literal(4L)), Literal(0L))
+ val onValue = EqualTo(v, Literal("42"))
+ val onBoth = GreaterThan(Add(k, Cast(v, LongType)), Literal(0L))
+ val condition = And(And(And(And(notNull, onKey), bloom), onValue),
onBoth)
+ def offered(condition: Expression): Seq[Expression] =
+ scanOf(planned(LogicalFilter(condition, relation))).storageFilters
+ val withBloom = offered(condition)
+ assert(withBloom.length == 3 && withBloom.head.semanticEquals(bloom) &&
+ withBloom.tail.exists(_.semanticEquals(notNull)) &&
+ withBloom.tail.exists(_.semanticEquals(onKey)),
+ s"expected the bloom, then $notNull and $onKey; got $withBloom")
+ val filter = planned(LogicalFilter(condition, relation)).collectFirst {
+ case f: FilterExec => f
+ }.get
+
assert(splitConjunctivePredicates(filter.condition).exists(_.semanticEquals(notNull)),
+ s"the Filter must keep $notNull bare; got ${filter.condition}")
+ assert(offered(And(onKey, onValue)).isEmpty, "nothing is offered
without a bloom")
+
withSQLConf(SQLConf.PARQUET_STORAGE_FILTER_PUSHDOWN_KEY_CONDITIONS_ENABLED.key
-> "false") {
+ assert(offered(condition) == Seq(bloom), "only the bloom with the
conf off")
+ }
+ withSQLConf(SQLConf.WHOLESTAGE_CODEGEN_ENABLED.key -> "false") {
+ assert(offered(condition) == Seq(bloom), "only the bloom without the
checked column")
+ }
+ assert(!ParquetStorageFilter.isSupportedFilterOnKeys(
+ EqualTo(k, ScalarSubquery(spark.range(1).queryExecution.analyzed))),
+ "a conjunct holding a subquery is not offered")
+ }
+ }
+ }
+
+ // Reads keys 0 until 1000 under `condition` with the feature off and on,
and hands `check` the
+ // rows of each and the rows the reader excluded. `condition` takes the key
and value columns.
+ // With `oddKeysNull`, every odd key is null.
+ private def readKeysUnder(
+ condition: (Attribute, Attribute) => Expression, oddKeysNull: Boolean =
false)(
+ check: (Seq[Long], Seq[Long], Long) => Unit): Unit = withTempDir { dir =>
+ val keys = (0L until 1000L).map(k => Option.when(!oddKeysNull || k % 2 ==
0)(k))
+ val path = writeKeyParquetFile(dir, keys, rowGroupSize = 1024L)
+ val relation = spark.read.parquet(path).queryExecution.analyzed
+ val filtered =
+ LogicalFilter(condition(columnOf(relation, "k"), columnOf(relation,
"v")), relation)
+ def run(pushdown: Boolean): (SparkPlan, Seq[Long]) = withSQLConf(
+ SQLConf.PARQUET_STORAGE_FILTER_PUSHDOWN_ENABLED.key ->
pushdown.toString,
+ SQLConf.PARQUET_STORAGE_FILTER_PUSHDOWN_KEY_CONDITIONS_ENABLED.key ->
pushdown.toString) {
+ val df = classic.Dataset.ofRows(spark, filtered)
+ (df.queryExecution.executedPlan,
df.collect().toSeq.map(_.getLong(0)).sorted)
+ }
+ val (_, plain) = run(pushdown = false)
+ val (plan, rows) = run(pushdown = true)
+ assert(scanOf(plan).storageFilters.length > 1,
+ s"the key conjuncts must be offered with the bloom.\nPlan:\n$plan")
+ check(plain, rows, rowsExcludedByStorageFilters(plan))
+ }
+
+ test("a conjunct over the key columns narrows the rows the reader keeps") {
+ // The reader keeps exactly the rows the query returns, where the bloom
alone would also keep
+ // the even keys. The small bloom has false positives too, which the
conjunct keeps if odd.
+ readKeysUnder { (k, _) =>
+ And(BloomFilterMightContain(bloomLiteralOf(41L, 42L, 43L, 501L), new
XxHash64(Seq(k))),
+ EqualTo(Remainder(k, Literal(2L)), Literal(1L)))
+ } { (plain, rows, excluded) =>
+ assert(rows == plain, s"got $rows, a plain read got $plain")
+ assert(Seq(41L, 43L, 501L).forall(rows.contains) && rows.forall(_ % 2 ==
1),
+ s"expected the odd keys the bloom holds; got $rows")
+ assert(excluded == 1000L - rows.size, "the reader must keep exactly the
rows it returns")
+ }
+ }
+
+ test("a conjunct over the key columns is evaluated only on the rows the
bloom keeps") {
+ // The two conjuncts share `100 % k`, which throws on k = 0 under ANSI. In
one `And` with the
+ // bloom it would be computed on every row, k = 0 included, and the reader
would give up that
+ // row group. Under an `If` on the bloom it is computed only on the rows
the bloom keeps.
+ withSQLConf(SQLConf.ANSI_ENABLED.key -> "true") {
+ readKeysUnder { (k, _) =>
+ val mod = Remainder(Literal(100L), k)
+ And(And(bloomOn(k), GreaterThanOrEqual(mod, Literal(0L))),
+ LessThanOrEqual(mod, Literal(99L)))
+ } { (plain, rows, excluded) =>
+ assert(rows == Seq(42L) && rows == plain, s"got $rows, a plain read
got $plain")
+ assert(excluded == 999L, s"the reader must not give a row group up; it
excluded $excluded")
+ }
+ }
+ }
+
+ test("every conjunct over the key columns is enforced, not only the first") {
+ // Each conjunct rejects a key the bloom keeps that the other keeps: 3
fails `k % 3 <> 0` and 5
+ // fails `k % 5 <> 0`. The Filter skips both on the rows the reader marks,
so the reader has to
+ // apply them together. Neither is pushed to Parquet, so no statistic
drops those rows first.
+ readKeysUnder { (k, _) =>
+ And(And(BloomFilterMightContain(bloomLiteralOf(3L, 5L, 7L), new
XxHash64(Seq(k))),
+ Not(EqualTo(Remainder(k, Literal(3L)), Literal(0L)))),
+ Not(EqualTo(Remainder(k, Literal(5L)), Literal(0L))))
+ } { (plain, rows, excluded) =>
+ assert(rows == plain, s"got $rows, a plain read got $plain")
+ assert(rows.contains(7L) && !rows.contains(3L) && !rows.contains(5L),
+ s"expected 7 and neither 3 nor 5; got $rows")
+ assert(excluded == 1000L - rows.size, "the reader must keep exactly the
rows it returns")
+ }
+ }
+
+ test("a conjunct over the key columns that throws on a row the bloom keeps
gives the row " +
+ "group up") {
+ // Under ANSI, `100 % k` throws on k = 0, which the bloom keeps, while `v
<> '0'` drops it
+ // first in the Filter. The reader evaluates the key conjunct first, so it
gives that row group
+ // up, and the Filter then decides in plan order, with the same rows as a
plain read.
+ withSQLConf(SQLConf.ANSI_ENABLED.key -> "true") {
+ readKeysUnder { (k, v) =>
+ And(And(BloomFilterMightContain(bloomLiteralOf(0L, 42L), new
XxHash64(Seq(k))),
+ Not(EqualTo(v, Literal("0")))),
GreaterThanOrEqual(Remainder(Literal(100L), k),
+ Literal(0L)))
+ } { (plain, rows, excluded) =>
+ assert(rows == plain, s"got $rows, a plain read got $plain")
+ assert(rows.contains(42L) && !rows.contains(0L), s"expected 42 and not
0; got $rows")
+ assert(excluded < 1000L - rows.size,
Review Comment:
**Finding 4.** This holds whether or not the reader gave anything up:
- The bloom keeps `0` and `42`, so a reader that never sees `100 % k >= 0`
still excludes 998 rows, and `998 < 999`.
- The guard in `readKeysUnder` (`storageFilters.length > 1`, L2358) is met
by the inferred `isnotnull(k)` alone.
With `supportsStorageFilterOnKeys` patched to accept only `IsNotNull`, the
throwing conjunct is never offered, and this test still passes. The suite's
`withLogged(evaluationGivenUp)`, as used at L839, pins the give-up itself.
Wrapping this test's `readKeysUnder` call in it counts 1 at the head and 0 with
that patch:
```scala
val (_, givenUp) = withLogged(evaluationGivenUp) {
readKeysUnder { (k, v) => ... } { (plain, rows, excluded) => ... }
}
assert(givenUp > 0, "the reader must give up the row group holding k =
0")
```
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategy.scala:
##########
@@ -345,20 +359,23 @@ object FileSourceStrategy extends Strategy with
PredicateHelper with Logging {
// Offered conjuncts become `storageFilters` on the scan and stay in the
post-scan Filter too.
// This runs here because eligibility depends on the data columns the
reader reads.
val readerDataColumns = readDataColumns ++ generatedMetadataColumns
- val storageFilters = storageFiltersFor(
- normalizedFilters, fsRelation, readDataColumns,
readerDataColumns.toStructType)
// The column the reader marks the rows it checked in, see
// `FileFormat.buildReaderWithStorageFilters`. It is added only where
the scan still returns
// columnar batches with it, as `FileSourceScanExec.supportsColumnar`
decides, so the column
// never changes how the scan runs. Above a row-based scan, the Project
that drops the
// column would copy every surviving row.
- val storageFilterCheckedColumn = Option.when(storageFilters.nonEmpty) {
+ val checkedColumn =
AttributeReference(FileFormat.STORAGE_FILTER_CHECKED_COLUMN_NAME,
BooleanType)()
- }.filter { checked =>
+ val marksChecked = {
Review Comment:
**Finding 14.** Before this push, the checked column and the
`isTooManyFields` walk ran only when `storageFilters` was non-empty
(`Option.when(storageFilters.nonEmpty) { ... }.filter { ... }`). Now every file
scan the planner sees builds the `AttributeReference` with a fresh `ExprId`,
builds the output `StructType`, and walks its fields, even with the feature off.
Making both `lazy val`s, and having `storageFiltersFor` take `marksChecked:
=> Boolean`, forces them only after a bloom is offered.
`Option.when(storageFilters.nonEmpty && marksChecked)` then short-circuits too.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetStorageFilter.scala:
##########
@@ -142,17 +144,26 @@ object ParquetStorageFilter {
maxSplicedRowGroupBytes: Long): ParquetStorageFilter = {
require(boundExpressions.nonEmpty,
"storage filters must be non-empty; callers with nothing to push must
not call create")
- val expr = boundExpressions.reduce(And)
+ // The conjuncts over the blooms' key columns go under an `If` on the
blooms, so they run only
+ // on the rows the blooms keep. In one `And`, subexpression elimination
would compute what
+ // they share on every row.
+ val (blooms, onKeys) =
boundExpressions.partition(_.isInstanceOf[BloomFilterMightContain])
Review Comment:
**Finding 9.** `onKeys` keeps the order of the plan's conjuncts. So for
`cond AND isnotnull(k)` the reader evaluates `cond` first, on every key the
bloom keeps, null keys included.
`FilterExec` does the opposite. Before each conjunct, it emits the
`IsNotNull` checks of the columns that conjunct references
(`basicPhysicalOperators.scala:193-209`). So in a plain read `cond` never sees
a null `k`. Null keys pass the bloom whenever its build side has one, which is
why `IsNotNull` is offered.
I ran a probe with `readKeysUnder` and `oddKeysNull = true`. The bloom held
`42` and a null key, and the condition was `AssertNotNull(k) > 0 AND
isnotnull(k)`. That conjunct throws on null, as a UDF over a boxed or string
key can.
- At the head, the rows equal a plain read's (`[42]`), but the reader gives
every row group up, so `excluded = 0`.
- With the change below, the rows are the same and `excluded = 999`. The
suite still passes.
Moving `IsNotNull` first is enough (with `IsNotNull` added to the import):
```suggestion
val (blooms, conditions) =
boundExpressions.partition(_.isInstanceOf[BloomFilterMightContain])
// `IsNotNull` first, as `FilterExec` checks it before the conjuncts
over its column.
val onKeys = conditions.sortBy(e => if (e.isInstanceOf[IsNotNull]) 0
else 1)
```
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileFormat.scala:
##########
@@ -193,9 +193,13 @@ trait FileFormat {
* A format that does not apply storage filters returns `None`, the default,
and the caller then
* builds an ordinary reader. It is an `Option` rather than a call to that
builder, so the two
* builders cannot call each other. `storageFilters` arrive with their
scalar subqueries
- * materialized, bound to `requiredSchema`, while [[supportsStorageFilter]]
is asked about the
- * unbound expression. `storageFilterMetrics` are the ones this format's
[[storageFilterMetrics]]
- * created, which the scan shows in the SQL UI.
+ * materialized, bound to `requiredSchema`, while [[supportsStorageFilter]]
and
Review Comment:
**Finding 10.** Two sentences of this doc no longer hold for the key
conditions:
- L177-178, "A reader must not fail the query for that". A key condition's
fatal error does fail it now, as the new conf's doc says, since
`rethrowIfMustPropagate` rethrows it.
- L183, the `Filter` "evaluates each storage filter as `IF(checked, true,
filter)`". An offered `IsNotNull` stays bare (`FileSourceStrategy.scala:457`).
Both are outside the diff, so here is the text rather than a suggestion:
```scala
* an error on a row a plain scan never evaluates it on. A reader must not
fail the query for
* a non-fatal one. It gives the filter up there and lets the `Filter`
decide.
```
```scala
* of the relation, and the `Filter` evaluates each storage filter as
`IF(checked, true, filter)`,
* except an `IsNotNull`, which it keeps bare.
```
--
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]
