From 334f14c306a777ea319095c7aae6895d613da227 Mon Sep 17 00:00:00 2001 From: Uros Stankovic Date: Mon, 3 Aug 2026 10:55:29 +0200 Subject: [PATCH 1/2] Add new tests --- ...BCV2JoinPushdownIntegrationSuiteBase.scala | 55 +++++++++++++++++++ 1 file changed, 55 insertions(+) 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 4c46f5587a3f1..cef59a53902ad 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 @@ -497,6 +497,61 @@ trait JDBCV2JoinPushdownIntegrationSuiteBase } } + test("Test aggregate with group by on top of join") { + val sqlQuery = + s""" + |SELECT t1.id, t1.address, min(t2.salary), count(1) + |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 rows = withSQLConf(SQLConf.DATA_SOURCE_V2_JOIN_PUSHDOWN.key -> "false") { + sql(sqlQuery).collect().toSeq + } + + assert(rows.nonEmpty) + + withSQLConf(SQLConf.DATA_SOURCE_V2_JOIN_PUSHDOWN.key -> "true") { + val df = sql(sqlQuery) + + if (supportsJoinPushdown) { + checkJoinPushed(df) + checkAggregateRemoved(df, supportsAggregatePushdown) + } + checkAnswer(df, rows) + } + } + + test("Test multi-way left outer join with function in join condition") { + val sqlQuery = + s""" + |SELECT a.id, c.address, d.address, a.amount + |FROM $catalogAndNamespace.$casedJoinTableName1 a + |LEFT JOIN $catalogAndNamespace.$casedJoinTableName1 c + | ON a.address = c.address + |LEFT JOIN $catalogAndNamespace.$casedJoinTableName1 d + | ON LOWER(a.address) = LOWER(d.address) + |WHERE a.amount >= 1000 + |""".stripMargin + + val rows = withSQLConf(SQLConf.DATA_SOURCE_V2_JOIN_PUSHDOWN.key -> "false") { + sql(sqlQuery).collect().toSeq + } + + assert(rows.nonEmpty) + + withSQLConf(SQLConf.DATA_SOURCE_V2_JOIN_PUSHDOWN.key -> "true") { + val df = sql(sqlQuery) + + if (supportsJoinPushdown) { + checkJoinPushed(df) + } + checkAnswer(df, rows) + } + } + test("Test sort limit on top of join is pushed down") { val sqlQuery = s""" |SELECT min(a.id + b.id), a.id, b.id From 6d8289751a8c4e90a5d229bad4cb1e5ed0cf7446 Mon Sep 17 00:00:00 2001 From: Uros Stankovic Date: Mon, 3 Aug 2026 11:07:02 +0200 Subject: [PATCH 2/2] Simplify --- ...BCV2JoinPushdownIntegrationSuiteBase.scala | 32 +++++++------------ 1 file changed, 12 insertions(+), 20 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 cef59a53902ad..3b36a3e73de21 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 = @@ -507,48 +505,42 @@ trait JDBCV2JoinPushdownIntegrationSuiteBase |GROUP BY t1.id, t1.address |""".stripMargin - val rows = withSQLConf(SQLConf.DATA_SOURCE_V2_JOIN_PUSHDOWN.key -> "false") { + val rowsNoPushdown = withSQLConf(SQLConf.DATA_SOURCE_V2_JOIN_PUSHDOWN.key -> "false") { sql(sqlQuery).collect().toSeq } - assert(rows.nonEmpty) + assert(rowsNoPushdown.nonEmpty) withSQLConf(SQLConf.DATA_SOURCE_V2_JOIN_PUSHDOWN.key -> "true") { val df = sql(sqlQuery) - - if (supportsJoinPushdown) { - checkJoinPushed(df) - checkAggregateRemoved(df, supportsAggregatePushdown) - } - checkAnswer(df, rows) + checkJoinPushed(df) + checkAggregateRemoved(df, supportsAggregatePushdown) + checkAnswer(df, rowsNoPushdown) } } - test("Test multi-way left outer join with function in join condition") { + 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 - |LEFT JOIN $catalogAndNamespace.$casedJoinTableName1 c + |JOIN $catalogAndNamespace.$casedJoinTableName1 c | ON a.address = c.address - |LEFT JOIN $catalogAndNamespace.$casedJoinTableName1 d + |JOIN $catalogAndNamespace.$casedJoinTableName1 d | ON LOWER(a.address) = LOWER(d.address) |WHERE a.amount >= 1000 |""".stripMargin - val rows = withSQLConf(SQLConf.DATA_SOURCE_V2_JOIN_PUSHDOWN.key -> "false") { + val rowsNoPushdown = withSQLConf(SQLConf.DATA_SOURCE_V2_JOIN_PUSHDOWN.key -> "false") { sql(sqlQuery).collect().toSeq } - assert(rows.nonEmpty) + assert(rowsNoPushdown.nonEmpty) withSQLConf(SQLConf.DATA_SOURCE_V2_JOIN_PUSHDOWN.key -> "true") { val df = sql(sqlQuery) - - if (supportsJoinPushdown) { - checkJoinPushed(df) - } - checkAnswer(df, rows) + checkJoinPushed(df) + checkAnswer(df, rowsNoPushdown) } }