dongjoon-hyun commented on code in PR #59810:
URL: https://github.com/apache/spark/pull/59810#discussion_r4389456109
##########
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:
Thank you, @peter-toth. I rebased this PR on #59818 and added the merged
case to the `SPARK-60577` test as you suggested, in a3a57dd75a1. It passes for
`DATE`, `TIMESTAMP` and `DECIMAL`. On master, without the main change of this
PR, the merged case fails with `3 did not equal 0`.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/DataSourceScanExec.scala:
##########
@@ -548,9 +547,14 @@ trait FileSourceScanLike extends DataSourceScanExec with
SessionStateHelper {
location.getClass.getSimpleName +
Utils.buildLocationMetadata(location.rootPaths, maxMetadataValueLength)
// `metadata` is accessed during planning and the scalar subquery is not
executed yet. Here
- // we get the pretty string of the scalar subquery, for display purpose
only.
- val pushedFiltersForDisplay = translateToV1Filters(
- dataFilters, s => Literal("ScalarSubquery#" + s.exprId.id))
+ // we get the pretty string of the scalar subquery, for display purpose
only. A field read out
+ // of a subquery that `MergeSubplans` merged with others gets the string
as a whole, since the
+ // string has no field to read.
+ val pushedFiltersForDisplay = translateToV1Filters(dataFilters,
_.transform {
+ case GetStructField(s: execution.ScalarSubquery, _, _) =>
Review Comment:
Thank you. I confirmed it by running it. On the previous head, an `IN` over
merged scalar subqueries on `DATE`, `TIMESTAMP`, `TIMESTAMP_NTZ` and `DECIMAL`
columns failed with `ClassCastException`, and the unmerged form already failed
the same way since v4.0.0. #59818 (SPARK-60577) leaves a filter with an `IN`
over scalar subqueries out of `PushedFilters`, which covers the merged form
too, and this PR is now rebased on it. The `SPARK-60577` test also runs the
merged form under the default configuration now (a3a57dd75a1).
--
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]
