comment address, ut fix

Author: xuanyuanking

python/pyspark/sql/functions.py

@@ -2766,15 +2766,13 @@ def map_concat(*cols):
     :param cols: list of column names (string) or list of :class:`Column` expressions
 
     >>> from pyspark.sql.functions import map_concat
-    >>> spark.conf.set("spark.sql.deduplicateMapKey.lastWinsPolicy.enabled", "true")
-    >>> df = spark.sql("SELECT map(1, 'a', 2, 'b') as map1, map(3, 'c', 1, 'd') as map2")
+    >>> df = spark.sql("SELECT map(1, 'a', 2, 'b') as map1, map(3, 'c') as map2")
     >>> df.select(map_concat("map1", "map2").alias("map3")).show(truncate=False)
     +------------------------+
     |map3                    |
     +------------------------+
     |[1 -> d, 2 -> b, 3 -> c]|
     +------------------------+
-    >>> spark.conf.unset("spark.sql.deduplicateMapKey.lastWinsPolicy.enabled")
     """
     sc = SparkContext._active_spark_context
     if len(cols) == 1 and isinstance(cols[0], (list, set)):

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/ArrayBasedMapBuilder.scala

@@ -50,6 +50,9 @@ class ArrayBasedMapBuilder(keyType: DataType, valueType: DataType) extends Seria
   private lazy val keyGetter = InternalRow.getAccessor(keyType)
   private lazy val valueGetter = InternalRow.getAccessor(valueType)
 
+  private val deduplicateWithLastWinsPolicy =
+    SQLConf.get.getConf(DEDUPLICATE_MAP_KEY_WITH_LAST_WINS_POLICY)
+
   def put(key: Any, value: Any): Unit = {
     if (key == null) {
       throw new RuntimeException("Cannot use null as map key.")
@@ -65,7 +68,7 @@ class ArrayBasedMapBuilder(keyType: DataType, valueType: DataType) extends Seria
       keys.append(key)
       values.append(value)
     } else {
-      if (!SQLConf.get.getConf(DEDUPLICATE_MAP_KEY_WITH_LAST_WINS_POLICY)) {
+      if (!deduplicateWithLastWinsPolicy) {
         throw new RuntimeException(s"Duplicate map key $key was founded, please set " +
           s"${DEDUPLICATE_MAP_KEY_WITH_LAST_WINS_POLICY.key} to true to remove it with " +
           "last wins policy.")

sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala

@@ -2171,7 +2171,9 @@ object SQLConf {
     buildConf("spark.sql.deduplicateMapKey.lastWinsPolicy.enabled")
       .doc("When true, use last wins policy to remove duplicated map keys in built-in functions, " +
         "this config takes effect in below build-in functions: CreateMap, MapFromArrays, " +
-        "MapFromEntries, StringToMap, MapConcat and TransformKeys.")
+        "MapFromEntries, StringToMap, MapConcat and TransformKeys. Otherwise, if this is false, " +
+        "which is the default, Spark will throw an exception while duplicated map keys are " +
+        "detected.")
       .booleanConf
       .createWithDefault(false)
 

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/util/ArrayBasedMapBuilderSuite.scala

@@ -44,6 +44,13 @@ class ArrayBasedMapBuilderSuite extends SparkFunSuite with SQLHelper {
     assert(e.getMessage.contains("Cannot use null as map key"))
   }
 
+  test("fail while duplicated keys detected") {
+    val builder = new ArrayBasedMapBuilder(IntegerType, IntegerType)
+    builder.put(1, 1)
+    val e = intercept[RuntimeException](builder.put(1, 2))
+    assert(e.getMessage.contains("Duplicate map key 1 was founded"))
+  }
+
   test("remove duplicated keys with last wins policy") {
     withSQLConf(SQLConf.DEDUPLICATE_MAP_KEY_WITH_LAST_WINS_POLICY.key -> "true") {
       val builder = new ArrayBasedMapBuilder(IntegerType, IntegerType)

sql/core/src/test/scala/org/apache/spark/sql/SQLQuerySuite.scala

@@ -188,11 +188,13 @@ class SQLQuerySuite extends QueryTest with SharedSparkSession with AdaptiveSpark
           checkExampleSyntax(example)
           example.split("  > ").toList.foreach(_ match {
             case exampleRe(sql, output) =>
-              val df = clonedSpark.sql(sql)
-              val actual = unindentAndTrim(
-                hiveResultString(df.queryExecution.executedPlan).mkString("\n"))
-              val expected = unindentAndTrim(output)
-              assert(actual === expected)
+              withSQLConf(SQLConf.DEDUPLICATE_MAP_KEY_WITH_LAST_WINS_POLICY.key -> "true") {
+                val df = clonedSpark.sql(sql)
+                val actual = unindentAndTrim(
+                  hiveResultString(df.queryExecution.executedPlan).mkString("\n"))
+                val expected = unindentAndTrim(output)
+                assert(actual === expected)
+              }
             case _ =>
           })
         }