[SPARK-59030][SQL] Support DSv2 Join pushdown for DB2 connector

connector/docker-integration-tests/src/test/scala/org/apache/spark/sql/jdbc/v2/join/DB2JoinPushdownIntegrationSuite.scala

@@ -0,0 +1,56 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.jdbc.v2.join
+
+import java.sql.Connection
+import java.util.Locale
+
+import org.apache.spark.sql.jdbc.{DB2DatabaseOnDocker, DB2Dialect, DockerJDBCIntegrationSuite, JdbcDialect}
+import org.apache.spark.sql.jdbc.v2.JDBCV2JoinPushdownIntegrationSuiteBase
+import org.apache.spark.tags.DockerTest
+
+/**
+ * To run this test suite for a specific version (e.g., icr.io/db2_community/db2:11.5.9.0):
+ * {{{
+ *   ENABLE_DOCKER_INTEGRATION_TESTS=1 DB2_DOCKER_IMAGE_NAME=icr.io/db2_community/db2:11.5.9.0
+ *     ./build/sbt -Pdocker-integration-tests
+ *     "testOnly org.apache.spark.sql.jdbc.v2.join.DB2JoinPushdownIntegrationSuite"
+ * }}}
+ */
+@DockerTest
+class DB2JoinPushdownIntegrationSuite
+  extends DockerJDBCIntegrationSuite
+  with JDBCV2JoinPushdownIntegrationSuiteBase {
+
+  override val namespace: String = "DB2INST1"
+
+  override val db = new DB2DatabaseOnDocker
+
+  override lazy val url = db.getJdbcUrl(dockerIp, externalPort)
+
+  override val jdbcDialect: JdbcDialect = DB2Dialect()
+
+  override def caseConvert(identifier: String): String = identifier.toUpperCase(Locale.ROOT)
+
+  override def schemaPreparation(): Unit = {}
+
+  // This method comes from DockerJDBCIntegrationSuite
+  override def dataPreparation(connection: Connection): Unit = {
+    super.dataPreparation()
+  }
+}

sql/core/src/main/scala/org/apache/spark/sql/jdbc/DB2Dialect.scala

@@ -218,7 +218,7 @@ private case class DB2Dialect() extends JdbcDialect with SQLConfHelper with NoLe
       val offsetClause = dialect.getOffsetClause(offset)
 
       options.prepareQuery +
-        s"SELECT $columnList FROM ${options.tableOrQuery}" +
+        s"SELECT $columnList FROM $tableOrQuery" +
         s" $whereClause $groupByClause $orderByClause $offsetClause $limitClause"
     }
   }
@@ -229,4 +229,6 @@ private case class DB2Dialect() extends JdbcDialect with SQLConfHelper with NoLe
   override def supportsLimit: Boolean = true
 
   override def supportsOffset: Boolean = true
+
+  override def supportsJoin: Boolean = true
 }

sql/core/src/test/scala/org/apache/spark/sql/jdbc/JDBCSuite.scala

@@ -1741,6 +1741,29 @@ class JDBCSuite extends SharedSparkSession {
       "SELECT a,b FROM test     FETCH FIRST 123 ROWS ONLY")
   }
 
+  test("DB2Dialect Join pushdown query test") {
+    // JDBC url is a required option but is not used in this test.
+    val options = new JDBCOptions(Map("url" -> "jdbc:db2://host:port", "dbtable" -> "test"))
+    val dialect = DB2Dialect()
+    val left = dialect
+      .getJdbcSQLQueryBuilder(options)
+      .withColumns(Array("a"))
+    val right = dialect
+      .getJdbcSQLQueryBuilder(options)
+      .withColumns(Array("b"))
+
+    val query = dialect
+      .getJdbcSQLQueryBuilder(options)
+      .withJoin(left, right, "L", "R", Array("a", "b"), "INNER JOIN", "L.a = R.b")
+      .build()
+      .replaceAll("\\s+", " ")
+
+    assert(query.contains("INNER JOIN"))
+    assert(query.contains("ON L.a = R.b"))
+    assert(query.contains("SELECT a FROM test"))
+    assert(query.contains("SELECT b FROM test"))
+  }
+
   test("table exists query by jdbc dialect") {
     val MySQL = JdbcDialects.get("jdbc:mysql://127.0.0.1/db")
     val Postgres = JdbcDialects.get("jdbc:postgresql://127.0.0.1/db")