Change function signature.

Author: dongjoon-hyun

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

@@ -182,7 +182,7 @@ class DataFrameReader private[sql](sparkSession: SparkSession) extends Logging {
         "read files of Hive data source directly.")
     }
 
-    val cls = DataSource.lookupDataSource(sparkSession.sessionState.conf, source)
+    val cls = DataSource.lookupDataSource(source, sparkSession.sessionState.conf)
     if (classOf[DataSourceV2].isAssignableFrom(cls)) {
       val options = new DataSourceV2Options(extraOptions.asJava)
 

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

@@ -234,7 +234,7 @@ final class DataFrameWriter[T] private[sql](ds: Dataset[T]) {
 
     assertNotBucketed("save")
 
-    val cls = DataSource.lookupDataSource(df.sparkSession.sessionState.conf, source)
+    val cls = DataSource.lookupDataSource(source, df.sparkSession.sessionState.conf)
     if (classOf[DataSourceV2].isAssignableFrom(cls)) {
       cls.newInstance() match {
         case ds: WriteSupport =>

sql/core/src/main/scala/org/apache/spark/sql/execution/command/tables.scala

@@ -231,7 +231,7 @@ case class AlterTableAddColumnsCommand(
     }
 
     if (DDLUtils.isDatasourceTable(catalogTable)) {
-      DataSource.lookupDataSource(conf, catalogTable.provider.get).newInstance() match {
+      DataSource.lookupDataSource(catalogTable.provider.get, conf).newInstance() match {
         // For datasource table, this command can only support the following File format.
         // TextFileFormat only default to one column "value"
         // Hive type is already considered as hive serde table, so the logic will not

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

@@ -88,7 +88,7 @@ case class DataSource(
   case class SourceInfo(name: String, schema: StructType, partitionColumns: Seq[String])
 
   lazy val providingClass: Class[_] =
-    DataSource.lookupDataSource(sparkSession.sessionState.conf, className)
+    DataSource.lookupDataSource(className, sparkSession.sessionState.conf)
   lazy val sourceInfo: SourceInfo = sourceSchema()
   private val caseInsensitiveOptions = CaseInsensitiveMap(options)
   private val equality = sparkSession.sessionState.conf.resolver
@@ -574,7 +574,7 @@ object DataSource extends Logging {
     "org.apache.spark.Logging")
 
   /** Given a provider name, look up the data source class definition. */
-  def lookupDataSource(conf: SQLConf, provider: String): Class[_] = {
+  def lookupDataSource(provider: String, conf: SQLConf): Class[_] = {
     val provider1 = backwardCompatibilityMap.getOrElse(provider, provider) match {
       case name if name.equalsIgnoreCase("orc") && conf.getConf(SQLConf.ORC_USE_NEW_VERSION) =>
         classOf[OrcFileFormat].getCanonicalName

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

@@ -109,8 +109,8 @@ case class PreprocessTableCreation(sparkSession: SparkSession) extends Rule[Logi
 
       // Check if the specified data source match the data source of the existing table.
       val conf = sparkSession.sessionState.conf
-      val existingProvider = DataSource.lookupDataSource(conf, existingTable.provider.get)
-      val specifiedProvider = DataSource.lookupDataSource(conf, tableDesc.provider.get)
+      val existingProvider = DataSource.lookupDataSource(existingTable.provider.get, conf)
+      val specifiedProvider = DataSource.lookupDataSource(tableDesc.provider.get, conf)
       // TODO: Check that options from the resolved relation match the relation that we are
       // inserting into (i.e. using the same compression).
       if (existingProvider != specifiedProvider) {