ForVic commented on code in PR #59486:
URL: https://github.com/apache/spark/pull/59486#discussion_r4356097370
##########
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:
Agreed. They're now behind their own conf,
`spark.sql.parquet.storageFilterPushdown.keyConditions.enabled`, off by
default. The bloom alone stays under
`spark.sql.parquet.storageFilterPushdown.enabled`. The new conf's doc lists the
fatal-error case as the extra exception. `supportsStorageFilterOnKeys` now
takes the session so the format can read the conf.
##########
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:
Done. `FileSourceStrategy` now works out whether the scan will get the
checked column before it asks for storage filters, and offers the key conjuncts
only when it will. Without the column, only the bloom is offered. The planner
test covers this with whole-stage codegen off.
##########
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:
Good catch. `IsNotNull` over the key is offered now, and the post-scan
Filter keeps it bare (it's left out of the `If(checked, true, f)` wrap), so
nullability is unchanged. I added a test with a bloom built over a null key and
every odd key null. The reader drops all 999 non-matching rows, null keys
included.
##########
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:
Added `a conjunct over the key columns that throws on a row the bloom keeps
gives the row group up`, using your example: under ANSI, a bloom holding 0 and
42 with `v <> '0' AND 100 % k >= 0`. It returns the same rows as a plain read,
and the reader excludes fewer rows than it would without the fallback.
##########
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:
Fixed, thanks.
--
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]
