Fix

Author: maropu

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

@@ -309,8 +309,7 @@ class DataFrameReader private[sql](sparkSession: SparkSession) extends Logging {
     val parts: Array[Partition] = predicates.zipWithIndex.map { case (part, i) =>
       JDBCPartition(part, i) : Partition
     }
-    val schema = JDBCRelation.getSchema(sparkSession.sessionState.conf.resolver, options)
-    val relation = JDBCRelation(schema, parts, options)(sparkSession)
+    val relation = JDBCRelation(parts, options)(sparkSession)
     sparkSession.baseRelationToDataFrame(relation)
   }
 

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

@@ -145,6 +145,17 @@ private[sql] object JDBCRelation extends Logging {
       case None => tableSchema
     }
   }
+
+  /**
+   * Resolves a Catalyst schema of a JDBC table and returns [[JDBCRelation]] with the schema.
+   */
+  def apply(
+      parts: Array[Partition],
+      jdbcOptions: JDBCOptions)(
+      sparkSession: SparkSession): JDBCRelation = {
+    val schema = JDBCRelation.getSchema(sparkSession.sessionState.conf.resolver, jdbcOptions)
+    JDBCRelation(schema, parts, jdbcOptions)(sparkSession)
+  }
 }
 
 private[sql] case class JDBCRelation(