(spark) branch branch-4.x updated: [SPARK-58555][SQL] Add new tests for JDBC DS Join pushdown

[we...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

cloud-fan pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git


The following commit(s) were added to refs/heads/branch-4.x by this push:
     new 0abc99f31db9 [SPARK-58555][SQL] Add new tests for JDBC DS Join pushdown
0abc99f31db9 is described below

commit 0abc99f31db93c256cd18ac2c1a180ea3b3565db
Author: Uros Stankovic <ur...@databricks.com>
AuthorDate: Mon Sep 21 05:09:30 2026 +0000

    [SPARK-58555][SQL] Add new tests for JDBC DS Join pushdown
    
    ### What changes were proposed in this pull request?
    - Add two new tests in suite that verifies behaviour of JDBC Data Source join pushdown
    
    ### Why are the changes needed?
    - Optimizer rules may be reordered or new rule may be added, so join on top of aggregates may not be pushed down anymore. In this way, we make feature coverage more robust.
    
    ### Does this PR introduce _any_ user-facing change?
    No
    
    ### How was this patch tested?
    Test addtion/NA
    
    ### Was this patch authored or co-authored using generative AI tooling?
    No
    
    Closes #57714 from urosstan-db/urosstan-db/additional-join-tests.
    
    Authored-by: Uros Stankovic <ur...@databricks.com>
    Signed-off-by: Wenchen Fan <we...@databricks.com>
    (cherry picked from commit a8c3675ebc726c2f9c18100f5ba06d61108df176)
    Signed-off-by: Wenchen Fan <we...@databricks.com>
---
 .../JDBCV2JoinPushdownIntegrationSuiteBase.scala   | 51 +++++++++++++++++++++-
 1 file changed, 49 insertions(+), 2 deletions(-)

diff --git a/sql/core/src/test/scala/org/apache/spark/sql/jdbc/v2/JDBCV2JoinPushdownIntegrationSuiteBase.scala b/sql/core/src/test/scala/org/apache/spark/sql/jdbc/v2/JDBCV2JoinPushdownIntegrationSuiteBase.scala
index 4c46f5587a3f..617370a55175 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/jdbc/v2/JDBCV2JoinPushdownIntegrationSuiteBase.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/jdbc/v2/JDBCV2JoinPushdownIntegrationSuiteBase.scala
@@ -229,8 +229,6 @@ trait JDBCV2JoinPushdownIntegrationSuiteBase
 
   protected val supportsColumnPruning: Boolean = true
 
-  protected val supportsJoinPushdown: Boolean = true
-
   // Condition-less joins are not supported in join pushdown
   test("Test that 2-way join without condition should not have join pushed down") {
     val sqlQuery =
@@ -497,6 +495,55 @@ trait JDBCV2JoinPushdownIntegrationSuiteBase
     }
   }
 
+  test("Test aggregate with group by on top of join") {
+    val sqlQuery =
+      s"""
+         |SELECT t1.id, t1.address, min(t2.salary)
+         |FROM $catalogAndNamespace.$casedJoinTableName1 t1
+         |JOIN $catalogAndNamespace.$casedJoinTableName2 t2 ON t1.id = t2.id
+         |WHERE t1.amount > 1000
+         |GROUP BY t1.id, t1.address
+         |""".stripMargin
+
+    val rowsNoPushdown = withSQLConf(SQLConf.DATA_SOURCE_V2_JOIN_PUSHDOWN.key -> "false") {
+      sql(sqlQuery).collect().toSeq
+    }
+
+    assert(rowsNoPushdown.nonEmpty)
+
+    withSQLConf(SQLConf.DATA_SOURCE_V2_JOIN_PUSHDOWN.key -> "true") {
+      val df = sql(sqlQuery)
+      checkJoinPushed(df)
+      checkAggregateRemoved(df, supportsAggregatePushdown)
+      checkAnswer(df, rowsNoPushdown)
+    }
+  }
+
+  test("Test multi-way join with function in join condition") {
+    val sqlQuery =
+      s"""
+         |SELECT a.id, c.address, d.address, a.amount
+         |FROM $catalogAndNamespace.$casedJoinTableName1 a
+         |JOIN $catalogAndNamespace.$casedJoinTableName1 c
+         |  ON a.address = c.address
+         |JOIN $catalogAndNamespace.$casedJoinTableName1 d
+         |  ON IF(a.id > 5, a.id - 1, a.id + 1) = d.id
+         |WHERE a.amount >= 1000
+         |""".stripMargin
+
+    val rowsNoPushdown = withSQLConf(SQLConf.DATA_SOURCE_V2_JOIN_PUSHDOWN.key -> "false") {
+      sql(sqlQuery).collect().toSeq
+    }
+
+    assert(rowsNoPushdown.nonEmpty)
+
+    withSQLConf(SQLConf.DATA_SOURCE_V2_JOIN_PUSHDOWN.key -> "true") {
+      val df = sql(sqlQuery)
+      checkJoinPushed(df)
+      checkAnswer(df, rowsNoPushdown)
+    }
+  }
+
   test("Test sort limit on top of join is pushed down") {
     val sqlQuery = s"""
       |SELECT min(a.id + b.id), a.id, b.id


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org