[SPARK-32592][SQL] Make DataFrameReader.table take the specified options

Author: huaxingao

sql/core/src/main/scala/org/apache/spark/sql/DataFrameReader.scala

@@ -40,7 +40,6 @@ import org.apache.spark.sql.execution.datasources.csv._
 import org.apache.spark.sql.execution.datasources.jdbc._
 import org.apache.spark.sql.execution.datasources.json.TextInputJsonDataSource
 import org.apache.spark.sql.execution.datasources.v2.{DataSourceV2Relation, DataSourceV2Utils}
-import org.apache.spark.sql.internal.SQLConf
 import org.apache.spark.sql.types.StructType
 import org.apache.spark.sql.util.CaseInsensitiveStringMap
 import org.apache.spark.unsafe.types.UTF8String
@@ -822,6 +821,8 @@ class DataFrameReader private[sql](sparkSession: SparkSession) extends Logging {
    */
   def table(tableName: String): DataFrame = {
     assertNoSpecifiedSchema("table")
+    for ((k, v) <- this.extraOptions)
+      sparkSession.conf.set(k, v)
     sparkSession.table(tableName)
   }
 

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/jdbc/JDBCOptions.scala

@@ -232,7 +232,7 @@ class JdbcOptionsInWrite(
 
 object JDBCOptions {
   private val curId = new java.util.concurrent.atomic.AtomicLong(0L)
-  private val jdbcOptionNames = collection.mutable.Set[String]()
+  val jdbcOptionNames = collection.mutable.Set[String]()
 
   private def newOption(name: String): String = {
     jdbcOptionNames += name.toLowerCase(Locale.ROOT)

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/jdbc/JDBCTableCatalog.scala

@@ -21,6 +21,7 @@ import java.sql.{Connection, SQLException}
 import scala.collection.JavaConverters._
 
 import org.apache.spark.internal.Logging
+import org.apache.spark.sql.SparkSession
 import org.apache.spark.sql.catalyst.analysis.{NoSuchNamespaceException, NoSuchTableException}
 import org.apache.spark.sql.connector.catalog.{Identifier, Table, TableCatalog, TableChange}
 import org.apache.spark.sql.connector.expressions.Transform
@@ -94,8 +95,14 @@ class JDBCTableCatalog extends TableCatalog with Logging {
 
   override def loadTable(ident: Identifier): Table = {
     checkNamespace(ident.namespace())
+    var newOptions = options.parameters
+    SparkSession.active.sessionState.conf.settings.asScala.foreach {
+      case (key, value) =>
+        for (option <- JDBCOptions.jdbcOptionNames)
+          if (key.equalsIgnoreCase(option)) newOptions += (key -> value)
+    }
     val optionsWithTableName = new JDBCOptions(
-      options.parameters + (JDBCOptions.JDBC_TABLE_NAME -> getTableName(ident)))
+      newOptions + (JDBCOptions.JDBC_TABLE_NAME -> getTableName(ident)))
     try {
       val schema = JDBCRDD.resolveTable(optionsWithTableName)
       JDBCTable(ident, schema, optionsWithTableName)