peter-toth commented on code in PR #59486:
URL: https://github.com/apache/spark/pull/59486#discussion_r4414454033
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetStorageFilterSuite.scala:
##########
@@ -2676,6 +2723,171 @@ 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")
Review Comment:
**Finding 16.** This is met by the `isnotnull(k)` that the optimizer infers
from any null-intolerant condition on `k`, as noted under finding 4. The
fallback test now counts its give-up. But two others still pass when the
conjunct they test never reaches the reader. With `supportsStorageFilterOnKeys`
patched to accept only `IsNotNull`:
- `a conjunct over the key columns is evaluated only on the rows the bloom
keeps` passes. Its bloom holds only 42, so it excludes 999 rows on its own.
- `an IsNotNull over the key columns is evaluated before the other conjuncts
over them` passes. `assert_not_null(k) > 0` is never evaluated, so nothing is
given up.
Asserting that every key-only conjunct of the condition is offered fails
both under that patch, and all six tests that use the helper pass at the head:
```suggestion
val k = columnOf(relation, "k")
val offered = scanOf(plan).storageFilters
val onKey = splitConjunctivePredicates(filtered.condition)
.filter(_.references.forall(_.semanticEquals(k)))
assert(onKey.forall(c => offered.exists(_.semanticEquals(c))),
s"every conjunct over the key must be offered; got
$offered.\nPlan:\n$plan")
```
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -2088,6 +2088,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 same
exceptions as " +
Review Comment:
**Finding 20.** `StorageFilterPlanning.checkedColumn` also leaves the
checked column out when the output with it has too many fields for whole-stage
codegen (`isTooManyFields`), and then only the bloom is offered. Parquet's
`supportBatch` has no such limit, so a wide scan gets the bloom but not the key
conditions. The description names that case, but the doc doesn't:
```suggestion
"stage codegen, and few enough fields for it, take them. The answer
does not change, " +
"with the same exceptions as " +
```
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetStorageFilter.scala:
##########
@@ -149,17 +151,30 @@ 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, conditions) =
+ boundExpressions.partition(_.isInstanceOf[BloomFilterMightContain])
+ // `IsNotNull` first, as `FilterExec` checks it before the conjuncts over
its column, so a
+ // conjunct that throws on null never sees the null keys the blooms keep.
+ val onKeys = conditions.sortBy(e => if (e.isInstanceOf[IsNotNull]) 0 else
1)
Review Comment:
**Finding 15.** The sort I suggested for finding 9 is too broad.
`FilterExec` hoists only an `IsNotNull` over a column:
- `splitNotNullPreds` takes `IsNotNull(a)` only for a null-intolerant `a`
(`basicPhysicalOperators.scala:163`). One over another expression, such as a
UDF, stays in plan order.
- It checks one over a null-intolerant expression only before a conjunct
that contains the expression (`nestedNullChecks`, L370), or else after all the
others (L388).
This sort moves every `IsNotNull` first. So the reader can evaluate an
expression the plain plan never evaluates on that row.
I ran this on this head with `readKeysUnder`, under ANSI, with a bloom
holding 0 and 42 and the condition `k <> 0 AND isnotnull(100 % k)`:
- At the head, the rows equal a plain read's (`[42]`). But the reader
computes `100 % 0`, gives up the row group holding `k = 0` (one
`evaluationGivenUp` logged), and excludes 900 rows instead of 999.
- With the change below, it gives nothing up and excludes 999. The
key-condition tests still pass.
A fatal error inside such an `IsNotNull`, for example a UDF that overflows
the stack, would fail the query instead. The condition that guards it is on the
same column there, so this falls outside the one exception the new conf's doc
lists.
```suggestion
// An `IsNotNull` over a key column first, as `FilterExec` checks one
before the conjuncts
// over its column, so a conjunct that throws on null never sees the
null keys the blooms
// keep. One over an expression keeps its place, since `FilterExec`
checks it only before a
// conjunct that contains the expression, or after the others.
val onKeys = conditions.sortBy {
case IsNotNull(_: BoundReference) => 0
case _ => 1
}
```
The description's "`IS NOT NULL` goes first among them, as `FilterExec`
checks it first" would then say "an `IS NOT NULL` on the key".
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -2088,6 +2088,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 same
exceptions as " +
+ s"${PARQUET_STORAGE_FILTER_PUSHDOWN_ENABLED.key} and one more. A
condition that raises a " +
+ "fatal error, such as a regular expression or a UDF overflowing the
stack, on a row that " +
+ "a condition on another column would have dropped first fails the
query.")
Review Comment:
**Finding 17.** A fatal error does more than fail the query. The task's
`Executor.isFatalError` check hands it to `SparkUncaughtExceptionHandler`,
which calls `System.exit`. Finding 1's probe saw exactly that: the regex
overflowed the stack and took the JVM down with the task. Each retry takes down
the executor it runs on. So the stage aborts only after
`spark.task.maxFailures` attempts, each costing an executor. That is the case
worth warning about:
```suggestion
"fatal error, such as a regular expression or a UDF overflowing the
stack, on a row that " +
"a condition on another column would have dropped first exits the
executor. The query " +
"fails once the task's retries have done the same.")
```
The description's "If the error is fatal, ... the query fails" would say the
same.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategy.scala:
##########
@@ -168,24 +168,42 @@ object FileSourceStrategy extends Strategy with
PredicateHelper with Logging {
*
* Per conjunct, see [[StorageFilterPlanning.offeredConjuncts]]. The key
candidates are the
* projected data columns, and [[FileFormat.supportsStorageFilter]] must
accept the conjunct.
+ *
+ * Then, where the scan gets the checked column, every other conjunct that
references only their
+ * key columns is offered after them where
[[FileFormat.supportsStorageFilterOnKeys]] accepts it.
+ * Without the column, the post-scan `Filter` would evaluate it again on
every row the reader
+ * returns. Returns the offered conjuncts and the checked column, see
+ * [[StorageFilterPlanning.checkedColumn]], which is built only once a
conjunct is offered.
*/
private def storageFiltersFor(
normalizedFilters: Seq[Expression],
fsRelation: HadoopFsRelation,
readDataColumns: Seq[Attribute],
- readerDataSchema: StructType): Seq[Expression] = {
+ readerDataSchema: StructType,
+ scanOutput: Seq[Attribute]): (Seq[Expression],
Option[AttributeReference]) = {
val sparkSession = fsRelation.sparkSession
- if (!fsRelation.fileFormat.supportsStorageFilterPushdown(sparkSession))
return Nil
+ if (!fsRelation.fileFormat.supportsStorageFilterPushdown(sparkSession))
return (Nil, None)
if (fsRelation.schema.fieldNames.exists(
_.equalsIgnoreCase(FileFormat.STORAGE_FILTER_CHECKED_COLUMN_NAME))) {
- return Nil
+ return (Nil, None)
}
val resultSchema = StructType(fsRelation.partitionSchema.fields ++
readerDataSchema.fields)
- if (!fsRelation.fileFormat.supportBatch(sparkSession, resultSchema))
return Nil
+ if (!fsRelation.fileFormat.supportBatch(sparkSession, resultSchema))
return (Nil, None)
val dataAttrs = AttributeSet(readDataColumns)
- StorageFilterPlanning.offeredConjuncts(
+ val offered = StorageFilterPlanning.offeredConjuncts(
normalizedFilters, dataAttrs, dataAttrs,
fsRelation.fileFormat.supportsStorageFilter)
+ val checked = StorageFilterPlanning.checkedColumn(
+ sparkSession.sessionState.conf, offered, scanOutput)
+ if (checked.isEmpty) return (offered, None)
+ // A conjunct over the key columns alone adds no column to the read.
+ val storageKeyAttrs =
AttributeSet.fromAttributeSets(offered.map(_.references))
Review Comment:
**Finding 19.** An alternative to the new conf's fatal-error exception, not
a request to block on. The reader evaluates the key conditions ahead of the
query's conditions on other columns. Findings 1, 15 and 17 all come from that
reordering, and so does the exception in the conf's doc. Offering a key
condition only when the post-scan `Filter` evaluates no conjunct over another
column before it removes the reordering:
- The reader then evaluates a key condition only on rows that the plain plan
evaluates it on too, after the same conditions. The blooms only narrow those
rows.
- So an error the reader meets, fatal or not, the plain plan meets on the
same row. Nothing is lost when the reader gives a row group up, since the query
fails anyway. The doc's "one more" exception goes, and with it finding 17.
- An expensive condition no longer runs ahead of a cheaper one written
before it.
What it gives up are key conditions written after a condition on another
column. The PR's motivating cases keep theirs, since every condition on the
fact side is on the key there, in the LEFT OUTER JOIN test and in the benchmark.
I prototyped it on this head. The rule is computed over `normalizedFilters`,
whose order the post-scan `Filter` keeps. An `IsNotNull` does not count as a
conjunct over another column, since `FilterExec` checks one only before a
conjunct that reads its column:
```scala
val firstOther = normalizedFilters.indexWhere { e =>
!e.isInstanceOf[IsNotNull] && e.references.exists(a =>
dataAttrs.contains(a) &&
!storageKeyAttrs.contains(a))
}
val prefix = if (firstOther < 0) normalizedFilters else
normalizedFilters.take(firstOther)
val onKeys = ExpressionSet(prefix).toSeq.filter { expr =>
```
149 of the suite's 150 tests pass. The one that fails is `a conjunct over
the key columns that throws on a row the bloom keeps gives the row group up`.
Its key condition comes after `v <> '0'`, so it is no longer offered, which is
the point. That test would then assert that instead.
One case needs care. Once a conjunct that can be NULL is generated,
`FilterExec` checks the `IsNotNull`s written before a later conjunct ahead of
it, when that conjunct can raise (SPARK-60315). So an `IsNotNull` over another
column, written before a key condition, can drop a row first there. Counting
such an `IsNotNull` as a stop when the key condition can raise
(`ExprUtils.canEvaluateUnconditionally` is false) would close that.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategy.scala:
##########
@@ -168,24 +168,42 @@ object FileSourceStrategy extends Strategy with
PredicateHelper with Logging {
*
* Per conjunct, see [[StorageFilterPlanning.offeredConjuncts]]. The key
candidates are the
* projected data columns, and [[FileFormat.supportsStorageFilter]] must
accept the conjunct.
+ *
+ * Then, where the scan gets the checked column, every other conjunct that
references only their
+ * key columns is offered after them where
[[FileFormat.supportsStorageFilterOnKeys]] accepts it.
+ * Without the column, the post-scan `Filter` would evaluate it again on
every row the reader
+ * returns. Returns the offered conjuncts and the checked column, see
+ * [[StorageFilterPlanning.checkedColumn]], which is built only once a
conjunct is offered.
*/
private def storageFiltersFor(
normalizedFilters: Seq[Expression],
fsRelation: HadoopFsRelation,
readDataColumns: Seq[Attribute],
- readerDataSchema: StructType): Seq[Expression] = {
+ readerDataSchema: StructType,
+ scanOutput: Seq[Attribute]): (Seq[Expression],
Option[AttributeReference]) = {
val sparkSession = fsRelation.sparkSession
- if (!fsRelation.fileFormat.supportsStorageFilterPushdown(sparkSession))
return Nil
+ if (!fsRelation.fileFormat.supportsStorageFilterPushdown(sparkSession))
return (Nil, None)
if (fsRelation.schema.fieldNames.exists(
_.equalsIgnoreCase(FileFormat.STORAGE_FILTER_CHECKED_COLUMN_NAME))) {
- return Nil
+ return (Nil, None)
}
val resultSchema = StructType(fsRelation.partitionSchema.fields ++
readerDataSchema.fields)
- if (!fsRelation.fileFormat.supportBatch(sparkSession, resultSchema))
return Nil
+ if (!fsRelation.fileFormat.supportBatch(sparkSession, resultSchema))
return (Nil, None)
val dataAttrs = AttributeSet(readDataColumns)
- StorageFilterPlanning.offeredConjuncts(
+ val offered = StorageFilterPlanning.offeredConjuncts(
normalizedFilters, dataAttrs, dataAttrs,
fsRelation.fileFormat.supportsStorageFilter)
+ val checked = StorageFilterPlanning.checkedColumn(
+ sparkSession.sessionState.conf, offered, scanOutput)
+ if (checked.isEmpty) return (offered, None)
+ // A conjunct over the key columns alone adds no column to the read.
+ val storageKeyAttrs =
AttributeSet.fromAttributeSets(offered.map(_.references))
+ val onKeys = ExpressionSet(normalizedFilters).toSeq.filter { expr =>
+ val refs = expr.references
+ refs.nonEmpty && refs.subsetOf(storageKeyAttrs) &&
!offered.contains(expr) &&
+ fsRelation.fileFormat.supportsStorageFilterOnKeys(sparkSession, expr)
+ }
Review Comment:
**Finding 18.** This repeats the per-conjunct filter of
`StorageFilterPlanning.offeredConjuncts` (`StorageFilterPlanning.scala:43`),
minus its `expr.deterministic` check. That check is the one whose reason is
"the post-scan `Filter` takes the reader's word on the rows it checked". It
holds here only because `apply` passes `filters.filter(_.deterministic)`. This
also builds a second `ExpressionSet` of `normalizedFilters`, and it excludes
the blooms with a structural `Seq.contains`. The rebase note says the PR now
uses `StorageFilterPlanning` instead of its own copies, and this is the one
copy left.
The helper can take the blooms' key columns as its key candidates. Passing
`dataAttrs` as its read columns keeps its "nothing if they cover every read
column" check a no-op here, since the blooms already don't cover them. The
whole suite passes with this change (150 tests):
```suggestion
val offeredSet = ExpressionSet(offered)
val onKeys = StorageFilterPlanning.offeredConjuncts(normalizedFilters,
storageKeyAttrs,
dataAttrs, expr => !offeredSet.contains(expr) &&
fsRelation.fileFormat.supportsStorageFilterOnKeys(sparkSession,
expr))
```
--
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]
