move around function withSessionConfig.

Author: jiangxb1987

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

@@ -20,7 +20,6 @@ package org.apache.spark.sql
 import java.util.{Locale, Properties}
 
 import scala.collection.JavaConverters._
-import scala.collection.immutable
 
 import org.apache.spark.Partition
 import org.apache.spark.annotation.InterfaceStability
@@ -33,8 +32,8 @@ import org.apache.spark.sql.execution.datasources.{DataSource, FailureSafeParser
 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.DataSourceV2ConfigSupport
 import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Relation
-import org.apache.spark.sql.internal.SQLConf
 import org.apache.spark.sql.sources.v2._
 import org.apache.spark.sql.types.{StringType, StructType}
 import org.apache.spark.unsafe.types.UTF8String
@@ -171,7 +170,7 @@ class DataFrameReader private[sql](sparkSession: SparkSession) extends Logging {
     option("path", path).load(Seq.empty: _*) // force invocation of `load(...varargs...)`
   }
 
-  import DataFrameReader._
+  import DataSourceV2ConfigSupport._
 
   /**
    * Loads input in as a `DataFrame`, for data sources that support multiple paths.
@@ -743,25 +742,3 @@ class DataFrameReader private[sql](sparkSession: SparkSession) extends Logging {
   private val extraOptions = new scala.collection.mutable.HashMap[String, String]
 
 }
-
-private[sql] object DataFrameReader {
-
-  /**
-   * Helper method to filter session configs with config key that matches at least one of the given
-   * prefixes.
-   *
-   * @param cs the config key-prefixes that should be filtered.
-   * @param conf the session conf
-   * @return an immutable map that contains all the session configs that should be propagated to
-   *         the data source.
-   */
-  def withSessionConfig(
-      cs: ConfigSupport,
-      conf: SQLConf): immutable.Map[String, String] = {
-    val prefixes = cs.getConfigPrefixes
-    require(prefixes != null, "The config key-prefixes cann't be null.")
-    conf.getAllConfs.filterKeys { confKey =>
-      prefixes.asScala.exists(confKey.startsWith(_))
-    }
-  }
-}

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

@@ -0,0 +1,46 @@
+/*
+ * 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.execution.datasources.v2
+
+import scala.collection.JavaConverters._
+import scala.collection.immutable
+
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.sources.v2.ConfigSupport
+
+private[sql] object DataSourceV2ConfigSupport {
+
+  /**
+   * Helper method to filter session configs with config key that matches at least one of the given
+   * prefixes.
+   *
+   * @param cs the config key-prefixes that should be filtered.
+   * @param conf the session conf
+   * @return an immutable map that contains all the session configs that should be propagated to
+   *         the data source.
+   */
+  def withSessionConfig(
+      cs: ConfigSupport,
+      conf: SQLConf): immutable.Map[String, String] = {
+    val prefixes = cs.getConfigPrefixes
+    require(prefixes != null, "The config key-prefixes cann't be null.")
+    conf.getAllConfs.filterKeys { confKey =>
+      prefixes.asScala.exists(confKey.startsWith(_))
+    }
+  }
+}

sql/core/src/test/scala/org/apache/spark/sql/sources/v2/DataSourceV2Suite.scala

@@ -22,8 +22,9 @@ import java.util.{ArrayList, List => JList}
 import test.org.apache.spark.sql.sources.v2._
 
 import org.apache.spark.SparkException
-import org.apache.spark.sql.{AnalysisException, DataFrameReader, QueryTest, Row}
+import org.apache.spark.sql.{AnalysisException, QueryTest, Row}
 import org.apache.spark.sql.catalyst.expressions.UnsafeRow
+import org.apache.spark.sql.execution.datasources.v2.DataSourceV2ConfigSupport
 import org.apache.spark.sql.internal.SQLConf
 import org.apache.spark.sql.sources.{Filter, GreaterThan}
 import org.apache.spark.sql.sources.v2.reader._
@@ -50,7 +51,7 @@ class DataSourceV2Suite extends QueryTest with SharedSQLContext {
         SQLConf.PARALLEL_PARTITION_DISCOVERY_THRESHOLD.key -> "32",
         SQLConf.PARALLEL_PARTITION_DISCOVERY_PARALLELISM.key -> "10000") {
       val cs = classOf[DataSourceV2WithConfig].newInstance().asInstanceOf[ConfigSupport]
-      val confs = DataFrameReader.withSessionConfig(cs, SQLConf.get)
+      val confs = DataSourceV2ConfigSupport.withSessionConfig(cs, SQLConf.get)
       assert(confs.size == 3)
       assert(confs.keySet.filter(_.startsWith("spark.sql.parquet")).size == 2)
       assert(confs.keySet.filter(