peter-toth commented on code in PR #59486:
URL: https://github.com/apache/spark/pull/59486#discussion_r4348267872
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetStorageFilter.scala:
##########
@@ -216,6 +225,17 @@ object ParquetStorageFilter {
case _ => false
}
+ /**
+ * Whether the reader can evaluate `expr`, a conjunct over the key columns
of the accepted
+ * storage filters, together with them: no subquery, and nothing unevaluable
but column
+ * references.
+ */
+ def isSupportedFilterOnKeys(expr: Expression): Boolean =
Review Comment:
**Finding 1.** The reader evaluates these conjuncts before the plan's
conditions on other columns. On a non-fatal error it gives the row group up,
and the `Filter` decides in plan order. On a fatal one,
`rethrowIfMustPropagate` rethrows. So a key conjunct that fails fatally on a
row another condition drops now fails a query the feature off answers.
I ran this on this head, over keys `"short"`, `"a" * 50000` and `"other"`,
with `v = k` and a bloom holding the first two:
- with the feature off, `bloom AND v = 'short' AND k RLIKE '(a|b)*c'` counts
0 rows, since `v = 'short'` drops the long key first;
- with it on, the scan offers `might_contain` and `rlike`, `java.util.regex`
overflows the stack in the reader, and `SparkUncaughtExceptionHandler` takes
the JVM down with the task.
A Scala or Hive UDF can fail the same way, and so can any function that runs
a user's regex. The bloom alone runs no user code, so this is new with this PR.
I don't see a short list of safe expressions to allow, and a fatal error must
not be swallowed. So I'd put the key conjuncts behind their own conf, off by
default, with this risk in its doc as a third exception to "the answer does not
change". The bloom alone then stays on under the existing conf without it.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategy.scala:
##########
@@ -191,13 +194,22 @@ object FileSourceStrategy extends Strategy with
PredicateHelper with Logging {
val dataAttrs = AttributeSet(readDataColumns)
// Deduplicated through an `ExpressionSet`, as `afterScanFilters` is, once
a scan can take any.
- val offered = ExpressionSet(normalizedFilters).toSeq.filter { expr =>
+ val candidates = ExpressionSet(normalizedFilters).toSeq
+ val offered = candidates.filter { expr =>
val refs = expr.references
refs.nonEmpty && refs.subsetOf(dataAttrs) &&
fsRelation.fileFormat.supportsStorageFilter(expr)
}
if (offered.isEmpty) return Nil
val storageKeyAttrs =
AttributeSet.fromAttributeSets(offered.map(_.references))
- if (dataAttrs.subsetOf(storageKeyAttrs)) Nil else offered
+ if (dataAttrs.subsetOf(storageKeyAttrs)) return Nil
+ // A conjunct over the key columns alone adds no column to the read. An
`IsNotNull` is left
+ // out, since the post-scan `Filter` marks its column not nullable only
while it is bare there.
+ val onKeys = candidates.filter { expr =>
+ val refs = expr.references
+ refs.nonEmpty && refs.subsetOf(storageKeyAttrs) &&
!offered.contains(expr) &&
+ !expr.isInstanceOf[IsNotNull] &&
fsRelation.fileFormat.supportsStorageFilterOnKeys(expr)
Review Comment:
**Finding 3.** Keeping `isnotnull(k)` bare in the post-scan `Filter` does
not require keeping it from the reader. Null keys can pass the bloom:
- `XxHash64` of a null key is its seed, not null;
- `BloomFilterAggregate.update` skips only a null hash, so the bloom holds
the seed whenever its build side has a null key, as the preserved side of this
PR's LEFT OUTER JOIN can.
The reader then reads the other columns of every null-key row, and the
`Filter` drops them. Offering `IsNotNull` too, and leaving it out of the wrap
at L451 (`if (offered.contains(f) && !f.isInstanceOf[IsNotNull])`), would drop
them in phase 1. I traced this by reading the code.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategy.scala:
##########
@@ -191,13 +194,22 @@ object FileSourceStrategy extends Strategy with
PredicateHelper with Logging {
val dataAttrs = AttributeSet(readDataColumns)
// Deduplicated through an `ExpressionSet`, as `afterScanFilters` is, once
a scan can take any.
- val offered = ExpressionSet(normalizedFilters).toSeq.filter { expr =>
+ val candidates = ExpressionSet(normalizedFilters).toSeq
+ val offered = candidates.filter { expr =>
val refs = expr.references
refs.nonEmpty && refs.subsetOf(dataAttrs) &&
fsRelation.fileFormat.supportsStorageFilter(expr)
}
if (offered.isEmpty) return Nil
val storageKeyAttrs =
AttributeSet.fromAttributeSets(offered.map(_.references))
- if (dataAttrs.subsetOf(storageKeyAttrs)) Nil else offered
+ if (dataAttrs.subsetOf(storageKeyAttrs)) return Nil
+ // A conjunct over the key columns alone adds no column to the read. An
`IsNotNull` is left
+ // out, since the post-scan `Filter` marks its column not nullable only
while it is bare there.
+ val onKeys = candidates.filter { expr =>
+ val refs = expr.references
+ refs.nonEmpty && refs.subsetOf(storageKeyAttrs) &&
!offered.contains(expr) &&
+ !expr.isInstanceOf[IsNotNull] &&
fsRelation.fileFormat.supportsStorageFilterOnKeys(expr)
+ }
+ offered ++ onKeys
Review Comment:
**Finding 2.** The key conjuncts are offered even where the scan gets no
checked column, which is when `storageFilterCheckedColumn` (L367) comes out
empty: whole-stage codegen off, or too many fields once the column is added.
The post-scan `Filter` then keeps them bare, so it evaluates them again on
every row the reader returns. For the bloom alone that second evaluation is a
cheap probe, but a key conjunct can cost much more. Whether the column is added
depends only on the conf and the scan's output, so `storageFiltersFor` could
offer the key conjuncts only where it will be. I traced this by reading the
code.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetStorageFilter.scala:
##########
@@ -142,7 +143,15 @@ 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
Review Comment:
**Finding 5.** The scaladoc above (L135-137) still says the storage filters
are "combined with AND". With both groups present they are now `If(blooms,
conjuncts, false)`.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetStorageFilter.scala:
##########
@@ -142,7 +143,15 @@ 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])
+ val expr = if (blooms.isEmpty || onKeys.isEmpty) {
Review Comment:
**Finding 6.** With both groups present, `expr` is the `If`, so the
`require` messages at L162 and L164 now read "storage filter if has no bound
reference ..." and "storage filter if references ordinals ...". Naming
`boundExpressions.map(_.prettyName)` would keep them naming the filters.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetStorageFilterSuite.scala:
##########
@@ -2249,6 +2282,80 @@ 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. A conjunct over
+ // another column does not, nor an `IsNotNull`, which the Filter needs
bare, nor a conjunct over
+ // the key with no bloom to go with.
+ withTempDir { dir =>
+ val path = writeKeyParquetFile(dir, 0L until 50L, rowGroupSize = 64 *
1024L)
+ withSQLConf(SQLConf.PARQUET_STORAGE_FILTER_PUSHDOWN_ENABLED.key ->
"true") {
+ val relation = spark.read.parquet(path).queryExecution.optimizedPlan
+ val k = columnOf(relation, "k")
+ val bloom = bloomOn(k)
+ val onKey = EqualTo(Remainder(k, Literal(4L)), Literal(0L))
+ val onValue = EqualTo(columnOf(relation, "v"), Literal("42"))
+ def offered(condition: Expression): Seq[Expression] =
+ scanOf(planned(LogicalFilter(condition, relation))).storageFilters
+ val withBloom = offered(And(And(And(IsNotNull(k), onKey), bloom),
onValue))
+ assert(withBloom.length == 2 && withBloom.head.semanticEquals(bloom) &&
+ withBloom(1).semanticEquals(onKey), s"expected the bloom, then
$onKey; got $withBloom")
+ assert(offered(And(onKey, onValue)).isEmpty, "nothing is offered
without a bloom")
+ 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.
+ private def readKeysUnder(condition: Attribute => Expression)(
+ check: (Seq[Long], Seq[Long], Long) => Unit): Unit = withTempDir { dir =>
+ val path = writeKeyParquetFile(dir, 0L until 1000L, rowGroupSize = 1024L)
+ val relation = spark.read.parquet(path).queryExecution.analyzed
+ val filtered = LogicalFilter(condition(columnOf(relation, "k")), relation)
+ def run(pushdown: Boolean): (SparkPlan, Seq[Long]) = withSQLConf(
+ SQLConf.PARQUET_STORAGE_FILTER_PUSHDOWN_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") {
Review Comment:
**Finding 4.** The description says that a key conjunct that throws on a row
another condition would drop makes the reader fall back to a plain read of that
row group, with the same answer. No test covers it. This test shows that the
`If` keeps the conjunct off the rows the bloom rejects, not what happens when
it throws on a row the bloom keeps. For example, under ANSI, a bloom holding 0
and 42 with `v <> '0' AND 100 % k >= 0` reaches `100 % 0` in the reader, and
the rows must still equal a plain read's.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetStorageFilter.scala:
##########
@@ -216,6 +225,17 @@ object ParquetStorageFilter {
case _ => false
}
+ /**
+ * Whether the reader can evaluate `expr`, a conjunct over the key columns
of the accepted
+ * storage filters, together with them: no subquery, and nothing unevaluable
but column
+ * references.
+ */
+ def isSupportedFilterOnKeys(expr: Expression): Boolean =
+ !expr.containsPattern(PLAN_EXPRESSION) && !expr.exists {
Review Comment:
**Finding 8.** Optional, and fine as a follow-up.
`containsPattern(PLAN_EXPRESSION)` also leaves out an uncorrelated scalar
subquery, such as `k > (SELECT max(x) FROM t)`.
`FileSourceScanExec.preparedStorageFilters` already turns every
`execution.ScalarSubquery` in `storageFilters` into a literal, which is how the
bloom's own bytes reach the reader. So the rule `FileSourceStrategy` uses for
data filters at L256, `!(e.containsPattern(PLAN_EXPRESSION) &&
!e.containsPattern(SCALAR_SUBQUERY))`, would let them through.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetStorageFilterSuite.scala:
##########
@@ -558,9 +558,11 @@ class ParquetStorageFilterSuite extends QueryTest with
SharedSparkSession
// A runtime bloom's `Literal` holding one long key, built the way the
planner builds one. It is
Review Comment:
**Finding 7.** `bloomLiteralOf` takes several keys now, but its comment
still says it holds "one long key" and that "the key is hashed with the same
seed".
--
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]
