cloud-fan commented on code in PR #59486:
URL: https://github.com/apache/spark/pull/59486#discussion_r4348266861
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileFormat.scala:
##########
@@ -229,6 +229,14 @@ trait FileFormat {
*/
def supportsStorageFilter(expr: Expression): Boolean = false
+ /**
+ * Whether this format's reader can evaluate `expr` along with the storage
filters
+ * [[supportsStorageFilter]] accepted. The planner asks only about a
conjunct that references
+ * nothing but their key columns, and offers it after them. A reader should
evaluate it only on
Review Comment:
**Nit (P3):** This asks a reader to evaluate the key conjuncts only on the
rows the accepted storage filters keep. But `buildReaderWithStorageFilters`
gets them in the same flat `storageFilters` list (`offered ++ onKeys` in
`FileSourceStrategy.storageFiltersFor`). Its doc still says only that
`storageFilters` arrive bound while `supportsStorageFilter` "is asked about the
unbound expression", which is no longer true of the key conjuncts. Nothing
tells a reader where the accepted filters end. `ParquetStorageFilter.create`
recovers the split with `partition(_.isInstanceOf[BloomFilterMightContain])`,
which is exact only because Parquet's `supportsStorageFilter` accepts nothing
but a bare bloom.
Rows are unaffected either way, since `If(b, r, false)` and `And(b, r)` keep
the same rows, so this is only about keeping the internal contract followable.
Could the `buildReaderWithStorageFilters` doc say that the list ends with the
conjuncts `supportsStorageFilterOnKeys` accepted, and give a rule for telling
the two groups apart on the bound expressions, as Parquet does by expression
type?
**Verification:**
- **Inspection:** The FileFormat documentation for the storage-filter reader
entry point describes both groups, their order, and a rule that identifies each
one on the bound expressions, consistent with the planner never offering a
conjunct in both groups and with how the Parquet reader separates blooms from
key conjuncts.
##########
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
// serialized as `BloomFilterAggregate` serializes it, and the key is hashed
with the same seed.
- private def bloomLiteralOf(key: Long): Literal = {
+ private def bloomLiteralOf(keys: Long*): Literal = {
Review Comment:
**Nit (P3):** The helper now takes `keys: Long*` and puts every key into the
bloom (the new narrowing test passes four), but this comment still says the
literal holds one long key. Suggested wording:
```suggestion
// A runtime bloom's `Literal` holding the given long keys, built the way
the planner builds one.
// Serialized as `BloomFilterAggregate` serializes it, and each key is
hashed with the same seed.
private def bloomLiteralOf(keys: Long*): Literal = {
```
**Verification:**
- **Inspection:** The helper's comment states that the bloom holds all of
the given keys, each hashed with the planner's seed, and no longer says that it
holds one key.
##########
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") {
+ // 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))),
Review Comment:
**Nit (P3):** This is the only test that offers two key conjuncts. But
`bloomOn(k)` holds only 42 and, with 9 hash functions over 128 bits,
effectively keeps no other key in `0 until 1000`, and `100 % 42` satisfies both
conjuncts. Neither conjunct ever rejects a bloom-kept row. So replacing
`onKeys.reduce(And)` in `ParquetStorageFilter.create` with `onKeys.head` or
`reduce(Or)` would still pass every test, although the post-scan Filter skips
every offered conjunct on checked rows and would then return wrong rows.
Range bounds such as `k > lo AND k < hi` would not make a reliable test,
because they are also pushed as Parquet data filters and row-group pruning
could hide the rows in question. Two non-pushed conjuncts would work, for
example `k % 3 <> 0` and `k % 5 <> 0` with a bloom over 3, 5 and 7, where each
conjunct rejects a key the other keeps.
**Verification:**
- **Regression:** With two or more key conjuncts that pushed data filters do
not prune, each rejecting some bloom-kept row the others keep, the feature-on
read returns the same rows as the feature-off read, and combining the key
conjuncts with anything weaker than AND, such as keeping only the first or
using OR, makes the test fail.
##########
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))
Review Comment:
**Nit (P3):** The only negative case for the key-column rule here is
`onValue`, which references `v` and no key column. Suppose
`refs.subsetOf(storageKeyAttrs)` in `storageFiltersFor` were loosened to an
overlap check. This test and the rest of the suite would still pass, and a
conjunct such as `k + CAST(v AS BIGINT) > 0` would be offered.
`ParquetStorageFilter.create` would then bind `v` as a key column and decode it
in phase 1 for every row, or fail its key-type `require` for a complex column.
Adding a mixed key/non-key conjunct to this condition and asserting that it is
not offered, while `onKey` still follows the bloom, would pin the rule.
**Verification:**
- **Regression:** With a bloom on the key, a conjunct over the key and a
non-key column is absent from the scan's storage filters, while the key-only
conjunct is still offered after the bloom.
--
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]
