diff --git a/backends-velox/src/test/scala/org/apache/gluten/execution/MiscOperatorSuite.scala b/backends-velox/src/test/scala/org/apache/gluten/execution/MiscOperatorSuite.scala
index 948df8f8a31..7b43ad18e55 100644
--- a/backends-velox/src/test/scala/org/apache/gluten/execution/MiscOperatorSuite.scala
+++ b/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);")
diff --git a/backends-velox/src/test/scala/org/apache/gluten/functions/ScalarFunctionsValidateSuite.scala b/backends-velox/src/test/scala/org/apache/gluten/functions/ScalarFunctionsValidateSuite.scala
index 557b86a17dd..78051d4539d 100644
--- a/backends-velox/src/test/scala/org/apache/gluten/functions/ScalarFunctionsValidateSuite.scala
+++ b/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) {
diff --git a/cpp/velox/substrait/SubstraitToVeloxPlanValidator.cc b/cpp/velox/substrait/SubstraitToVeloxPlanValidator.cc
index 618ffa1c0ce..7ef0e807f1a 100644
--- a/cpp/velox/substrait/SubstraitToVeloxPlanValidator.cc
+++ b/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(
diff --git a/docs/velox-backend-scalar-function-support.md b/docs/velox-backend-scalar-function-support.md
index 744b4365d65..f4abc7a9627 100644
--- a/docs/velox-backend-scalar-function-support.md
+++ b/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        |                                                                             |
diff --git a/gluten-ut/spark34/src/test/scala/org/apache/gluten/utils/clickhouse/ClickHouseTestSettings.scala b/gluten-ut/spark34/src/test/scala/org/apache/gluten/utils/clickhouse/ClickHouseTestSettings.scala
index 3493681aad5..5203a075f9c 100644
--- a/gluten-ut/spark34/src/test/scala/org/apache/gluten/utils/clickhouse/ClickHouseTestSettings.scala
+++ b/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(
diff --git a/gluten-ut/spark34/src/test/scala/org/apache/gluten/utils/velox/VeloxTestSettings.scala b/gluten-ut/spark34/src/test/scala/org/apache/gluten/utils/velox/VeloxTestSettings.scala
index 6d431785125..2346d5f7d38 100644
--- a/gluten-ut/spark34/src/test/scala/org/apache/gluten/utils/velox/VeloxTestSettings.scala
+++ b/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]
diff --git a/gluten-ut/spark34/src/test/scala/org/apache/spark/sql/GlutenCharVarcharTestSuite.scala b/gluten-ut/spark34/src/test/scala/org/apache/spark/sql/GlutenCharVarcharTestSuite.scala
index 4f0d826f57b..5930d7ce7b6 100644
--- a/gluten-ut/spark34/src/test/scala/org/apache/spark/sql/GlutenCharVarcharTestSuite.scala
+++ b/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))
+        }
+    }
+  }
 }
diff --git a/gluten-ut/spark34/src/test/scala/org/apache/spark/sql/GlutenDataFrameFunctionsSuite.scala b/gluten-ut/spark34/src/test/scala/org/apache/spark/sql/GlutenDataFrameFunctionsSuite.scala
index ebe1fc50fc5..a55403c356e 100644
--- a/gluten-ut/spark34/src/test/scala/org/apache/spark/sql/GlutenDataFrameFunctionsSuite.scala
+++ b/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))),
diff --git a/gluten-ut/spark35/src/test/scala/org/apache/gluten/utils/clickhouse/ClickHouseTestSettings.scala b/gluten-ut/spark35/src/test/scala/org/apache/gluten/utils/clickhouse/ClickHouseTestSettings.scala
index 49d637681b5..64bf9212178 100644
--- a/gluten-ut/spark35/src/test/scala/org/apache/gluten/utils/clickhouse/ClickHouseTestSettings.scala
+++ b/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]
diff --git a/gluten-ut/spark35/src/test/scala/org/apache/gluten/utils/velox/VeloxTestSettings.scala b/gluten-ut/spark35/src/test/scala/org/apache/gluten/utils/velox/VeloxTestSettings.scala
index d26fcbf494e..72ef076d04d 100644
--- a/gluten-ut/spark35/src/test/scala/org/apache/gluten/utils/velox/VeloxTestSettings.scala
+++ b/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]
diff --git a/gluten-ut/spark35/src/test/scala/org/apache/spark/sql/GlutenCharVarcharTestSuite.scala b/gluten-ut/spark35/src/test/scala/org/apache/spark/sql/GlutenCharVarcharTestSuite.scala
index 689946547d1..af23dc8a1e1 100644
--- a/gluten-ut/spark35/src/test/scala/org/apache/spark/sql/GlutenCharVarcharTestSuite.scala
+++ b/gluten-ut/spark35/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))
     }
   }
 
@@ -300,4 +304,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))
+        }
+    }
+  }
 }
diff --git a/gluten-ut/spark35/src/test/scala/org/apache/spark/sql/GlutenDataFrameFunctionsSuite.scala b/gluten-ut/spark35/src/test/scala/org/apache/spark/sql/GlutenDataFrameFunctionsSuite.scala
index e64f760ab55..d19bb066490 100644
--- a/gluten-ut/spark35/src/test/scala/org/apache/spark/sql/GlutenDataFrameFunctionsSuite.scala
+++ b/gluten-ut/spark35/src/test/scala/org/apache/spark/sql/GlutenDataFrameFunctionsSuite.scala
@@ -16,11 +16,59 @@
  */
 package org.apache.spark.sql
 
+import org.apache.gluten.exception.GlutenException
+
+import org.apache.spark.SparkException
 import org.apache.spark.sql.functions._
+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))),
diff --git a/gluten-ut/spark35/src/test/scala/org/apache/spark/sql/GlutenRuntimeNullChecksV2Writes.scala b/gluten-ut/spark35/src/test/scala/org/apache/spark/sql/GlutenRuntimeNullChecksV2Writes.scala
index abd997bea82..0ef7eb2da9c 100644
--- a/gluten-ut/spark35/src/test/scala/org/apache/spark/sql/GlutenRuntimeNullChecksV2Writes.scala
+++ b/gluten-ut/spark35/src/test/scala/org/apache/spark/sql/GlutenRuntimeNullChecksV2Writes.scala
@@ -16,4 +16,137 @@
  */
 package org.apache.spark.sql
 
-class GlutenRuntimeNullChecksV2Writes extends RuntimeNullChecksV2Writes with GlutenSQLTestsTrait {}
+import org.apache.spark.SparkException
+import org.apache.spark.sql.connector.catalog.{Column => ColumnV2, Identifier}
+import org.apache.spark.sql.connector.expressions.Transform
+import org.apache.spark.sql.types.{IntegerType, MapType, StructType}
+
+import java.util.Collections
+
+class GlutenRuntimeNullChecksV2Writes extends RuntimeNullChecksV2Writes with GlutenSQLTestsTrait {
+
+  /**
+   * Shadows Spark's `assertNotNullException`, which is private and so cannot be reused.
+   *
+   * Spark requires the cause to be a `NullPointerException` and matches the offending column path
+   * against `colPath.mkString("\n", "\n", "\n")`. Velox raises a `VeloxUserError` that Gluten
+   * surfaces as a `SparkException`, carrying the same reason text but not Spark's column path
+   * formatting, so `colPath` is reported on failure rather than asserted. The reason match is case
+   * insensitive because Spark 3.5 words it "Null value appeared ..." and Spark 4.x "NULL value
+   * appeared ...".
+   */
+  private def assertNotNullException(e: SparkException, colPath: Seq[String]): Unit = {
+    val messages = Iterator
+      .iterate[Throwable](e)(_.getCause)
+      .takeWhile(_ != null)
+      .flatMap(t => Option(t.getMessage))
+      .mkString("\n")
+
+    assert(
+      messages.toLowerCase(java.util.Locale.ROOT).contains("value appeared in non-nullable field"),
+      s"expected a not-null violation for ${colPath.mkString(".")}, got:\n$messages"
+    )
+  }
+
+  testGluten("NOT NULL checks for nullable map with required values (byName)") {
+    checkNullableMapWithNonNullValues(byName = true)
+  }
+
+  testGluten("NOT NULL checks for nullable map with required values (byPosition)") {
+    checkNullableMapWithNonNullValues(byName = false)
+  }
+
+  private def checkNullableMapWithNonNullValues(byName: Boolean): Unit = {
+    withTable("t") {
+      catalog.createTable(
+        ident = Identifier.of(Array(), "t"),
+        columns = Array(
+          ColumnV2.create("i", IntegerType),
+          ColumnV2.create("m", MapType(IntegerType, IntegerType, valueContainsNull = false))),
+        partitions = Array.empty[Transform],
+        properties = Collections.emptyMap[String, String]
+      )
+
+      if (byName) {
+        val inputDF = sql("SELECT 1 AS i, null AS m")
+        inputDF.writeTo("t").append()
+      } else {
+        sql("INSERT INTO t VALUES (1 AS i, null AS m)")
+      }
+      checkAnswer(spark.table("t"), Row(1, null))
+
+      val e = intercept[SparkException] {
+        if (byName) {
+          val inputDF = sql("SELECT 1 AS i, map(1, null) AS m")
+          inputDF.writeTo("t").append()
+        } else {
+          sql("INSERT INTO t VALUES (1 AS i, map(1, null) AS m)")
+        }
+      }
+      assertNotNullException(e, Seq("m", "value"))
+    }
+  }
+
+  /** Only the byPosition case is overridden. */
+  testGluten("NOT NULL checks for fields inside nullable maps (byPosition)") {
+    checkNotNullFieldsInsideNullableMap(byName = false)
+  }
+
+  private def checkNotNullFieldsInsideNullableMap(byName: Boolean): Unit = {
+    withTable("t") {
+      val structType = new StructType().add("x", "int", nullable = false).add("y", "int")
+      catalog.createTable(
+        ident = Identifier.of(Array(), "t"),
+        columns = Array(
+          ColumnV2.create("i", IntegerType),
+          ColumnV2.create("m", MapType(structType, structType, valueContainsNull = true))),
+        partitions = Array.empty[Transform],
+        properties = Collections.emptyMap[String, String]
+      )
+
+      if (byName) {
+        val inputDF = sql("SELECT 1 AS i, map(named_struct('x', 1, 'y', 1), null) AS m")
+        inputDF.writeTo("t").append()
+      } else {
+        sql("INSERT INTO t VALUES (1 AS i, map(named_struct('x', 1, 'y', 1), null) AS m)")
+      }
+      checkAnswer(spark.table("t"), Row(1, Map(Row(1, 1) -> null)))
+
+      val e1 = intercept[SparkException] {
+        if (byName) {
+          val inputDF = sql(
+            s"""SELECT
+               | 1 AS i,
+               | map(named_struct('x', null, 'y', 1), null) AS m
+             """.stripMargin)
+          inputDF.writeTo("t").append()
+        } else {
+          sql(
+            s"""INSERT INTO t VALUES (
+               | 1 AS i,
+               | map(named_struct('x', null, 'y', 1), null) AS m)
+             """.stripMargin)
+        }
+      }
+      assertNotNullException(e1, Seq("m", "key", "x"))
+
+      val e2 = intercept[SparkException] {
+        if (byName) {
+          val inputDF = sql(
+            s"""SELECT
+               | 1 AS i,
+               | map(named_struct('x', 1, 'y', 1), named_struct('x', null, 'y', 1)) AS m
+             """.stripMargin)
+          inputDF.writeTo("t").append()
+        } else {
+          sql(
+            s"""INSERT INTO t VALUES (
+               | 1 AS i,
+               | map(named_struct('x', 1, 'y', 1), named_struct('x', null, 'y', 1)) AS m)
+             """.stripMargin)
+        }
+      }
+      assertNotNullException(e2, Seq("m", "value", "x"))
+    }
+  }
+}
diff --git a/gluten-ut/spark40/src/test/scala/org/apache/gluten/utils/clickhouse/ClickHouseTestSettings.scala b/gluten-ut/spark40/src/test/scala/org/apache/gluten/utils/clickhouse/ClickHouseTestSettings.scala
index 90ad2f598d3..decdc1990df 100644
--- a/gluten-ut/spark40/src/test/scala/org/apache/gluten/utils/clickhouse/ClickHouseTestSettings.scala
+++ b/gluten-ut/spark40/src/test/scala/org/apache/gluten/utils/clickhouse/ClickHouseTestSettings.scala
@@ -489,6 +489,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")
@@ -2002,6 +2003,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]
diff --git a/gluten-ut/spark40/src/test/scala/org/apache/gluten/utils/velox/VeloxTestSettings.scala b/gluten-ut/spark40/src/test/scala/org/apache/gluten/utils/velox/VeloxTestSettings.scala
index 10f467378e1..0ae89656191 100644
--- a/gluten-ut/spark40/src/test/scala/org/apache/gluten/utils/velox/VeloxTestSettings.scala
+++ b/gluten-ut/spark40/src/test/scala/org/apache/gluten/utils/velox/VeloxTestSettings.scala
@@ -983,6 +983,8 @@ class VeloxTestSettings extends BackendTestSettings {
     // Vanilla spark throw SparkRuntimeException, gluten throw SparkException.
     .exclude("map_concat function")
     .exclude("transform keys function - primitive data types")
+    // Overridden.
+    .exclude("map with arrays")
   enableSuite[GlutenDataFrameHintSuite]
   enableSuite[GlutenDataFrameImplicitsSuite]
   enableSuite[GlutenDataFrameJoinSuite]
@@ -1227,6 +1229,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]
     // Replaced by Gluten versions that handle wrapped exceptions
diff --git a/gluten-ut/spark40/src/test/scala/org/apache/spark/sql/GlutenDataFrameFunctionsSuite.scala b/gluten-ut/spark40/src/test/scala/org/apache/spark/sql/GlutenDataFrameFunctionsSuite.scala
index 0d2caf6100d..40ac1edd601 100644
--- a/gluten-ut/spark40/src/test/scala/org/apache/spark/sql/GlutenDataFrameFunctionsSuite.scala
+++ b/gluten-ut/spark40/src/test/scala/org/apache/spark/sql/GlutenDataFrameFunctionsSuite.scala
@@ -16,6 +16,8 @@
  */
 package org.apache.spark.sql
 
+import org.apache.gluten.exception.GlutenException
+
 import org.apache.spark.SparkException
 import org.apache.spark.sql.functions._
 import org.apache.spark.sql.internal.SQLConf
@@ -24,6 +26,54 @@ import org.apache.spark.sql.types.{IntegerType, MapType, StringType, StructField
 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"))
+      },
+      condition = "DATATYPE_MISMATCH.UNEXPECTED_INPUT_TYPE",
+      parameters = Map(
+        "sqlExpr" -> "\"map_from_arrays(k, v)\"",
+        "paramIndex" -> "first",
+        "requiredType" -> "\"ARRAY\"",
+        "inputSql" -> "\"k\"",
+        "inputType" -> "\"INT\""
+      ),
+      queryContext = Array(
+        ExpectedContext(
+          fragment = "map_from_arrays",
+          callSitePattern = getCurrentClassCallSitePattern))
+    )
+
+    val df5 = Seq((Seq("a", null), Seq(1, 2))).toDF("k", "v")
+    // Gluten exception differs from Spark
+    val e1 = intercept[SparkException] {
+      df5.select(map_from_arrays($"k", $"v")).collect()
+    }
+    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))),
diff --git a/gluten-ut/spark40/src/test/scala/org/apache/spark/sql/GlutenRuntimeNullChecksV2Writes.scala b/gluten-ut/spark40/src/test/scala/org/apache/spark/sql/GlutenRuntimeNullChecksV2Writes.scala
index abd997bea82..d0cd89b6439 100644
--- a/gluten-ut/spark40/src/test/scala/org/apache/spark/sql/GlutenRuntimeNullChecksV2Writes.scala
+++ b/gluten-ut/spark40/src/test/scala/org/apache/spark/sql/GlutenRuntimeNullChecksV2Writes.scala
@@ -16,4 +16,139 @@
  */
 package org.apache.spark.sql
 
-class GlutenRuntimeNullChecksV2Writes extends RuntimeNullChecksV2Writes with GlutenSQLTestsTrait {}
+import org.apache.spark.SparkException
+import org.apache.spark.sql.connector.catalog.{Column => ColumnV2, Identifier}
+import org.apache.spark.sql.connector.expressions.Transform
+import org.apache.spark.sql.types.{IntegerType, MapType, StructType}
+
+import java.util.Collections
+
+class GlutenRuntimeNullChecksV2Writes extends RuntimeNullChecksV2Writes with GlutenSQLTestsTrait {
+
+  /**
+   * Shadows Spark's `assertNotNullException`, which is private and so cannot be reused.
+   *
+   * Spark expects a `SparkRuntimeException` carrying `NOT_NULL_ASSERT_VIOLATION`, asserts nothing
+   * else, and never reads `colPath`. Velox raises a `VeloxUserError` that Gluten surfaces as a
+   * `SparkException`, which carries the same reason text but not the error condition, so this
+   * asserts the reason and reports `colPath` on failure. The reason match is case insensitive
+   * because Spark 3.5 words it "Null value appeared ..." and Spark 4.x "NULL value appeared ...".
+   */
+  private def assertNotNullException(e: SparkException, colPath: Seq[String]): Unit = {
+    val messages = Iterator
+      .iterate[Throwable](e)(_.getCause)
+      .takeWhile(_ != null)
+      .flatMap(t => Option(t.getMessage))
+      .mkString("\n")
+
+    assert(
+      messages.toLowerCase(java.util.Locale.ROOT).contains("value appeared in non-nullable field"),
+      s"expected a not-null violation for ${colPath.mkString(".")}, got:\n$messages"
+    )
+  }
+
+  testGluten("NOT NULL checks for nullable map with required values (byName)") {
+    checkNullableMapWithNonNullValues(byName = true)
+  }
+
+  testGluten("NOT NULL checks for nullable map with required values (byPosition)") {
+    checkNullableMapWithNonNullValues(byName = false)
+  }
+
+  private def checkNullableMapWithNonNullValues(byName: Boolean): Unit = {
+    withTable("t") {
+      catalog.createTable(
+        ident = Identifier.of(Array(), "t"),
+        columns = Array(
+          ColumnV2.create("i", IntegerType),
+          ColumnV2.create("m", MapType(IntegerType, IntegerType, valueContainsNull = false))),
+        partitions = Array.empty[Transform],
+        properties = Collections.emptyMap[String, String]
+      )
+
+      if (byName) {
+        val inputDF = sql("SELECT 1 AS i, null AS m")
+        inputDF.writeTo("t").append()
+      } else {
+        sql("INSERT INTO t VALUES (1 AS i, null AS m)")
+      }
+      checkAnswer(spark.table("t"), Row(1, null))
+
+      // Gluten exception differs from Spark
+      val e = intercept[SparkException] {
+        if (byName) {
+          val inputDF = sql("SELECT 1 AS i, map(1, null) AS m")
+          inputDF.writeTo("t").append()
+        } else {
+          sql("INSERT INTO t VALUES (1 AS i, map(1, null) AS m)")
+        }
+      }
+      assertNotNullException(e, Seq("m", "value"))
+    }
+  }
+
+  /** Only the byPosition case is overridden. */
+  testGluten("NOT NULL checks for fields inside nullable maps (byPosition)") {
+    checkNotNullFieldsInsideNullableMap(byName = false)
+  }
+
+  private def checkNotNullFieldsInsideNullableMap(byName: Boolean): Unit = {
+    withTable("t") {
+      val structType = new StructType().add("x", "int", nullable = false).add("y", "int")
+      catalog.createTable(
+        ident = Identifier.of(Array(), "t"),
+        columns = Array(
+          ColumnV2.create("i", IntegerType),
+          ColumnV2.create("m", MapType(structType, structType, valueContainsNull = true))),
+        partitions = Array.empty[Transform],
+        properties = Collections.emptyMap[String, String]
+      )
+
+      if (byName) {
+        val inputDF = sql("SELECT 1 AS i, map(named_struct('x', 1, 'y', 1), null) AS m")
+        inputDF.writeTo("t").append()
+      } else {
+        sql("INSERT INTO t VALUES (1 AS i, map(named_struct('x', 1, 'y', 1), null) AS m)")
+      }
+      checkAnswer(spark.table("t"), Row(1, Map(Row(1, 1) -> null)))
+
+      // Gluten exception differs from Spark
+      val e1 = intercept[SparkException] {
+        if (byName) {
+          val inputDF = sql(
+            s"""SELECT
+               | 1 AS i,
+               | map(named_struct('x', null, 'y', 1), null) AS m
+             """.stripMargin)
+          inputDF.writeTo("t").append()
+        } else {
+          sql(
+            s"""INSERT INTO t VALUES (
+               | 1 AS i,
+               | map(named_struct('x', null, 'y', 1), null) AS m)
+             """.stripMargin)
+        }
+      }
+      assertNotNullException(e1, Seq("m", "key", "x"))
+
+      // Gluten exception differs from Spark
+      val e2 = intercept[SparkException] {
+        if (byName) {
+          val inputDF = sql(
+            s"""SELECT
+               | 1 AS i,
+               | map(named_struct('x', 1, 'y', 1), named_struct('x', null, 'y', 1)) AS m
+             """.stripMargin)
+          inputDF.writeTo("t").append()
+        } else {
+          sql(
+            s"""INSERT INTO t VALUES (
+               | 1 AS i,
+               | map(named_struct('x', 1, 'y', 1), named_struct('x', null, 'y', 1)) AS m)
+             """.stripMargin)
+        }
+      }
+      assertNotNullException(e2, Seq("m", "value", "x"))
+    }
+  }
+}
diff --git a/gluten-ut/spark41/src/test/scala/org/apache/gluten/utils/clickhouse/ClickHouseTestSettings.scala b/gluten-ut/spark41/src/test/scala/org/apache/gluten/utils/clickhouse/ClickHouseTestSettings.scala
index 90ad2f598d3..decdc1990df 100644
--- a/gluten-ut/spark41/src/test/scala/org/apache/gluten/utils/clickhouse/ClickHouseTestSettings.scala
+++ b/gluten-ut/spark41/src/test/scala/org/apache/gluten/utils/clickhouse/ClickHouseTestSettings.scala
@@ -489,6 +489,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")
@@ -2002,6 +2003,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]
diff --git a/gluten-ut/spark41/src/test/scala/org/apache/gluten/utils/velox/VeloxTestSettings.scala b/gluten-ut/spark41/src/test/scala/org/apache/gluten/utils/velox/VeloxTestSettings.scala
index 5b2757a95c1..0890acd58fb 100644
--- a/gluten-ut/spark41/src/test/scala/org/apache/gluten/utils/velox/VeloxTestSettings.scala
+++ b/gluten-ut/spark41/src/test/scala/org/apache/gluten/utils/velox/VeloxTestSettings.scala
@@ -967,6 +967,8 @@ class VeloxTestSettings extends BackendTestSettings {
     // Vanilla spark throw SparkRuntimeException, gluten throw SparkException.
     .exclude("map_concat function")
     .exclude("transform keys function - primitive data types")
+    // Overridden.
+    .exclude("map with arrays")
   enableSuite[GlutenDataFrameHintSuite]
   enableSuite[GlutenDataFrameImplicitsSuite]
   enableSuite[GlutenDataFrameJoinSuite]
@@ -1224,6 +1226,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]
     // Replaced by Gluten versions that handle wrapped exceptions
diff --git a/gluten-ut/spark41/src/test/scala/org/apache/spark/sql/GlutenDataFrameFunctionsSuite.scala b/gluten-ut/spark41/src/test/scala/org/apache/spark/sql/GlutenDataFrameFunctionsSuite.scala
index 0d2caf6100d..40ac1edd601 100644
--- a/gluten-ut/spark41/src/test/scala/org/apache/spark/sql/GlutenDataFrameFunctionsSuite.scala
+++ b/gluten-ut/spark41/src/test/scala/org/apache/spark/sql/GlutenDataFrameFunctionsSuite.scala
@@ -16,6 +16,8 @@
  */
 package org.apache.spark.sql
 
+import org.apache.gluten.exception.GlutenException
+
 import org.apache.spark.SparkException
 import org.apache.spark.sql.functions._
 import org.apache.spark.sql.internal.SQLConf
@@ -24,6 +26,54 @@ import org.apache.spark.sql.types.{IntegerType, MapType, StringType, StructField
 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"))
+      },
+      condition = "DATATYPE_MISMATCH.UNEXPECTED_INPUT_TYPE",
+      parameters = Map(
+        "sqlExpr" -> "\"map_from_arrays(k, v)\"",
+        "paramIndex" -> "first",
+        "requiredType" -> "\"ARRAY\"",
+        "inputSql" -> "\"k\"",
+        "inputType" -> "\"INT\""
+      ),
+      queryContext = Array(
+        ExpectedContext(
+          fragment = "map_from_arrays",
+          callSitePattern = getCurrentClassCallSitePattern))
+    )
+
+    val df5 = Seq((Seq("a", null), Seq(1, 2))).toDF("k", "v")
+    // Gluten exception differs from Spark
+    val e1 = intercept[SparkException] {
+      df5.select(map_from_arrays($"k", $"v")).collect()
+    }
+    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))),
diff --git a/gluten-ut/spark41/src/test/scala/org/apache/spark/sql/GlutenRuntimeNullChecksV2Writes.scala b/gluten-ut/spark41/src/test/scala/org/apache/spark/sql/GlutenRuntimeNullChecksV2Writes.scala
index abd997bea82..642ad786fb6 100644
--- a/gluten-ut/spark41/src/test/scala/org/apache/spark/sql/GlutenRuntimeNullChecksV2Writes.scala
+++ b/gluten-ut/spark41/src/test/scala/org/apache/spark/sql/GlutenRuntimeNullChecksV2Writes.scala
@@ -16,4 +16,136 @@
  */
 package org.apache.spark.sql
 
-class GlutenRuntimeNullChecksV2Writes extends RuntimeNullChecksV2Writes with GlutenSQLTestsTrait {}
+import org.apache.spark.SparkException
+import org.apache.spark.sql.connector.catalog.{Column => ColumnV2, Identifier, TableInfo}
+import org.apache.spark.sql.types.{IntegerType, MapType, StructType}
+
+class GlutenRuntimeNullChecksV2Writes extends RuntimeNullChecksV2Writes with GlutenSQLTestsTrait {
+
+  /**
+   * Shadows Spark's `assertNotNullException`, which is private and so cannot be reused.
+   *
+   * Spark expects a `SparkRuntimeException` carrying `NOT_NULL_ASSERT_VIOLATION`, asserts nothing
+   * else, and never reads `colPath`. Velox raises a `VeloxUserError` that Gluten surfaces as a
+   * `SparkException`, which carries the same reason text but not the error condition, so this
+   * asserts the reason and reports `colPath` on failure. The reason match is case insensitive
+   * because Spark 3.5 words it "Null value appeared ..." and Spark 4.x "NULL value appeared ...".
+   */
+  private def assertNotNullException(e: SparkException, colPath: Seq[String]): Unit = {
+    val messages = Iterator
+      .iterate[Throwable](e)(_.getCause)
+      .takeWhile(_ != null)
+      .flatMap(t => Option(t.getMessage))
+      .mkString("\n")
+
+    assert(
+      messages.toLowerCase(java.util.Locale.ROOT).contains("value appeared in non-nullable field"),
+      s"expected a not-null violation for ${colPath.mkString(".")}, got:\n$messages"
+    )
+  }
+
+  testGluten("NOT NULL checks for nullable map with required values (byName)") {
+    checkNullableMapWithNonNullValues(byName = true)
+  }
+
+  testGluten("NOT NULL checks for nullable map with required values (byPosition)") {
+    checkNullableMapWithNonNullValues(byName = false)
+  }
+
+  private def checkNullableMapWithNonNullValues(byName: Boolean): Unit = {
+    withTable("t") {
+      val tableInfo = new TableInfo.Builder()
+        .withColumns(Array(
+          ColumnV2.create("i", IntegerType),
+          ColumnV2.create("m", MapType(IntegerType, IntegerType, valueContainsNull = false))))
+        .build()
+      catalog.createTable(
+        ident = Identifier.of(Array(), "t"),
+        tableInfo = tableInfo)
+
+      if (byName) {
+        val inputDF = sql("SELECT 1 AS i, null AS m")
+        inputDF.writeTo("t").append()
+      } else {
+        sql("INSERT INTO t VALUES (1 AS i, null AS m)")
+      }
+      checkAnswer(spark.table("t"), Row(1, null))
+
+      // Gluten exception differs from Spark
+      val e = intercept[SparkException] {
+        if (byName) {
+          val inputDF = sql("SELECT 1 AS i, map(1, null) AS m")
+          inputDF.writeTo("t").append()
+        } else {
+          sql("INSERT INTO t VALUES (1 AS i, map(1, null) AS m)")
+        }
+      }
+      assertNotNullException(e, Seq("m", "value"))
+    }
+  }
+
+  /** Only the byPosition case is overridden. */
+  testGluten("NOT NULL checks for fields inside nullable maps (byPosition)") {
+    checkNotNullFieldsInsideNullableMap(byName = false)
+  }
+
+  private def checkNotNullFieldsInsideNullableMap(byName: Boolean): Unit = {
+    withTable("t") {
+      val structType = new StructType().add("x", "int", nullable = false).add("y", "int")
+      val tableInfo = new TableInfo.Builder()
+        .withColumns(Array(
+          ColumnV2.create("i", IntegerType),
+          ColumnV2.create("m", MapType(structType, structType, valueContainsNull = true))))
+        .build()
+      catalog.createTable(
+        ident = Identifier.of(Array(), "t"),
+        tableInfo = tableInfo)
+
+      if (byName) {
+        val inputDF = sql("SELECT 1 AS i, map(named_struct('x', 1, 'y', 1), null) AS m")
+        inputDF.writeTo("t").append()
+      } else {
+        sql("INSERT INTO t VALUES (1 AS i, map(named_struct('x', 1, 'y', 1), null) AS m)")
+      }
+      checkAnswer(spark.table("t"), Row(1, Map(Row(1, 1) -> null)))
+
+      // Gluten exception differs from Spark
+      val e1 = intercept[SparkException] {
+        if (byName) {
+          val inputDF = sql(
+            s"""SELECT
+               | 1 AS i,
+               | map(named_struct('x', null, 'y', 1), null) AS m
+             """.stripMargin)
+          inputDF.writeTo("t").append()
+        } else {
+          sql(
+            s"""INSERT INTO t VALUES (
+               | 1 AS i,
+               | map(named_struct('x', null, 'y', 1), null) AS m)
+             """.stripMargin)
+        }
+      }
+      assertNotNullException(e1, Seq("m", "key", "x"))
+
+      // Gluten exception differs from Spark
+      val e2 = intercept[SparkException] {
+        if (byName) {
+          val inputDF = sql(
+            s"""SELECT
+               | 1 AS i,
+               | map(named_struct('x', 1, 'y', 1), named_struct('x', null, 'y', 1)) AS m
+             """.stripMargin)
+          inputDF.writeTo("t").append()
+        } else {
+          sql(
+            s"""INSERT INTO t VALUES (
+               | 1 AS i,
+               | map(named_struct('x', 1, 'y', 1), named_struct('x', null, 'y', 1)) AS m)
+             """.stripMargin)
+        }
+      }
+      assertNotNullException(e2, Seq("m", "value", "x"))
+    }
+  }
+}