ForVic commented on code in PR #59486:
URL: https://github.com/apache/spark/pull/59486#discussion_r4356098200
##########
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:
Fixed. The messages now list `boundExpressions.map(_.prettyName)`.
##########
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:
Fixed, thanks.
##########
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:
Makes sense, thanks. I'd like to keep this PR focused, so I'll do it as a
follow-up and file a JIRA for it.
##########
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:
Fixed, thanks.
##########
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:
Updated the `buildReaderWithStorageFilters` doc. It now says the filters
`supportsStorageFilter` accepted come first and the ones
`supportsStorageFilterOnKeys` accepted come second, with none in both. Since
the planner doesn't mark where the first group ends, a format that takes the
second group must accept a first-group shape it can tell apart, the way Parquet
accepts only a bare `BloomFilterMightContain` there.
##########
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:
Added `k + CAST(v AS BIGINT) > 0` to the planner test. It asserts that
conjunct isn't offered, while `k % 4 = 0` still follows 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]
