ForVic commented on code in PR #59486:
URL: https://github.com/apache/spark/pull/59486#discussion_r4356099227
##########
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:
Added `every conjunct over the key columns is enforced, not only the first`,
with your `k % 3 <> 0` / `k % 5 <> 0` over a bloom of 3, 5 and 7.
--
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]
