[GLUTEN-6101][VL] Enable map_from_arrays function

[pedrumj2]

What changes are proposed in this pull request?

facebookincubator/velox#18630 implemented the Spark version of map_from_arrays in Velox.

#12968 pulled that commit (a6b9f7754) into the Velox revision this repo pins.

This PR drops map_from_arrays from kBlackList, so a query using it now runs in Velox instead of falling back to the JVM. The function was denylisted by #2440 and moved into kBlackList by #6690.

GlutenConfig.getNativeSessionConf already forwards spark.sql.mapKeyDedupPolicy to Velox and ExpressionMappings already maps the expression, so nothing else had to be wired up. The scalar function support doc row is updated to mark the function supported.

Fixes #6101

How was this patch tested?

Local Testing

Save this as verify-map-from-arrays.sh and run it against a checkout of this branch. It builds the Velox backend and the Spark 3.5 jars in the CI dev image, runs the query in a real Spark session, and asserts on the executed plan. It exits non-zero if the projection falls back to the JVM.

#!/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]
#
# Builds the Velox backend and the Spark 3.5 jars in the CI dev image, then runs
# the query in a real Spark session and asserts that the projection carrying
# map_from_arrays executes as a ProjectExecTransformer. Exits non-zero if the
# operator falls back to the JVM.
#
# 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

    # A range-backed view keeps the arguments non-literal, so Spark cannot
    # constant-fold the call and the validator actually sees map_from_arrays.
    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

    # Assert on the executed plan, not on the exit code. Only the final plan
    # counts, and the projection holding the function must be the native one.
    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"

Output on this branch, from a clean tree. The plan and the rows are contiguous runs from the script's own log; [...] marks where Spark's other output was cut.

== 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]
[...]
{0:"0",1:"0x"}
{1:"1",2:"1x"}
{2:"2",3:"2x"}
{3:"3",4:"3x"}
{4:"4",5:"4x"}
[...]
PASS: map_from_arrays executed in Velox as a ProjectExecTransformer

The projection carrying the function is a ProjectExecTransformer, so it ran in Velox. Restoring the kBlackList entry turns that line into *(1) Project [map_from_arrays(...)], which both of the script's plan assertions reject.

Automated Tests

Four tests in ScalarFunctionsValidateSuite cover the offload, duplicate keys under both spark.sql.mapKeyDedupPolicy values, and the lower-case form of that config.

Was this patch authored or co-authored using generative AI tooling?

Generated-by: Co-authored with claude

[pedrumj2]

@kevinwilfong

[philo-he]

Looks good overall. Thanks.

[philo-he]

Spark's SQLConf parses enum configs case-insensitively, so Gluten always receives a normalized enum value from Spark. Testing lower-case inputs in Gluten isn't necessary since that behavior is already guaranteed by Spark.

https://github.com/apache/spark/blob/21c906234f27a5744457aa8fa3da215b0c3be5de/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala#L6737

[pedrumj2]

@philo-he I see thanks for sharing that. I remove this test.

[philo-he]

Can we consolidate the above two tests into one test with two cases included? One is for EXCEPTION policy, producing exception in duplicate case, and the other is for LAST_WIN policy.

[pedrumj2]

@philo-he Thanks for the review. Please see the latest changes. Three tests:

• Duplicate + EXCEPTION policy --> exception
• Duplicate + LAST_WIN policy --> no exception
• No Duplicate + EXCEPTION policy --> no exception

[philo-he]

Thanks for the update.

[philo-he]

@pedrumj2, please check the test failures which should be related. Thanks.