simplify ConfigSupport interface.

Author: jiangxb1987

sql/core/src/main/java/org/apache/spark/sql/sources/v2/ConfigSupport.java

@@ -30,33 +30,9 @@
 public interface ConfigSupport {
 
     /**
-     * Create a list of key-prefixes, all session configs that match at least one of the prefixes
-     * will be propagated to the data source options.
-     * If the returned list is empty, no session config will be propagated.
+     * Name for the specified data source, will extract all session configs that starts with
+     * `spark.datasource.$name`, turn `spark.datasource.$name.xxx -> yyy` into
+     * `xxx -> yyy`, and propagate them to all data source operations in this session.
      */
-    List<String> getConfigPrefixes();
-
-    /**
-     * Create a mapping from session config names to data source option names. If a propagated
-     * session config's key doesn't exist in this mapping, the "spark.sql.${source}" prefix will
-     * be trimmed. For example, if the data source name is "parquet", perform the following config
-     * key mapping by default:
-     * "spark.sql.parquet.int96AsTimestamp" -&gt; "int96AsTimestamp",
-     * "spark.sql.parquet.compression.codec" -&gt; "compression.codec",
-     * "spark.sql.columnNameOfCorruptRecord" -&gt; "columnNameOfCorruptRecord".
-     *
-     * If the mapping is specified, for example, the returned map contains an entry
-     * ("spark.sql.columnNameOfCorruptRecord" -&gt; "colNameCorrupt"), then the session config
-     * "spark.sql.columnNameOfCorruptRecord" will be converted to "colNameCorrupt" in
-     * {@link DataSourceV2Options}.
-     */
-    Map<String, String> getConfigMapping();
-
-    /**
-     * Create a list of valid data source option names. When the list is specified, a session
-     * config will NOT be propagated if its corresponding option name is not in the list.
-     *
-     * If the returned list is empty, don't check the option names.
-     */
-    List<String> getValidOptions();
+    String name();
 }

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

@@ -190,7 +190,7 @@ class DataFrameReader private[sql](sparkSession: SparkSession) extends Logging {
       val dataSource = cls.newInstance()
       val options = dataSource match {
         case cs: ConfigSupport =>
-          val confs = withSessionConfig(cs, source, sparkSession.sessionState.conf)
+          val confs = withSessionConfig(cs.name, sparkSession.sessionState.conf)
           new DataSourceV2Options((confs ++ extraOptions).asJava)
         case _ =>
           new DataSourceV2Options(extraOptions.asJava)

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

@@ -29,57 +29,29 @@ import org.apache.spark.sql.sources.v2.ConfigSupport
 private[sql] object DataSourceV2ConfigSupport extends Logging {
 
   /**
-   * Helper method to propagate session configs with config key that matches at least one of the
-   * given prefixes to the corresponding data source options.
+   * Helper method that turns session configs with config keys that start with
+   * `spark.datasource.$name` into k/v pairs, the k/v pairs will be used to create data source
+   * options.
+   * A session config `spark.datasource.$name.xxx -> yyy` will be transformed into
+   * `xxx -> yyy`.
    *
-   * @param cs the session config propagate help class
-   * @param source the data source format
+   * @param name the data source name
    * @param conf the session conf
-   * @return an immutable map that contains all the session configs that should be propagated to
-   *         the data source.
+   * @return an immutable map that contains all the extracted and transformed k/v pairs.
    */
   def withSessionConfig(
-      cs: ConfigSupport,
-      source: String,
+      name: String,
       conf: SQLConf): immutable.Map[String, String] = {
-    val prefixes = cs.getConfigPrefixes
-    require(prefixes != null, "The config key-prefixes cann't be null.")
-    val mapping = cs.getConfigMapping.asScala
-    val validOptions = cs.getValidOptions
-    require(validOptions != null, "The valid options list cann't be null.")
+    require(name != null, "The data source name can't be null.")
 
-    val pattern = Pattern.compile(s"spark\\.sql(\\.$source)?\\.(.*)")
+    val pattern = Pattern.compile(s"spark\\.datasource\\.$name\\.(.*)")
     val filteredConfigs = conf.getAllConfs.filterKeys { confKey =>
-      prefixes.asScala.exists(confKey.startsWith(_))
+      confKey.startsWith(s"spark.datasource.$name")
     }
-    val convertedConfigs = filteredConfigs.map{ entry =>
-      val newKey = mapping.get(entry._1).getOrElse {
-        val m = pattern.matcher(entry._1)
-        if (m.matches()) {
-          m.group(2)
-        } else {
-          // Unable to recognize the session config key.
-          logWarning(s"Unrecognizable session config name ${entry._1}.")
-          entry._1
-        }
-      }
-      (newKey, entry._2)
-    }
-    if (validOptions.size == 0) {
-      convertedConfigs
-    } else {
-      // Check whether all the valid options are propagated.
-      validOptions.asScala.foreach { optionName =>
-        if (!convertedConfigs.keySet.contains(optionName)) {
-          logWarning(s"Data source option '$optionName' is required, but not propagated from " +
-            "session config, please check the config settings.")
-        }
-      }
-
-      // Filter the valid options.
-      convertedConfigs.filterKeys { optionName =>
-        validOptions.contains(optionName)
-      }
+    filteredConfigs.map { entry =>
+      val m = pattern.matcher(entry._1)
+      require(m.matches() && m.groupCount() > 0, s"Fail in matching ${entry._1} with $pattern.")
+      (m.group(1), entry._2)
     }
   }
 }

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

@@ -17,7 +17,6 @@
 
 package org.apache.spark.sql.sources.v2
 
-import java.util
 import java.util.{ArrayList, List => JList}
 
 import test.org.apache.spark.sql.sources.v2._
@@ -35,6 +34,8 @@ import org.apache.spark.sql.types.StructType
 class DataSourceV2Suite extends QueryTest with SharedSQLContext {
   import testImplicits._
 
+  private val dsName = "userDefinedDataSource"
+
   test("simplest implementation") {
     Seq(classOf[SimpleDataSourceV2], classOf[JavaSimpleDataSourceV2]).foreach { cls =>
       withClue(cls.getName) {
@@ -47,32 +48,18 @@ class DataSourceV2Suite extends QueryTest with SharedSQLContext {
   }
 
   test("simple implementation with config support") {
-    withSQLConf(SQLConf.PARQUET_SCHEMA_MERGING_ENABLED.key -> "false",
-        SQLConf.PARQUET_COMPRESSION.key -> "uncompressed",
-        SQLConf.PARALLEL_PARTITION_DISCOVERY_THRESHOLD.key -> "32",
-        SQLConf.PARALLEL_PARTITION_DISCOVERY_PARALLELISM.key -> "10000") {
+    // Only match configs with keys start with "spark.datasource.${dsName}".
+    withSQLConf(s"spark.datasource.$dsName.foo.bar" -> "false",
+        s"spark.datasource.$dsName.whateverConfigName" -> "123",
+        s"spark.sql.$dsName.config.name" -> "false",
+        s"spark.datasource.another.config.name" -> "123") {
       val cs = classOf[DataSourceV2WithConfig].newInstance().asInstanceOf[ConfigSupport]
-      val confs = DataSourceV2ConfigSupport.withSessionConfig(cs, "parquet", SQLConf.get)
-      assert(confs.size == 3)
-      assert(confs.keySet.filter(_.startsWith("spark.sql.parquet")).size == 0)
-      assert(confs.keySet.filter(_.startsWith("not.exist.prefix")).size == 0)
-      assert(confs.keySet.contains("compressionCodec"))
-      assert(confs.keySet.contains("sources.parallelPartitionDiscovery.threshold"))
-    }
-  }
-
-  test("config support with validOptions") {
-    withSQLConf(SQLConf.PARQUET_SCHEMA_MERGING_ENABLED.key -> "false",
-      SQLConf.PARQUET_COMPRESSION.key -> "uncompressed",
-      SQLConf.PARALLEL_PARTITION_DISCOVERY_THRESHOLD.key -> "32",
-      SQLConf.PARALLEL_PARTITION_DISCOVERY_PARALLELISM.key -> "10000") {
-      val cs = classOf[DataSourceV2WithValidOptions].newInstance().asInstanceOf[ConfigSupport]
-      val confs = DataSourceV2ConfigSupport.withSessionConfig(cs, "parquet", SQLConf.get)
+      val confs = DataSourceV2ConfigSupport.withSessionConfig(cs.name, SQLConf.get)
       assert(confs.size == 2)
-      assert(confs.keySet.filter(_.startsWith("spark.sql.parquet")).size == 0)
+      assert(confs.keySet.filter(_.startsWith("spark.datasource")).size == 0)
       assert(confs.keySet.filter(_.startsWith("not.exist.prefix")).size == 0)
-      assert(confs.keySet.contains("compressionCodec"))
-      assert(confs.keySet.contains("sources.parallelPartitionDiscovery.threshold"))
+      assert(confs.keySet.contains("foo.bar"))
+      assert(confs.keySet.contains("whateverConfigName"))
     }
   }
 
@@ -214,29 +201,7 @@ class SimpleReadTask(start: Int, end: Int) extends ReadTask[Row] with DataReader
 
 class DataSourceV2WithConfig extends SimpleDataSourceV2 with ConfigSupport {
 
-  override def getConfigPrefixes: JList[String] = {
-    java.util.Arrays.asList(
-      "spark.sql.parquet",
-      "spark.sql.sources.parallelPartitionDiscovery.threshold")
-  }
-
-  override def getConfigMapping: util.Map[String, String] = {
-    val configMap = new util.HashMap[String, String]()
-    configMap.put("spark.sql.parquet.compression.codec", "compressionCodec")
-    configMap
-  }
-
-  override def getValidOptions: JList[String] = new util.ArrayList[String]()
-}
-
-class DataSourceV2WithValidOptions extends DataSourceV2WithConfig {
-
-  override def getValidOptions: JList[String] = {
-    java.util.Arrays.asList(
-      "sources.parallelPartitionDiscovery.threshold",
-      "compressionCodec",
-      "not.exist.option")
-  }
+  override def name: String = "userDefinedDataSource"
 }
 
 class AdvancedDataSourceV2 extends DataSourceV2 with ReadSupport {