diff --git a/connector/docker-integration-tests/src/test/scala/org/apache/spark/sql/jdbc/v2/join/DB2JoinPushdownIntegrationSuite.scala b/connector/docker-integration-tests/src/test/scala/org/apache/spark/sql/jdbc/v2/join/DB2JoinPushdownIntegrationSuite.scala new file mode 100644 index 0000000000000..a58fab9121353 --- /dev/null +++ b/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() + } +} diff --git a/sql/core/src/main/scala/org/apache/spark/sql/jdbc/DB2Dialect.scala b/sql/core/src/main/scala/org/apache/spark/sql/jdbc/DB2Dialect.scala index 71753c1155ce6..9905313491efd 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/jdbc/DB2Dialect.scala +++ b/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 } diff --git a/sql/core/src/test/scala/org/apache/spark/sql/jdbc/JDBCSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/jdbc/JDBCSuite.scala index 6642b527e40e1..a0b908b85e24c 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/jdbc/JDBCSuite.scala +++ b/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")