From f153c66d667e4e058b73bc543d3f03d8edf68326 Mon Sep 17 00:00:00 2001 From: Pedrum Jalali Date: Thu, 10 Sep 2026 10:15:46 -0700 Subject: [PATCH] [GLUTEN-6101][VL] Enable map_from_arrays function MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## What changes are proposed in this pull request? Velox gained a Spark-compatible `map_from_arrays` in [facebookincubator/velox#18630](https://github.com/facebookincubator/velox/pull/18630). [#12968](https://github.com/apache/gluten/pull/12968) pulled [a6b9f7754](https://github.com/facebookincubator/velox/commit/a6b9f7754) into the Velox revision this repo pins, so the function can run natively here now. This PR removes it from [kBlackList](https://github.com/apache/gluten/blob/b77fdef08e4a733d1ed424bbf807b2904b92af86/cpp/velox/substrait/SubstraitToVeloxPlanValidator.cc#L59). [#2440](https://github.com/apache/gluten/pull/2440) added the denylist entry and [#6690](https://github.com/apache/gluten/pull/6690) moved it into [kBlackList](https://github.com/apache/gluten/blob/b77fdef08e4a733d1ed424bbf807b2904b92af86/cpp/velox/substrait/SubstraitToVeloxPlanValidator.cc#L59). [getNativeSessionConf](https://github.com/apache/gluten/blob/b77fdef08e4a733d1ed424bbf807b2904b92af86/gluten-substrait/src/main/scala/org/apache/gluten/config/GlutenConfig.scala#L555) already forwards [spark.sql.mapKeyDedupPolicy](https://github.com/apache/gluten/blob/b77fdef08e4a733d1ed424bbf807b2904b92af86/gluten-substrait/src/main/scala/org/apache/gluten/config/GlutenConfig.scala#L575) to Velox, and [ExpressionMappings](https://github.com/apache/gluten/blob/b77fdef08e4a733d1ed424bbf807b2904b92af86/gluten-substrait/src/main/scala/org/apache/gluten/expression/ExpressionMappings.scala#L268) already maps the expression. [docs/velox-backend-scalar-function-support.md](https://github.com/apache/gluten/blob/b77fdef08e4a733d1ed424bbf807b2904b92af86/docs/velox-backend-scalar-function-support.md) marks `map_from_arrays` as supported and bumps the fully-supported count from 246 to 247. Offloading the function moves who raises the error for a bad map, so several Spark UTs written while it fell back need adjusting. - [MiscOperatorSuite](https://github.com/apache/gluten/blob/b77fdef08e4a733d1ed424bbf807b2904b92af86/backends-velox/src/test/scala/org/apache/gluten/execution/MiscOperatorSuite.scala) used `map_from_arrays` as its example of an expression the backend cannot handle, so [ColumnarPartialProjectExec](https://github.com/apache/gluten/blob/b77fdef08e4a733d1ed424bbf807b2904b92af86/backends-velox/src/main/scala/org/apache/gluten/execution/ColumnarPartialProjectExec.scala#L141) no longer appears in its plan and that test is removed. The suite's `Blacklist expression can be handled by ColumnarPartialProject` test still asserts that `ColumnarPartialProjectExec` appears, so the operator keeps its coverage. - Spark builds the char/varchar write-side length check on a map column out of `MapFromArrays`, so that check now runs in Velox. - Velox reports a different message for it. Spark's: ``` Exceeds char/varchar type length limitation: 5 ``` and Velox's: ``` Exceeds allowed length limitation: 5 ``` - The three map overrides already in [GlutenFileSourceCharVarcharTestSuite](https://github.com/apache/gluten/blob/b77fdef08e4a733d1ed424bbf807b2904b92af86/gluten-ut/spark35/src/test/scala/org/apache/spark/sql/GlutenCharVarcharTestSuite.scala#L22) switch to the Velox message, matching how the array-nested siblings in that same suite already assert. - [GlutenDSV2CharVarcharTestSuite](https://github.com/apache/gluten/blob/b77fdef08e4a733d1ed424bbf807b2904b92af86/gluten-ut/spark35/src/test/scala/org/apache/spark/sql/GlutenCharVarcharTestSuite.scala#L188) gains overrides for its three map length checks and its two `SPARK-42611` map cases. - [GlutenDataFrameFunctionsSuite](https://github.com/apache/gluten/blob/b77fdef08e4a733d1ed424bbf807b2904b92af86/gluten-ut/spark35/src/test/scala/org/apache/spark/sql/GlutenDataFrameFunctionsSuite.scala#L21) gains an override of `map with arrays` in all four Spark versions. - Velox raises its own error for a null map key, so that override asserts Velox's `Cannot use null as map key!` reason rather than Spark's `NULL_MAP_KEY` error class, and the cause type it checks becomes `GlutenException`. No other assertion in the test changes. - [GlutenRuntimeNullChecksV2Writes](https://github.com/apache/gluten/blob/b77fdef08e4a733d1ed424bbf807b2904b92af86/gluten-ut/spark35/src/test/scala/org/apache/spark/sql/GlutenRuntimeNullChecksV2Writes.scala#L19) gains overrides of three of its four map tests in Spark 3.5, 4.0 and 4.1; `NOT NULL checks for fields inside nullable maps (byName)` is left unoverridden and unexcluded, so it keeps running the upstream assertions. Spark rewrites a map cast on the write path as `map_from_arrays(transform(map_keys(m), ...), transform(map_values(m), ...))`, so the `assert_not_null` inside it now runs in Velox's [AssertNotNullFunction](https://github.com/facebookincubator/velox/blob/a6b9f7754a7ba31a0fa26c7e52d51460a6b5d450/velox/functions/sparksql/AssertNotNull.cpp#L24). That raises the same `Null value appeared in non-nullable field` reason Spark uses, but as a `VeloxUserError` surfaced as a `SparkException` caused by a `GlutenException`. The overrides keep every step of the upstream test and relax only how the error is located, walking the cause chain. Two upstream assertions do not survive that, because Velox reports neither: on Spark 3.5 the column-path match against `colPath.mkString("\n", "\n", "\n")`, and on Spark 4.x the `NOT_NULL_ASSERT_VIOLATION` condition, which is that version's only assertion. `colPath` is reported in the failure message instead. - [VeloxTestSettings](https://github.com/apache/gluten/blob/b77fdef08e4a733d1ed424bbf807b2904b92af86/gluten-ut/spark35/src/test/scala/org/apache/gluten/utils/velox/VeloxTestSettings.scala) excludes each upstream test that an override replaces, in all four Spark versions. Every exclusion has a matching override, so no test is dropped. - [ClickHouseTestSettings](https://github.com/apache/gluten/blob/b77fdef08e4a733d1ed424bbf807b2904b92af86/gluten-ut/spark35/src/test/scala/org/apache/gluten/utils/clickhouse/ClickHouseTestSettings.scala#L46) excludes every Gluten override this PR adds or retargets, in all four Spark versions. The ClickHouse backend falls back to vanilla Spark for these writes, so it raises Spark's message. Only the `gluten-ut/spark34` and `gluten-ut/spark35` changes trigger the ClickHouse job, per [clickhouse_be_trigger.yml](https://github.com/apache/gluten/blob/b77fdef08e4a733d1ed424bbf807b2904b92af86/.github/workflows/clickhouse_be_trigger.yml#L37-L38). Fixes #6101 ## How was this patch tested? ### 🟢 Step 1: map_from_arrays runs in Velox end to end This script builds the Velox backend and the Spark 3.5 jars in the CI dev image and runs the query in a real Spark session. Save it as `verify-map-from-arrays.sh` and run it against a checkout of this branch. It needs a container runtime and leaves its build tree inside the container. ```bash #!/usr/bin/env bash # Verifies end to end that map_from_arrays is offloaded to Velox. # # ./verify-map-from-arrays.sh [path-to-gluten-checkout] # # Env: # DOCKER container runtime (default: docker) # NUM_THREADS build parallelism (default: nproc) # EXTRA_DOCKER_ARGS extra flags for your runtime, e.g. proxy or network settings set -euo pipefail GLUTEN_DIR=$(cd "${1:-$PWD}" && pwd) IMAGE=apache/gluten:centos-9-jdk8 DOCKER=${DOCKER:-docker} THREADS=${NUM_THREADS:-$(nproc)} "$DOCKER" pull "$IMAGE" # shellcheck disable=SC2086 "$DOCKER" run --rm ${EXTRA_DOCKER_ARGS:-} \ -v "$GLUTEN_DIR:/work/gluten" -w /work/gluten \ -e http_proxy -e https_proxy -e no_proxy \ -e NUM_THREADS="$THREADS" \ "$IMAGE" bash -eo pipefail -c ' ./dev/buildbundle-veloxbe.sh --run_setup_script=OFF --build_arrow=OFF --spark_version=3.5 JAR=$(ls /work/gluten/package/target/gluten-velox-bundle-spark3.5_*.jar) SPARK_HOME=/opt/shims/spark35/spark_home # range() keeps the arguments non-literal so Spark cannot constant-fold the call # away before the validator sees it. cat > /tmp/q.sql <<"SQL" CREATE OR REPLACE TEMPORARY VIEW t AS SELECT id AS k, CAST(id AS STRING) AS v FROM range(5); EXPLAIN SELECT map_from_arrays(array(k, k + 1), array(v, concat(v, "x"))) AS m FROM t; SELECT map_from_arrays(array(k, k + 1), array(v, concat(v, "x"))) AS m FROM t; SQL "$SPARK_HOME"/bin/spark-sql --master "local[2]" \ --conf spark.plugins=org.apache.gluten.GlutenPlugin \ --conf spark.driver.extraClassPath="$JAR" \ --conf spark.executor.extraClassPath="$JAR" \ --conf spark.memory.offHeap.enabled=true \ --conf spark.memory.offHeap.size=2g \ --conf spark.shuffle.manager=org.apache.spark.shuffle.sort.ColumnarShuffleManager \ -f /tmp/q.sql 2>&1 | tee /tmp/verify.out sed -n "/== Physical Plan ==/,/^$/p" /tmp/verify.out > /tmp/plan.out grep -q "ProjectExecTransformer \[map_from_arrays" /tmp/plan.out ! grep -qE "^\*?\([0-9]+\) Project \[map_from_arrays" /tmp/plan.out grep -q "{0:\"0\",1:\"0x\"}" /tmp/verify.out ' echo "PASS: map_from_arrays executed in Velox as a ProjectExecTransformer" ``` The complete `/tmp/plan.out` it wrote: ``` == Physical Plan == VeloxColumnarToRow +- ^(1) ProjectExecTransformer [map_from_arrays(array(k#11L, (k#11L + 1)), array(v#12, concat(v#12, x))) AS m#3] +- ^(1) ProjectExecTransformer [id#13L AS k#11L, cast(id#13L as string) AS v#12] +- ^(1) InputIteratorTransformer[id#13L] +- ArrowColumnarToVeloxColumnar +- OffloadArrowData +- ColumnarRange 0, 5, 1, 2, 5, [id#13L] ``` The complete set of rows the `SELECT` returned: ``` {0:"0",1:"0x"} {1:"1",2:"1x"} {2:"2",3:"2x"} {3:"3",4:"3x"} {4:"4",5:"4x"} ``` And the script's own verdict: ``` PASS: map_from_arrays executed in Velox as a ProjectExecTransformer ``` This step exercises the `kBlackList` change only. It also compiles `backends-velox`'s test sources, since `-DskipTests` skips test execution rather than test compilation; it compiles none of the `gluten-ut` modules, which need `-Pspark-ut`. --- ### 🟢 Step 2: Scala formatting ```bash ./dev/format-scala-code.sh check ``` Exit 0. The script runs `build/mvn -q spotless:check` across every profile, including `spark-3.4`, `spark-3.5`, `spark-4.0`, `spark-4.1`, `spark-ut` and `backends-velox`. --- ### 🟢 Step 3: C++ formatting ```bash python3 dev/check.py format main ``` ``` Ok : cpp/velox/substrait/SubstraitToVeloxPlanValidator.cc ``` Exit 0. Every other changed file is reported `Skip` because the checker covers `cpp,cc,h,inc,prolog` only. `python3 dev/check.py header main` was not run; the checker needs the `regex` module, which is unavailable in this environment. This change adds no file, so every license header in the diff is pre-existing and unmodified. --- ### 🟢 Step 4: every touched module test-compiles at this revision All five modules that this PR touches compile their test sources against the Spark version they target, so every override is type-checked: ```bash build/mvn -o test-compile -Pspark-3.4 -Pjava-17 -Pbackends-velox -Pspark-ut -pl gluten-ut/spark34 -am build/mvn -o test-compile -Pspark-3.5 -Pjava-17 -Pbackends-velox -Pspark-ut -pl gluten-ut/spark35 -am build/mvn -o test-compile -Pspark-4.0 -Pscala-2.13 -Pjava-17 -Pbackends-velox -Pspark-ut -pl gluten-ut/spark40 -am mvn -o test-compile -Pspark-4.1 -Pscala-2.13 -Pjava-17 -Pbackends-velox -Pspark-ut -pl gluten-ut/spark41 -am build/mvn -o test-compile -Pspark-3.5 -Pjava-17 -Pbackends-velox -pl backends-velox -am ``` All five exit 0. The `gluten-ut/spark41` one is invoked as `mvn` rather than `build/mvn` because it needs Maven >= 3.8.1 for `scala-maven-plugin:4.9.5`, and it was run against Maven 3.9.9 directly; the version this repo declares in `` is 3.9.16. That module's `GlutenRuntimeNullChecksV2Writes` override uses `TableInfo`, which exists only in Spark 4.1, so this is the check that confirms it resolves. ### ⚠️ Step 5: suite execution — covered by CI only Running the suites needs `libgluten.so` and `libvelox.so` over JNI plus a Spark distribution at `spark.test.home`, which Step 1's container build produces but this checkout does not carry. The suites CI must cover: - [ScalarFunctionsValidateSuite](https://github.com/apache/gluten/blob/b77fdef08e4a733d1ed424bbf807b2904b92af86/backends-velox/src/test/scala/org/apache/gluten/functions/ScalarFunctionsValidateSuite.scala) — one added test, covering the offload and duplicate map keys under both `spark.sql.mapKeyDedupPolicy` values. - [GlutenDataFrameFunctionsSuite](https://github.com/apache/gluten/blob/b77fdef08e4a733d1ed424bbf807b2904b92af86/gluten-ut/spark35/src/test/scala/org/apache/spark/sql/GlutenDataFrameFunctionsSuite.scala), for all four Spark versions - [GlutenFileSourceCharVarcharTestSuite and GlutenDSV2CharVarcharTestSuite](https://github.com/apache/gluten/blob/b77fdef08e4a733d1ed424bbf807b2904b92af86/gluten-ut/spark35/src/test/scala/org/apache/spark/sql/GlutenCharVarcharTestSuite.scala), for Spark 3.4 and 3.5 - [GlutenRuntimeNullChecksV2Writes](https://github.com/apache/gluten/blob/b77fdef08e4a733d1ed424bbf807b2904b92af86/gluten-ut/spark35/src/test/scala/org/apache/spark/sql/GlutenRuntimeNullChecksV2Writes.scala), for Spark 3.5, 4.0 and 4.1 - the ClickHouse UT job, which the `gluten-ut/spark34` and `gluten-ut/spark35` changes trigger ## Was this patch authored or co-authored using generative AI tooling? Generated-by: Claude claude-opus-5 Co-Authored-By: Claude Opus 5 (1M context) --- .../gluten/execution/MiscOperatorSuite.scala | 10 -- .../ScalarFunctionsValidateSuite.scala | 36 +++++ .../SubstraitToVeloxPlanValidator.cc | 2 +- docs/velox-backend-scalar-function-support.md | 4 +- .../clickhouse/ClickHouseTestSettings.scala | 11 ++ .../utils/velox/VeloxTestSettings.scala | 12 ++ .../sql/GlutenCharVarcharTestSuite.scala | 77 +++++++++- .../sql/GlutenDataFrameFunctionsSuite.scala | 45 ++++++ .../clickhouse/ClickHouseTestSettings.scala | 14 ++ .../utils/velox/VeloxTestSettings.scala | 18 +++ .../sql/GlutenCharVarcharTestSuite.scala | 77 +++++++++- .../sql/GlutenDataFrameFunctionsSuite.scala | 48 ++++++ .../sql/GlutenRuntimeNullChecksV2Writes.scala | 135 ++++++++++++++++- .../clickhouse/ClickHouseTestSettings.scala | 4 + .../utils/velox/VeloxTestSettings.scala | 8 + .../sql/GlutenDataFrameFunctionsSuite.scala | 50 +++++++ .../sql/GlutenRuntimeNullChecksV2Writes.scala | 137 +++++++++++++++++- .../clickhouse/ClickHouseTestSettings.scala | 4 + .../utils/velox/VeloxTestSettings.scala | 8 + .../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 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 kRegexFunctions = {"regexp_extract", "regexp_extract_all", "regexp_replace", "regexp_instr", "rlike", "split"}; -const std::unordered_set kBlackList = {"split_part", "sequence", "approx_percentile", "map_from_arrays"}; +const std::unordered_set 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) 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, 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>) 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) 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, 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>) 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")) + } + } +}