peter-toth commented on code in PR #59810:
URL: https://github.com/apache/spark/pull/59810#discussion_r4388280187
##########
sql/core/src/test/scala/org/apache/spark/sql/SubquerySuite.scala:
##########
@@ -2510,6 +2510,51 @@ class SubquerySuite extends SharedSparkSession
}
}
+ test("SPARK-60569: FileSourceScanExec pushes down data filters with merged
scalar subqueries") {
+ def checkFileSourceScan(query: String, answer: Seq[Row]): Unit = {
+ val df = sql(query)
+ checkAnswer(df, answer)
+ val fileSourceScanExec = collect(df.queryExecution.executedPlan) {
+ case f: FileSourceScanExec => f
+ }
+ sparkContext.listenerBus.waitUntilEmpty()
+ assert(fileSourceScanExec.size === 1)
+ // `MergeSubplans` merged the two subqueries into one, and each filter
reads its result out
+ // of the struct the merged one returns.
+ val mergedSubqueryResults =
fileSourceScanExec.head.dataFilters.flatMap(_.collect {
+ case g @ GetStructField(_: ScalarSubquery, _, _) => g
+ })
+ assert(mergedSubqueryResults.length === 2)
+ val pushedFilters = fileSourceScanExec.head.metadata("PushedFilters")
+ assert(pushedFilters.contains("GreaterThan(c1,ScalarSubquery#"))
+ assert(pushedFilters.contains("LessThan(c1,ScalarSubquery#"))
+ assert(fileSourceScanExec.head.metrics("numFiles").value === 1)
+ assert(fileSourceScanExec.head.metrics("numOutputRows").value ===
answer.size)
+ }
+
+ withTable("t1", "t2") {
+ withSQLConf(SQLConf.LEAF_NODE_DEFAULT_PARALLELISM.key -> "1") {
+ Seq(1, 2, 3).toDF("c1").write.format("parquet").saveAsTable("t1")
+ Seq(4, 5, 6).toDF("c2").write.format("parquet").saveAsTable("t2")
+
+ // The row group is skipped by `c1 > 4`.
+ checkFileSourceScan(
+ "SELECT * FROM t1 WHERE c1 > (SELECT min(c2) FROM t2) AND c1 <
(SELECT max(c2) FROM t2)",
Review Comment:
**Finding 1.** The `GetStructField` case in `metadata` and
`replaceWithLiterals` both change what an `IN` over merged subqueries does.
These tests only use comparisons on an `INT` column, which is why CI missed the
display regression at
[r4383934071](https://github.com/apache/spark/pull/59810#discussion_r4383934071).
After the rebase on #59818, its `SPARK-60577` test runs the two-subquery `IN`
only with `reuseSubquery=false`, so the merged form stays untested there too.
Running that query under the default conf as well covers both changes:
```scala
val twoSubqueries = s"SELECT * FROM t1 WHERE $column IN " +
s"((SELECT min($column) FROM t2), (SELECT max($column) FROM t2))"
// `MergeSubplans` merges the two subqueries into one, unless
subquery reuse is off.
checkFileSourceScan(twoSubqueries, 2)
withSQLConf(SQLConf.SUBQUERY_REUSE_ENABLED.key -> "false") {
checkFileSourceScan(twoSubqueries, 2)
}
```
I ran it on this PR with #59818 applied, and it passes for `DATE`,
`TIMESTAMP` and `DECIMAL`. With #59818 alone, the merged arm fails with `3 did
not equal 0`, since the `IN` is not pushed. With this PR alone, the merged
query throws `ClassCastException`.
--
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]
