This is an automated email from the ASF dual-hosted git repository.
zml1206 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/gluten.git
The following commit(s) were added to refs/heads/main by this push:
new a6202db285 [VL] Fix GlutenSubquerySuite in Spark 4.x (#12473)
a6202db285 is described below
commit a6202db2856fff1952fc1fc0d62df5edfa1b3168
Author: Mingliang Zhu <[email protected]>
AuthorDate: Wed Jul 8 16:15:58 2026 +0800
[VL] Fix GlutenSubquerySuite in Spark 4.x (#12473)
---
.../gluten/utils/velox/VeloxTestSettings.scala | 8 +-
.../org/apache/spark/sql/GlutenSubquerySuite.scala | 86 +++++++++++++++++++---
.../gluten/utils/velox/VeloxTestSettings.scala | 12 +--
.../org/apache/spark/sql/GlutenSubquerySuite.scala | 86 +++++++++++++++++++---
4 files changed, 152 insertions(+), 40 deletions(-)
diff --git
a/gluten-ut/spark40/src/test/scala/org/apache/gluten/utils/velox/VeloxTestSettings.scala
b/gluten-ut/spark40/src/test/scala/org/apache/gluten/utils/velox/VeloxTestSettings.scala
index f906b14455..a8d5cf2135 100644
---
a/gluten-ut/spark40/src/test/scala/org/apache/gluten/utils/velox/VeloxTestSettings.scala
+++
b/gluten-ut/spark40/src/test/scala/org/apache/gluten/utils/velox/VeloxTestSettings.scala
@@ -1099,14 +1099,10 @@ class VeloxTestSettings extends BackendTestSettings {
.exclude("column stats collection for null columns")
.exclude("analyze column command - result verification")
enableSuite[GlutenSubquerySuite]
- .excludeByPrefix(
- "SPARK-26893" // Rewrite this test because it checks Spark's physical
operators.
- )
- // exclude as it checks spark plan
+ // Rewrite as it checks spark plan.
+ .excludeByPrefix("SPARK-26893")
.exclude("SPARK-36280: Remove redundant aliases after
RewritePredicateSubquery")
- // TODO: fix in Spark-4.0
.excludeByPrefix("SPARK-51738")
- .excludeByPrefix("SPARK-43402")
.exclude("non-aggregated correlated scalar subquery")
.exclude("SPARK-18504 extra GROUP BY column in correlated scalar subquery
is not permitted")
.exclude("SPARK-43402: FileSourceScanExec supports push down data filter
with scalar subquery")
diff --git
a/gluten-ut/spark40/src/test/scala/org/apache/spark/sql/GlutenSubquerySuite.scala
b/gluten-ut/spark40/src/test/scala/org/apache/spark/sql/GlutenSubquerySuite.scala
index 05dae62902..01ee4cef09 100644
---
a/gluten-ut/spark40/src/test/scala/org/apache/spark/sql/GlutenSubquerySuite.scala
+++
b/gluten-ut/spark40/src/test/scala/org/apache/spark/sql/GlutenSubquerySuite.scala
@@ -16,10 +16,15 @@
*/
package org.apache.spark.sql
-import org.apache.gluten.execution.{FileSourceScanExecTransformer,
WholeStageTransformer}
+import org.apache.gluten.execution.FileSourceScanExecTransformer
+
+import org.apache.spark.sql.execution.{ColumnarShuffleExchangeExec,
ExecSubqueryExpression, ReusedSubqueryExec, ScalarSubquery}
+import org.apache.spark.sql.internal.SQLConf
class GlutenSubquerySuite extends SubquerySuite with GlutenSQLTestsTrait {
+ import testImplicits._
+
// Test Canceled: IntegratedUDFTestUtils.shouldTestPythonUDFs was false
override def testNameBlackList: Seq[String] = Seq(
"SPARK-28441: COUNT bug in WHERE clause (Filter) with PythonUDF",
@@ -32,24 +37,81 @@ class GlutenSubquerySuite extends SubquerySuite with
GlutenSQLTestsTrait {
"SPARK-28441: COUNT bug with attribute ref in subquery input and output
with PythonUDF"
)
- // === Following cases override super class's cases ===
- // TODO: fix in Spark-4.0
- ignoreGluten("SPARK-26893 Allow pushdown of partition pruning subquery
filters to file source") {
+ testGluten("SPARK-26893: Allow pushdown of partition pruning subquery
filters to file source") {
withTable("a", "b") {
spark.range(4).selectExpr("id", "id % 2 AS
p").write.partitionBy("p").saveAsTable("a")
spark.range(2).write.saveAsTable("b")
- // need to execute the query before we can examine fs.inputRDDs()
val df = sql("SELECT * FROM a WHERE p <= (SELECT MIN(id) FROM b)")
checkAnswer(df, Seq(Row(0, 0), Row(2, 0)))
- assert(stripAQEPlan(df.queryExecution.executedPlan).collectFirst {
- case t: WholeStageTransformer => t
- } match {
- case Some(WholeStageTransformer(fs: FileSourceScanExecTransformer, _))
=>
- fs.dynamicallySelectedPartitions.toPartitionArray
- .exists(_.filePath.toString.contains("p=0"))
- case _ => false
+ // need to execute the query before we can examine fs.inputRDDs()
+ val fileSourceScanExec =
collect(stripAQEPlan(df.queryExecution.executedPlan)) {
+ case fs: FileSourceScanExecTransformer => fs
+ }
+ assert(fileSourceScanExec.size === 1)
+ val scan = fileSourceScanExec.head
+ assert(scan.partitionFilters.exists(ExecSubqueryExpression.hasSubquery))
+ val selectedPartitions =
scan.dynamicallySelectedPartitions.toPartitionArray
+ assert(selectedPartitions.nonEmpty)
+ assert(selectedPartitions.forall(_.filePath.toString.contains("p=0")))
+ }
+ }
+
+ testGluten("SPARK-36280: Remove redundant aliases after
RewritePredicateSubquery") {
+ withTable("t1", "t2") {
+ sql("CREATE TABLE t1 USING parquet AS SELECT id AS a, id AS b, id AS c
FROM range(10)")
+ sql("CREATE TABLE t2 USING parquet AS SELECT id AS x, id AS y FROM
range(8)")
+ val df = sql(
+ """
+ |SELECT *
+ |FROM t1
+ |WHERE a IN (SELECT x
+ | FROM (SELECT x AS x,
+ | RANK() OVER (PARTITION BY x ORDER BY
SUM(y) DESC) AS ranking
+ | FROM t2
+ | GROUP BY x) tmp1
+ | WHERE ranking <= 5)
+ |""".stripMargin)
+
+ df.collect()
+ val exchanges = collect(df.queryExecution.executedPlan) {
+ case s: ColumnarShuffleExchangeExec => s
+ }
+ assert(exchanges.size === 1)
+ }
+ }
+
+ testGluten(
+ "SPARK-43402: FileSourceScanExec supports push down data filter with
scalar subquery") {
+ def checkFileSourceScan(query: String, answer: Seq[Row]): Unit = {
+ val df = sql(query)
+ checkAnswer(df, answer)
+ val fileSourceScanExec = collect(df.queryExecution.executedPlan) {
+ case f: FileSourceScanExecTransformer => f
+ }
+ sparkContext.listenerBus.waitUntilEmpty()
+ assert(fileSourceScanExec.size === 1)
+ val scalarSubquery =
fileSourceScanExec.head.dataFilters.flatMap(_.collect {
+ case s: ScalarSubquery => s
})
+ assert(scalarSubquery.length === 1)
+ assert(scalarSubquery.head.plan.isInstanceOf[ReusedSubqueryExec])
+ 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")
+
+ checkFileSourceScan(
+ "SELECT * FROM t1 WHERE c1 > (SELECT min(c2) FROM t2)",
+ Seq.empty)
+ checkFileSourceScan(
+ "SELECT * FROM t1 WHERE c1 < (SELECT min(c2) FROM t2)",
+ Row(1) :: Row(2) :: Row(3) :: Nil)
+ }
}
}
}
diff --git
a/gluten-ut/spark41/src/test/scala/org/apache/gluten/utils/velox/VeloxTestSettings.scala
b/gluten-ut/spark41/src/test/scala/org/apache/gluten/utils/velox/VeloxTestSettings.scala
index e13d98ba98..045df1600d 100644
---
a/gluten-ut/spark41/src/test/scala/org/apache/gluten/utils/velox/VeloxTestSettings.scala
+++
b/gluten-ut/spark41/src/test/scala/org/apache/gluten/utils/velox/VeloxTestSettings.scala
@@ -1100,18 +1100,10 @@ class VeloxTestSettings extends BackendTestSettings {
enableSuite[GlutenSTFunctionsSuite]
enableSuite[GlutenStringLiteralCoalescingSuite]
enableSuite[GlutenSubquerySuite]
- .excludeByPrefix(
- "SPARK-26893" // Rewrite this test because it checks Spark's physical
operators.
- )
- // exclude as it checks spark plan
+ // Rewrite as it checks spark plan.
+ .excludeByPrefix("SPARK-26893")
.exclude("SPARK-36280: Remove redundant aliases after
RewritePredicateSubquery")
- // TODO: fix in Spark-4.0
- .excludeByPrefix("SPARK-51738")
- .excludeByPrefix("SPARK-43402")
- .exclude("non-aggregated correlated scalar subquery")
- .exclude("SPARK-18504 extra GROUP BY column in correlated scalar subquery
is not permitted")
.exclude("SPARK-43402: FileSourceScanExec supports push down data filter
with scalar subquery")
- .exclude("SPARK-51738: IN subquery with struct type")
enableSuite[GlutenTypedImperativeAggregateSuite]
enableSuite[GlutenUnwrapCastInComparisonEndToEndSuite]
enableSuite[GlutenUnsafeRowChecksumSuite]
diff --git
a/gluten-ut/spark41/src/test/scala/org/apache/spark/sql/GlutenSubquerySuite.scala
b/gluten-ut/spark41/src/test/scala/org/apache/spark/sql/GlutenSubquerySuite.scala
index 05dae62902..01ee4cef09 100644
---
a/gluten-ut/spark41/src/test/scala/org/apache/spark/sql/GlutenSubquerySuite.scala
+++
b/gluten-ut/spark41/src/test/scala/org/apache/spark/sql/GlutenSubquerySuite.scala
@@ -16,10 +16,15 @@
*/
package org.apache.spark.sql
-import org.apache.gluten.execution.{FileSourceScanExecTransformer,
WholeStageTransformer}
+import org.apache.gluten.execution.FileSourceScanExecTransformer
+
+import org.apache.spark.sql.execution.{ColumnarShuffleExchangeExec,
ExecSubqueryExpression, ReusedSubqueryExec, ScalarSubquery}
+import org.apache.spark.sql.internal.SQLConf
class GlutenSubquerySuite extends SubquerySuite with GlutenSQLTestsTrait {
+ import testImplicits._
+
// Test Canceled: IntegratedUDFTestUtils.shouldTestPythonUDFs was false
override def testNameBlackList: Seq[String] = Seq(
"SPARK-28441: COUNT bug in WHERE clause (Filter) with PythonUDF",
@@ -32,24 +37,81 @@ class GlutenSubquerySuite extends SubquerySuite with
GlutenSQLTestsTrait {
"SPARK-28441: COUNT bug with attribute ref in subquery input and output
with PythonUDF"
)
- // === Following cases override super class's cases ===
- // TODO: fix in Spark-4.0
- ignoreGluten("SPARK-26893 Allow pushdown of partition pruning subquery
filters to file source") {
+ testGluten("SPARK-26893: Allow pushdown of partition pruning subquery
filters to file source") {
withTable("a", "b") {
spark.range(4).selectExpr("id", "id % 2 AS
p").write.partitionBy("p").saveAsTable("a")
spark.range(2).write.saveAsTable("b")
- // need to execute the query before we can examine fs.inputRDDs()
val df = sql("SELECT * FROM a WHERE p <= (SELECT MIN(id) FROM b)")
checkAnswer(df, Seq(Row(0, 0), Row(2, 0)))
- assert(stripAQEPlan(df.queryExecution.executedPlan).collectFirst {
- case t: WholeStageTransformer => t
- } match {
- case Some(WholeStageTransformer(fs: FileSourceScanExecTransformer, _))
=>
- fs.dynamicallySelectedPartitions.toPartitionArray
- .exists(_.filePath.toString.contains("p=0"))
- case _ => false
+ // need to execute the query before we can examine fs.inputRDDs()
+ val fileSourceScanExec =
collect(stripAQEPlan(df.queryExecution.executedPlan)) {
+ case fs: FileSourceScanExecTransformer => fs
+ }
+ assert(fileSourceScanExec.size === 1)
+ val scan = fileSourceScanExec.head
+ assert(scan.partitionFilters.exists(ExecSubqueryExpression.hasSubquery))
+ val selectedPartitions =
scan.dynamicallySelectedPartitions.toPartitionArray
+ assert(selectedPartitions.nonEmpty)
+ assert(selectedPartitions.forall(_.filePath.toString.contains("p=0")))
+ }
+ }
+
+ testGluten("SPARK-36280: Remove redundant aliases after
RewritePredicateSubquery") {
+ withTable("t1", "t2") {
+ sql("CREATE TABLE t1 USING parquet AS SELECT id AS a, id AS b, id AS c
FROM range(10)")
+ sql("CREATE TABLE t2 USING parquet AS SELECT id AS x, id AS y FROM
range(8)")
+ val df = sql(
+ """
+ |SELECT *
+ |FROM t1
+ |WHERE a IN (SELECT x
+ | FROM (SELECT x AS x,
+ | RANK() OVER (PARTITION BY x ORDER BY
SUM(y) DESC) AS ranking
+ | FROM t2
+ | GROUP BY x) tmp1
+ | WHERE ranking <= 5)
+ |""".stripMargin)
+
+ df.collect()
+ val exchanges = collect(df.queryExecution.executedPlan) {
+ case s: ColumnarShuffleExchangeExec => s
+ }
+ assert(exchanges.size === 1)
+ }
+ }
+
+ testGluten(
+ "SPARK-43402: FileSourceScanExec supports push down data filter with
scalar subquery") {
+ def checkFileSourceScan(query: String, answer: Seq[Row]): Unit = {
+ val df = sql(query)
+ checkAnswer(df, answer)
+ val fileSourceScanExec = collect(df.queryExecution.executedPlan) {
+ case f: FileSourceScanExecTransformer => f
+ }
+ sparkContext.listenerBus.waitUntilEmpty()
+ assert(fileSourceScanExec.size === 1)
+ val scalarSubquery =
fileSourceScanExec.head.dataFilters.flatMap(_.collect {
+ case s: ScalarSubquery => s
})
+ assert(scalarSubquery.length === 1)
+ assert(scalarSubquery.head.plan.isInstanceOf[ReusedSubqueryExec])
+ 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")
+
+ checkFileSourceScan(
+ "SELECT * FROM t1 WHERE c1 > (SELECT min(c2) FROM t2)",
+ Seq.empty)
+ checkFileSourceScan(
+ "SELECT * FROM t1 WHERE c1 < (SELECT min(c2) FROM t2)",
+ Row(1) :: Row(2) :: Row(3) :: Nil)
+ }
}
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
