[GLUTEN-6101][VL] Enable map_from_arrays function

backends-velox/src/test/scala/org/apache/gluten/execution/MiscOperatorSuite.scala

@@ -2340,16 +2340,6 @@ class MiscOperatorSuite extends VeloxWholeStageTransformerSuite with AdaptiveSpa
       })
   }
 
-  test("Expression unsupported by backend can be handled by ColumnarPartialProject") {
-    runQueryAndCompare(
-      "SELECT c_custkey, map_from_arrays(array(c_name), array(c_comment)) FROM customer") {
-      df =>
-        val executedPlan = getExecutedPlan(df)
-        assert(executedPlan.count(_.isInstanceOf[ProjectExec]) == 0)
-        assert(executedPlan.count(_.isInstanceOf[ColumnarPartialProjectExec]) == 1)
-    }
-  }
-
   testWithMinSparkVersion("Left single join should not result into exception", "4.0") {
     withSQLConf(SQLConf.ANSI_ENABLED.key -> "false") {
       spark.sql("create temp view x (x1, x2) as values (1, 1), (2, 2);")

backends-velox/src/test/scala/org/apache/gluten/functions/ScalarFunctionsValidateSuite.scala

@@ -550,6 +550,42 @@ class ScalarFunctionsValidateSuite extends FunctionsValidateSuite {
     }
   }
 
+  test("map_from_arrays offloads to Velox under both mapKeyDedupPolicy values") {
+    // l_orderkey is repeated.
+    val duplicateKeyQuery =
+      "select map_from_arrays(array(l_orderkey, l_orderkey + 1, l_orderkey), " +
+        "array(l_partkey, l_suppkey, l_linenumber)) as m, " +
+        "map_keys(map_from_arrays(array(l_orderkey, l_orderkey + 1, l_orderkey), " +
+        "array(l_partkey, l_suppkey, l_linenumber))) as k from lineitem limit 10"
+
+    // l_orderkey is not repeated.
+    val distinctKeyQuery =
+      "select map_from_arrays(array(l_orderkey, l_orderkey + 1), " +
+        "array(l_partkey, l_suppkey)) from lineitem limit 10"
+
+    withSQLConf(SQLConf.MAP_KEY_DEDUP_POLICY.key -> SQLConf.MapKeyDedupPolicy.EXCEPTION.toString) {
+      // EXCEPTION policy passes when there is no duplicate.
+      runQueryAndCompare(distinctKeyQuery) {
+        checkGlutenPlan[ProjectExecTransformer]
+      }
+
+      // EXCEPTION policy raises on a duplicate.
+      val df = sql(duplicateKeyQuery)
+      checkGlutenPlan[ProjectExecTransformer](df)
+      val e = intercept[SparkException] {
+        df.collect()
+      }
+      assert(e.getMessage.contains("Duplicate map key"))
+    }
+
+    withSQLConf(SQLConf.MAP_KEY_DEDUP_POLICY.key -> SQLConf.MapKeyDedupPolicy.LAST_WIN.toString) {
+      // LAST_WIN policy keeps the duplicate's first position and its last value.
+      runQueryAndCompare(duplicateKeyQuery) {
+        checkGlutenPlan[ProjectExecTransformer]
+      }
+    }
+  }
+
   test("raise_error, assert_true") {
     runQueryAndCompare("""SELECT assert_true(l_orderkey >= 1), l_orderkey
                          | from lineitem limit 100""".stripMargin) {

cpp/velox/substrait/SubstraitToVeloxPlanValidator.cc

@@ -56,7 +56,7 @@ const char* extractFileName(const char* file) {
 const std::unordered_set<std::string> kRegexFunctions =
     {"regexp_extract", "regexp_extract_all", "regexp_replace", "regexp_instr", "rlike", "split"};
 
-const std::unordered_set<std::string> kBlackList = {"split_part", "sequence", "approx_percentile", "map_from_arrays"};
+const std::unordered_set<std::string> kBlackList = {"split_part", "sequence", "approx_percentile"};
 } // namespace
 
 bool SubstraitToVeloxPlanValidator::parseVeloxType(

docs/velox-backend-scalar-function-support.md

@@ -1,6 +1,6 @@
 # Scalar Functions Support Status
 
-**Out of 357 scalar functions in Spark 3.5, Gluten currently fully supports 246 functions and partially supports 28 functions.**
+**Out of 357 scalar functions in Spark 3.5, Gluten currently fully supports 247 functions and partially supports 28 functions.**
 
 **Gluten also fully supports 2 additional functions introduced in Spark 4.0.**
 
@@ -215,7 +215,7 @@
 | map_concat        | MapConcat           | PS       |                                                                             |
 | map_contains_key  | MapContainsKey      | S        |                                                                             |
 | map_entries       | MapEntries          | S        |                                                                             |
-| map_from_arrays   | MapFromArrays       |          |                                                                             |
+| map_from_arrays   | MapFromArrays       | S        |                                                                             |
 | map_from_entries  | MapFromEntries      | S        |                                                                             |
 | map_keys          | MapKeys             | S        |                                                                             |
 | map_values        | MapValues           | S        |                                                                             |

gluten-ut/spark34/src/test/scala/org/apache/gluten/utils/clickhouse/ClickHouseTestSettings.scala

@@ -167,6 +167,13 @@ class ClickHouseTestSettings extends BackendTestSettings {
     .excludeGlutenTest("length check for input string values: with implicit cast")
     .excludeGlutenTest("char/varchar type values length check: partitioned columns of other types")
     .excludeGlutenTest("SPARK-42611: check char/varchar length in reordered structs within arrays")
+    .excludeGlutenTest("length check for input string values: nested in map key")
+    .excludeGlutenTest("length check for input string values: nested in map value")
+    .excludeGlutenTest("length check for input string values: nested in both map key and value")
+    .excludeGlutenTest(
+      "SPARK-42611: check char/varchar length in reordered structs within map keys")
+    .excludeGlutenTest(
+      "SPARK-42611: check char/varchar length in reordered structs within map values")
   enableSuite[GlutenDSV2SQLInsertTestSuite]
   enableSuite[GlutenDataFrameAggregateSuite]
     .exclude("average")
@@ -195,6 +202,7 @@ class ClickHouseTestSettings extends BackendTestSettings {
     // Expected exception org.apache.spark.SparkException to be thrown, but no exception was thrown
     .exclude("map_concat function")
     .exclude("map with arrays")
+    .excludeGlutenTest("map with arrays")
     .exclude("flatten function")
     .exclude("aggregate function - array for primitive type not containing null")
     .exclude("aggregate function - array for primitive type containing null")
@@ -391,6 +399,9 @@ class ClickHouseTestSettings extends BackendTestSettings {
     .excludeGlutenTest("length check for input string values: nested in array of array")
     .excludeGlutenTest("length check for input string values: nested in array of struct")
     .excludeGlutenTest("length check for input string values: nested in array")
+    .excludeGlutenTest("length check for input string values: nested in map key")
+    .excludeGlutenTest("length check for input string values: nested in map value")
+    .excludeGlutenTest("length check for input string values: nested in both map key and value")
   enableSuite[GlutenFileSourceSQLInsertTestSuite]
     .exclude("SPARK-33474: Support typed literals as partition spec values")
     .exclude(

gluten-ut/spark34/src/test/scala/org/apache/gluten/utils/velox/VeloxTestSettings.scala

@@ -686,6 +686,16 @@ class VeloxTestSettings extends BackendTestSettings {
     .exclude("length check for input string values: nested in array of struct")
     .exclude("length check for input string values: nested in array of array")
     .exclude("length check for input string values: with implicit cast")
+    // Overridden.
+    .exclude("length check for input string values: nested in map key")
+    // Overridden.
+    .exclude("length check for input string values: nested in map value")
+    // Overridden.
+    .exclude("length check for input string values: nested in both map key and value")
+    // Overridden.
+    .exclude("SPARK-42611: check char/varchar length in reordered structs within map keys")
+    // Overridden.
+    .exclude("SPARK-42611: check char/varchar length in reordered structs within map values")
 
   enableSuite[GlutenColumnExpressionSuite]
     // Velox raise_error('errMsg') throws a velox_user_error exception with the message 'errMsg'.
@@ -729,6 +739,8 @@ class VeloxTestSettings extends BackendTestSettings {
     .exclude("map_zip_with function - map of primitive types")
     // Exception class different.
     .exclude("array_insert functions")
+    // Overridden.
+    .exclude("map with arrays")
   enableSuite[GlutenDataFrameHintSuite]
   enableSuite[GlutenDataFrameImplicitsSuite]
   enableSuite[GlutenDataFrameJoinSuite]

gluten-ut/spark34/src/test/scala/org/apache/spark/sql/GlutenCharVarcharTestSuite.scala

@@ -63,7 +63,8 @@ class GlutenFileSourceCharVarcharTestSuite
       typeName =>
         sql(s"CREATE TABLE t(c MAP<$typeName(5), STRING>) USING $format")
         val e = intercept[SparkException](sql("INSERT INTO t VALUES (map('123456', 'a'))"))
-        assert(e.getMessage.contains(ERROR_MESSAGE))
+        // Gluten exception differs from Spark
+        assert(e.getMessage.contains(VELOX_ERROR_MESSAGE))
     }
   }
 
@@ -74,7 +75,8 @@ class GlutenFileSourceCharVarcharTestSuite
         sql("INSERT INTO t VALUES (map('a', null))")
         checkAnswer(spark.table("t"), Row(Map("a" -> null)))
         val e = intercept[SparkException](sql("INSERT INTO t VALUES (map('a', '123456'))"))
-        assert(e.getMessage.contains(ERROR_MESSAGE))
+        // Gluten exception differs from Spark
+        assert(e.getMessage.contains(VELOX_ERROR_MESSAGE))
     }
   }
 
@@ -83,9 +85,11 @@ class GlutenFileSourceCharVarcharTestSuite
       typeName =>
         sql(s"CREATE TABLE t(c MAP<$typeName(5), $typeName(5)>) USING $format")
         val e1 = intercept[SparkException](sql("INSERT INTO t VALUES (map('123456', 'a'))"))
-        assert(e1.getMessage.contains(ERROR_MESSAGE))
+        // Gluten exception differs from Spark
+        assert(e1.getMessage.contains(VELOX_ERROR_MESSAGE))
         val e2 = intercept[SparkException](sql("INSERT INTO t VALUES (map('a', '123456'))"))
-        assert(e2.getMessage.contains(ERROR_MESSAGE))
+        // Gluten exception differs from Spark
+        assert(e2.getMessage.contains(VELOX_ERROR_MESSAGE))
     }
   }
 
@@ -301,4 +305,69 @@ class GlutenDSV2CharVarcharTestSuite extends DSV2CharVarcharTestSuite with Glute
         }
     }
   }
+
+  testGluten("length check for input string values: nested in map key") {
+    testTableWrite {
+      typeName =>
+        sql(s"CREATE TABLE t(c MAP<$typeName(5), STRING>) USING $format")
+        val e = intercept[SparkException](sql("INSERT INTO t VALUES (map('123456', 'a'))"))
+        // Gluten exception differs from Spark
+        assert(e.getMessage.contains(VELOX_ERROR_MESSAGE))
+    }
+  }
+
+  testGluten("length check for input string values: nested in map value") {
+    testTableWrite {
+      typeName =>
+        sql(s"CREATE TABLE t(c MAP<STRING, $typeName(5)>) USING $format")
+        sql("INSERT INTO t VALUES (map('a', null))")
+        checkAnswer(spark.table("t"), Row(Map("a" -> null)))
+        val e = intercept[SparkException](sql("INSERT INTO t VALUES (map('a', '123456'))"))
+        // Gluten exception differs from Spark
+        assert(e.getMessage.contains(VELOX_ERROR_MESSAGE))
+    }
+  }
+
+  testGluten("length check for input string values: nested in both map key and value") {
+    testTableWrite {
+      typeName =>
+        sql(s"CREATE TABLE t(c MAP<$typeName(5), $typeName(5)>) USING $format")
+        val e1 = intercept[SparkException](sql("INSERT INTO t VALUES (map('123456', 'a'))"))
+        // Gluten exception differs from Spark
+        assert(e1.getMessage.contains(VELOX_ERROR_MESSAGE))
+        val e2 = intercept[SparkException](sql("INSERT INTO t VALUES (map('a', '123456'))"))
+        // Gluten exception differs from Spark
+        assert(e2.getMessage.contains(VELOX_ERROR_MESSAGE))
+    }
+  }
+
+  testGluten("SPARK-42611: check char/varchar length in reordered structs within map keys") {
+    Seq("CHAR(5)", "VARCHAR(5)").foreach {
+      typ =>
+        withTable("t") {
+          sql(s"CREATE TABLE t(m MAP<STRUCT<n_c: $typ, n_i: INT>, INT>) USING $format")
+
+          val inputDF = sql("SELECT map(named_struct('n_i', 1, 'n_c', '123456'), 1) AS m")
+
+          val e = intercept[SparkException](inputDF.writeTo("t").append())
+          // Gluten exception differs from Spark
+          assert(e.getMessage.contains(VELOX_ERROR_MESSAGE))
+        }
+    }
+  }
+
+  testGluten("SPARK-42611: check char/varchar length in reordered structs within map values") {
+    Seq("CHAR(5)", "VARCHAR(5)").foreach {
+      typ =>
+        withTable("t") {
+          sql(s"CREATE TABLE t(m MAP<INT, STRUCT<n_c: $typ, n_i: INT>>) USING $format")
+
+          val inputDF = sql("SELECT map(1, named_struct('n_i', 1, 'n_c', '123456')) AS m")
+
+          val e = intercept[SparkException](inputDF.writeTo("t").append())
+          // Gluten exception differs from Spark
+          assert(e.getMessage.contains(VELOX_ERROR_MESSAGE))
+        }
+    }
+  }
 }

gluten-ut/spark34/src/test/scala/org/apache/spark/sql/GlutenDataFrameFunctionsSuite.scala

@@ -21,10 +21,55 @@ import org.apache.gluten.exception.GlutenException
 import org.apache.spark.SparkException
 import org.apache.spark.sql.functions._
 import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.types.{IntegerType, MapType, StringType}
 
 class GlutenDataFrameFunctionsSuite extends DataFrameFunctionsSuite with GlutenSQLTestsTrait {
   import testImplicits._
 
+  testGluten("map with arrays") {
+    val df1 = Seq((Seq(1, 2), Seq("a", "b"))).toDF("k", "v")
+    val expectedType = MapType(IntegerType, StringType, valueContainsNull = true)
+    val row = df1.select(map_from_arrays($"k", $"v")).first()
+    assert(row.schema(0).dataType === expectedType)
+    assert(row.getMap[Int, String](0) === Map(1 -> "a", 2 -> "b"))
+    checkAnswer(df1.select(map_from_arrays($"k", $"v")), Seq(Row(Map(1 -> "a", 2 -> "b"))))
+
+    val df2 = Seq((Seq(1, 2), Seq(null, "b"))).toDF("k", "v")
+    checkAnswer(df2.select(map_from_arrays($"k", $"v")), Seq(Row(Map(1 -> null, 2 -> "b"))))
+
+    val df3 = Seq((null, null)).toDF("k", "v")
+    checkAnswer(df3.select(map_from_arrays($"k", $"v")), Seq(Row(null)))
+
+    val df4 = Seq((1, "a")).toDF("k", "v")
+    checkError(
+      exception = intercept[AnalysisException] {
+        df4.select(map_from_arrays($"k", $"v"))
+      },
+      errorClass = "DATATYPE_MISMATCH.UNEXPECTED_INPUT_TYPE",
+      parameters = Map(
+        "sqlExpr" -> "\"map_from_arrays(k, v)\"",
+        "paramIndex" -> "1",
+        "requiredType" -> "\"ARRAY\"",
+        "inputSql" -> "\"k\"",
+        "inputType" -> "\"INT\""
+      )
+    )
+
+    val df5 = Seq((Seq("a", null), Seq(1, 2))).toDF("k", "v")
+    val e1 = intercept[SparkException] {
+      df5.select(map_from_arrays($"k", $"v")).collect
+    }
+    // Gluten exception differs from Spark
+    assert(e1.getCause.isInstanceOf[GlutenException])
+    assert(e1.getCause.getMessage.contains("Cannot use null as map key"))
+
+    val df6 = Seq((Seq(1, 2), Seq("a"))).toDF("k", "v")
+    val msg2 = intercept[Exception] {
+      df6.select(map_from_arrays($"k", $"v")).collect
+    }.getMessage
+    assert(msg2.contains("The key array and value array of MapData must have the same length"))
+  }
+
   testGluten("map_zip_with function - map of primitive types") {
     val df = Seq(
       (Map(8 -> 6L, 3 -> 5L, 6 -> 2L), Map[Integer, Integer]((6, 4), (8, 2), (3, 2))),

gluten-ut/spark35/src/test/scala/org/apache/gluten/utils/clickhouse/ClickHouseTestSettings.scala

@@ -463,6 +463,13 @@ class ClickHouseTestSettings extends BackendTestSettings {
     .excludeGlutenTest("length check for input string values: with implicit cast")
     .excludeGlutenTest("char/varchar type values length check: partitioned columns of other types")
     .excludeGlutenTest("SPARK-42611: check char/varchar length in reordered structs within arrays")
+    .excludeGlutenTest("length check for input string values: nested in map key")
+    .excludeGlutenTest("length check for input string values: nested in map value")
+    .excludeGlutenTest("length check for input string values: nested in both map key and value")
+    .excludeGlutenTest(
+      "SPARK-42611: check char/varchar length in reordered structs within map keys")
+    .excludeGlutenTest(
+      "SPARK-42611: check char/varchar length in reordered structs within map values")
   enableSuite[GlutenDSV2SQLInsertTestSuite]
   enableSuite[GlutenDataFrameAggregateSuite]
     // Test for vanilla spark codegen, not apply for Gluten
@@ -504,6 +511,7 @@ class ClickHouseTestSettings extends BackendTestSettings {
     // Rewrite this test because Velox sorts rows by key for primitive data types, which disrupts the original row sequence.
     .includeCH("map_zip_with function - map of primitive types")
     .excludeCH("map with arrays")
+    .excludeGlutenTest("map with arrays")
     .excludeCH("flatten function")
     .excludeCH("SPARK-41233: array prepend")
     .excludeCH("array_insert functions")
@@ -882,6 +890,9 @@ class ClickHouseTestSettings extends BackendTestSettings {
     .excludeGlutenTest("length check for input string values: nested in array")
     .excludeGlutenTest("length check for input string values: nested in array of struct")
     .excludeGlutenTest("length check for input string values: nested in array of array")
+    .excludeGlutenTest("length check for input string values: nested in map key")
+    .excludeGlutenTest("length check for input string values: nested in map value")
+    .excludeGlutenTest("length check for input string values: nested in both map key and value")
   enableSuite[GlutenFileSourceCustomMetadataStructSuite]
   enableSuite[GlutenFileSourceSQLInsertTestSuite]
     .excludeCH("SPARK-33474: Support typed literals as partition spec values")
@@ -2034,6 +2045,9 @@ class ClickHouseTestSettings extends BackendTestSettings {
   enableSuite[GlutenResolvedDataSourceSuite]
   enableSuite[GlutenReuseExchangeAndSubquerySuite]
   enableSuite[GlutenRuntimeNullChecksV2Writes]
+    .excludeGlutenTest("NOT NULL checks for nullable map with required values (byName)")
+    .excludeGlutenTest("NOT NULL checks for nullable map with required values (byPosition)")
+    .excludeGlutenTest("NOT NULL checks for fields inside nullable maps (byPosition)")
   enableSuite[GlutenSQLAggregateFunctionSuite]
     .excludeGlutenTest("Return NaN or null when dividing by zero")
   enableSuite[GlutenSQLQuerySuite]

gluten-ut/spark35/src/test/scala/org/apache/gluten/utils/velox/VeloxTestSettings.scala

@@ -646,6 +646,16 @@ class VeloxTestSettings extends BackendTestSettings {
     .exclude("length check for input string values: with implicit cast")
     .exclude("char/varchar type values length check: partitioned columns of other types")
     .exclude("SPARK-42611: check char/varchar length in reordered structs within arrays")
+    // Overridden.
+    .exclude("length check for input string values: nested in map key")
+    // Overridden.
+    .exclude("length check for input string values: nested in map value")
+    // Overridden.
+    .exclude("length check for input string values: nested in both map key and value")
+    // Overridden.
+    .exclude("SPARK-42611: check char/varchar length in reordered structs within map keys")
+    // Overridden.
+    .exclude("SPARK-42611: check char/varchar length in reordered structs within map values")
   enableSuite[GlutenColumnExpressionSuite]
     // Velox raise_error('errMsg') throws a velox_user_error exception with the message 'errMsg'.
     // The final caught Spark exception's getCause().getMessage() contains 'errMsg' but does not
@@ -688,6 +698,8 @@ class VeloxTestSettings extends BackendTestSettings {
     .exclude("aggregate function - array for non-primitive type")
     // Rewrite this test because Velox sorts rows by key for primitive data types, which disrupts the original row sequence.
     .exclude("map_zip_with function - map of primitive types")
+    // Overridden.
+    .exclude("map with arrays")
   enableSuite[GlutenDataFrameHintSuite]
   enableSuite[GlutenDataFrameImplicitsSuite]
   enableSuite[GlutenDataFrameJoinSuite]
@@ -945,6 +957,12 @@ class VeloxTestSettings extends BackendTestSettings {
     .exclude("NOT NULL checks for nested structs, arrays, maps (byPosition)")
     .exclude("NOT NULL checks for nullable array with required element (byPosition)")
     .exclude("not null checks for fields inside nullable array (byPosition)")
+    // Overridden.
+    .exclude("NOT NULL checks for nullable map with required values (byName)")
+    // Overridden.
+    .exclude("NOT NULL checks for nullable map with required values (byPosition)")
+    // Overridden.
+    .exclude("NOT NULL checks for fields inside nullable maps (byPosition)")
   enableSuite[GlutenTableOptionsConstantFoldingSuite]
   enableSuite[GlutenDeltaBasedMergeIntoTableSuite]
   enableSuite[GlutenDeltaBasedMergeIntoTableUpdateAsDeleteAndInsertSuite]