resolve comments

Author: PetarVasiljevic-DB

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

@@ -22,6 +22,7 @@ import org.apache.logging.log4j.Level
 import org.apache.spark.rdd.RDD
 import org.apache.spark.sql.{AnalysisException, DataFrame}
 import org.apache.spark.sql.catalyst.analysis.{IndexAlreadyExistsException, NoSuchIndexException}
+import org.apache.spark.sql.connector.DataSourcePushdownTestUtils
 import org.apache.spark.sql.connector.catalog.{Catalogs, Identifier, TableCatalog}
 import org.apache.spark.sql.connector.catalog.index.SupportsIndex
 import org.apache.spark.sql.connector.expressions.NullOrdering
@@ -33,7 +34,7 @@ import org.apache.spark.tags.DockerTest
 
 @DockerTest
 private[v2] trait V2JDBCTest
-  extends V2JDBCPushdownTestUtils
+  extends DataSourcePushdownTestUtils
   with DockerIntegrationFunSuite
   with SharedSparkSession {
   import testImplicits._

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

@@ -23,7 +23,7 @@ import java.util.Locale
 import org.apache.spark.SparkConf
 import org.apache.spark.sql.execution.datasources.v2.jdbc.JDBCTableCatalog
 import org.apache.spark.sql.jdbc.{DockerJDBCIntegrationSuite, JdbcDialect, OracleDatabaseOnDocker, OracleDialect}
-import org.apache.spark.sql.jdbc.v2.JDBCJoinPushdownIntegrationSuite
+import org.apache.spark.sql.jdbc.v2.JDBCV2JoinPushdownIntegrationSuiteBase
 import org.apache.spark.tags.DockerTest
 
 /**
@@ -56,10 +56,10 @@ import org.apache.spark.tags.DockerTest
 @DockerTest
 class OracleJoinPushdownIntegrationSuite
   extends DockerJDBCIntegrationSuite
-  with JDBCJoinPushdownIntegrationSuite {
+  with JDBCV2JoinPushdownIntegrationSuiteBase {
   override val catalogName: String = "oracle"
 
-  override def namespaceOpt: Option[String] = Some("SYSTEM")
+  override val namespaceOpt: Option[String] = Some("SYSTEM")
 
   override val db = new OracleDatabaseOnDocker
 

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

@@ -208,7 +208,7 @@ class JdbcSQLQueryBuilder(dialect: JdbcDialect, options: JDBCOptions) {
 
   // If join has been pushed down, reuse join query as a subquery. Otherwise, fallback to
   // what is provided in options.
-  def tableOrQuery: String = joinQuery.getOrElse(options.tableOrQuery)
+  protected final def tableOrQuery: String = joinQuery.getOrElse(options.tableOrQuery)
 
   /**
    * Build the final SQL query that following dialect's SQL syntax.

sql/core/src/test/scala/org/apache/spark/sql/jdbc/v2/V2JDBCPushdownTestUtils.scala renamed to sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourcePushdownTestUtils.scala

@@ -15,15 +15,15 @@
  * limitations under the License.
  */
 
-package org.apache.spark.sql.jdbc.v2
+package org.apache.spark.sql.connector
 
 import org.apache.spark.sql.{DataFrame, ExplainSuiteHelper}
-import org.apache.spark.sql.catalyst.plans.logical.{Aggregate, Filter, GlobalLimit, Join, LocalLimit, Offset, Sample, Sort}
+import org.apache.spark.sql.catalyst.plans.logical._
 import org.apache.spark.sql.connector.expressions.aggregate.GeneralAggregateFunc
 import org.apache.spark.sql.execution.datasources.v2.{DataSourceV2ScanRelation, V1ScanWrapper}
 import org.apache.spark.sql.internal.SQLConf
 
-trait V2JDBCPushdownTestUtils extends ExplainSuiteHelper {
+trait DataSourcePushdownTestUtils extends ExplainSuiteHelper {
   protected def checkSamplePushed(df: DataFrame, pushed: Boolean = true): Unit = {
     val sample = df.queryExecution.optimizedPlan.collect {
       case s: Sample => s

sql/core/src/test/scala/org/apache/spark/sql/jdbc/v2/JDBCJoinPushdownIntegrationSuite.scala renamed to sql/core/src/test/scala/org/apache/spark/sql/jdbc/v2/JDBCV2JoinPushdownIntegrationSuiteBase.scala

@@ -21,18 +21,19 @@ import java.sql.{Connection, DriverManager}
 import java.util.Properties
 
 import org.apache.spark.sql.QueryTest
+import org.apache.spark.sql.connector.DataSourcePushdownTestUtils
 import org.apache.spark.sql.execution.datasources.jdbc.JdbcUtils
 import org.apache.spark.sql.internal.SQLConf
 import org.apache.spark.sql.jdbc.JdbcDialect
 import org.apache.spark.sql.test.SharedSparkSession
 import org.apache.spark.sql.types.{DataType, DataTypes}
 
-trait JDBCJoinPushdownIntegrationSuite
+trait JDBCV2JoinPushdownIntegrationSuiteBase
   extends QueryTest
   with SharedSparkSession
-  with V2JDBCPushdownTestUtils {
+  with DataSourcePushdownTestUtils {
   val catalogName: String
-  def namespaceOpt: Option[String] = None
+  val namespaceOpt: Option[String] = None
   val url: String
 
   val joinTableName1: String = "join_table_1"
@@ -46,9 +47,9 @@ trait JDBCJoinPushdownIntegrationSuite
   def qualifyTableName(tableName: String): String = namespaceOpt
     .map(namespace => s"$namespace.$tableName").getOrElse(tableName)
 
-  private val fullyQualifiedTableName1: String = qualifyTableName(joinTableName1)
+  private lazy val fullyQualifiedTableName1: String = qualifyTableName(joinTableName1)
 
-  private val fullyQualifiedTableName2: String = qualifyTableName(joinTableName2)
+  private lazy val fullyQualifiedTableName2: String = qualifyTableName(joinTableName2)
 
   protected def getJDBCTypeString(dt: DataType): String = {
     JdbcUtils.getJdbcType(dt, jdbcDialect).databaseTypeDefinition.toUpperCase()

sql/core/src/test/scala/org/apache/spark/sql/jdbc/JDBCV2JoinPushdownSuite.scala renamed to sql/core/src/test/scala/org/apache/spark/sql/jdbc/v2/JDBCV2JoinPushdownSuite.scala

@@ -15,28 +15,29 @@
  * limitations under the License.
  */
 
-package org.apache.spark.sql.jdbc
+package org.apache.spark.sql.jdbc.v2
 
 import java.sql.Connection
 
 import org.apache.spark.SparkConf
 import org.apache.spark.sql.{ExplainSuiteHelper, QueryTest}
+import org.apache.spark.sql.connector.DataSourcePushdownTestUtils
 import org.apache.spark.sql.execution.datasources.v2.jdbc.JDBCTableCatalog
-import org.apache.spark.sql.jdbc.v2.{JDBCJoinPushdownIntegrationSuite, V2JDBCPushdownTestUtils}
+import org.apache.spark.sql.jdbc.{H2Dialect, JdbcDialect}
 import org.apache.spark.sql.test.SharedSparkSession
 import org.apache.spark.util.Utils
 
 class JDBCV2JoinPushdownSuite
   extends QueryTest
   with SharedSparkSession
   with ExplainSuiteHelper
-  with V2JDBCPushdownTestUtils
-  with JDBCJoinPushdownIntegrationSuite {
+  with DataSourcePushdownTestUtils
+  with JDBCV2JoinPushdownIntegrationSuiteBase {
   val tempDir = Utils.createTempDir()
   override val url = s"jdbc:h2:${tempDir.getCanonicalPath};user=testUser;password=testPass"
 
   override val catalogName: String = "h2"
-  override def namespaceOpt: Option[String] = Some("test")
+  override val namespaceOpt: Option[String] = Some("test")
 
   override val jdbcDialect: JdbcDialect = H2Dialect()