This is an automated email from the ASF dual-hosted git repository.
philo-he pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/gluten.git
The following commit(s) were added to refs/heads/main by this push:
new b2b84dbc4d [GLUTEN-6101][VL] Enable map_from_arrays function (#12976)
b2b84dbc4d is described below
commit b2b84dbc4d82cf2fc00691234b6e9276d9b2ea78
Author: Pedrum Jalali <[email protected]>
AuthorDate: Tue Sep 15 20:36:48 2026 -0700
[GLUTEN-6101][VL] Enable map_from_arrays function (#12976)
---
.../gluten/execution/MiscOperatorSuite.scala | 10 --
.../functions/ScalarFunctionsValidateSuite.scala | 36 ++++++
.../substrait/SubstraitToVeloxPlanValidator.cc | 2 +-
docs/velox-backend-scalar-function-support.md | 4 +-
.../utils/clickhouse/ClickHouseTestSettings.scala | 11 ++
.../gluten/utils/velox/VeloxTestSettings.scala | 12 ++
.../spark/sql/GlutenCharVarcharTestSuite.scala | 77 +++++++++++-
.../spark/sql/GlutenDataFrameFunctionsSuite.scala | 45 +++++++
.../utils/clickhouse/ClickHouseTestSettings.scala | 14 +++
.../gluten/utils/velox/VeloxTestSettings.scala | 18 +++
.../spark/sql/GlutenCharVarcharTestSuite.scala | 77 +++++++++++-
.../spark/sql/GlutenDataFrameFunctionsSuite.scala | 48 ++++++++
.../sql/GlutenRuntimeNullChecksV2Writes.scala | 135 +++++++++++++++++++-
.../utils/clickhouse/ClickHouseTestSettings.scala | 4 +
.../gluten/utils/velox/VeloxTestSettings.scala | 8 ++
.../spark/sql/GlutenDataFrameFunctionsSuite.scala | 50 ++++++++
.../sql/GlutenRuntimeNullChecksV2Writes.scala | 137 ++++++++++++++++++++-
.../utils/clickhouse/ClickHouseTestSettings.scala | 4 +
.../gluten/utils/velox/VeloxTestSettings.scala | 8 ++
.../spark/sql/GlutenDataFrameFunctionsSuite.scala | 50 ++++++++
.../sql/GlutenRuntimeNullChecksV2Writes.scala | 134 +++++++++++++++++++-
21 files changed, 860 insertions(+), 24 deletions(-)
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 948df8f8a3..7b43ad18e5 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 557b86a17d..78051d4539 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 618ffa1c0c..7ef0e807f1 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 744b4365d6..f4abc7a962 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 3493681aad..5203a075f9 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 6d43178512..2346d5f7d3 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 4f0d826f57..5930d7ce7b 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 ebe1fc50fc..a55403c356 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 49d637681b..64bf921217 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 d26fcbf494..72ef076d04 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 689946547d..af23dc8a1e 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 e64f760ab5..d19bb06649 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 abd997bea8..0ef7eb2da9 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 90ad2f598d..decdc1990d 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 10f467378e..0ae8965619 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 0d2caf6100..40ac1edd60 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 abd997bea8..d0cd89b643 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 90ad2f598d..decdc1990d 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 5b2757a95c..0890acd58f 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 0d2caf6100..40ac1edd60 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 abd997bea8..642ad786fb 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"))
+ }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
