add method getValidOptions

Author: jiangxb1987

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

@@ -51,4 +51,12 @@ public interface ConfigSupport {
      * [[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();
 }

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

@@ -45,11 +45,14 @@ private[sql] object DataSourceV2ConfigSupport extends Logging {
     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.")
 
     val pattern = Pattern.compile(s"spark\\.sql(\\.$source)?\\.(.*)")
-    conf.getAllConfs.filterKeys { confKey =>
+    val filteredConfigs = conf.getAllConfs.filterKeys { confKey =>
       prefixes.asScala.exists(confKey.startsWith(_))
-    }.map{ entry =>
+    }
+    val convertedConfigs = filteredConfigs.map{ entry =>
       val newKey = mapping.get(entry._1).getOrElse {
         val m = pattern.matcher(entry._1)
         if (m.matches()) {
@@ -62,5 +65,21 @@ private[sql] object DataSourceV2ConfigSupport extends Logging {
       }
       (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)
+      }
+    }
   }
 }

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

@@ -61,6 +61,21 @@ class DataSourceV2Suite extends QueryTest with SharedSQLContext {
     }
   }
 
+  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)
+      assert(confs.size == 2)
+      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("advanced implementation") {
     Seq(classOf[AdvancedDataSourceV2], classOf[JavaAdvancedDataSourceV2]).foreach { cls =>
       withClue(cls.getName) {
@@ -210,6 +225,18 @@ class DataSourceV2WithConfig extends SimpleDataSourceV2 with ConfigSupport {
     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")
+  }
 }
 
 class AdvancedDataSourceV2 extends DataSourceV2 with ReadSupport {