pedrumj2 opened a new pull request, #13490: URL: https://github.com/apache/gluten/pull/13490
## What changes are proposed in this pull request? [`transform_keys`](https://spark.apache.org/docs/latest/api/sql/collection-functions/#transform_keys) now runs in Velox under both values of [`spark.sql.mapKeyDedupPolicy`](https://spark.apache.org/docs/latest/configuration.html#runtime-sql-configuration) and handles repeated keys the way Spark does. facebookincubator/velox#19548 added a Spark `transform_keys` to Velox that follows [`throw_exception_on_duplicate_map_keys`](https://facebookincubator.github.io/velox/configs.html#generic-configuration) and it reached Gluten with the Velox update in apache/gluten#13418. ## How was this patch tested? [`757ae07`](https://github.com/apache/gluten/commit/757ae0759c3c5452d9e8469933f15f2d71d5b2f9), which carries the Velox from apache/gluten#13418, is the base of every run below. ### Reproduction The `apache/gluten:centos-9-jdk8` image runs every step. Run each block from the same empty directory. #### [step1] Get the source and Spark ```bash git clone https://github.com/apache/gluten.git curl -LO https://archive.apache.org/dist/spark/spark-3.5.5/spark-3.5.5-bin-hadoop3.tgz tar -xzf spark-3.5.5-bin-hadoop3.tgz ``` Then fetch this pull request's branch into `gluten` as a local branch named `pr`. #### [step2] Write the queries ```bash cat > repro.sql <<'EOF' DROP TABLE IF EXISTS t; CREATE TABLE t USING parquet AS SELECT id FROM range(1, 3); SET spark.sql.mapKeyDedupPolicy=LAST_WIN; SELECT id, map_entries(transform_keys(map(id, 'a', id + 1, 'b', id + 2, 'c'), (k, v) -> k % 2)) AS m FROM t ORDER BY id; EXPLAIN SELECT map_entries(transform_keys(map(id, 'a', id + 1, 'b', id + 2, 'c'), (k, v) -> k % 2)) AS m FROM t; SET spark.sql.mapKeyDedupPolicy=EXCEPTION; SELECT id, transform_keys(map(id, 'a'), (k, v) -> array(k, CAST(NULL AS BIGINT))) AS m FROM t ORDER BY id; EOF ``` #### [step3] Build Gluten ```bash docker run --rm -v "$PWD":/work:z -w /work/gluten apache/gluten:centos-9-jdk8 bash -c ' source /opt/rh/gcc-toolset-12/enable export VELOX_BUILD_SHARED=ON ./dev/builddeps-veloxbe.sh --build_arrow=OFF --run_setup_script=OFF ./build/mvn clean install -Pbackends-velox -Pspark-3.5 -DskipTests' ``` #### [step4] Run the queries ```bash docker run --rm -v "$PWD":/work:z -w /work apache/gluten:centos-9-jdk8 \ spark-3.5.5-bin-hadoop3/bin/spark-sql --master 'local[2]' \ --jars gluten/package/target/gluten-velox-bundle-spark3.5_2.12-linux_amd64-1.8.0-SNAPSHOT.jar \ --driver-class-path gluten/package/target/gluten-velox-bundle-spark3.5_2.12-linux_amd64-1.8.0-SNAPSHOT.jar \ --conf spark.plugins=org.apache.gluten.GlutenPlugin \ --conf spark.memory.offHeap.enabled=true \ --conf spark.memory.offHeap.size=2g \ --conf spark.shuffle.manager=org.apache.spark.shuffle.sort.ColumnarShuffleManager \ --hiveconf hive.cli.errors.ignore=true \ -f repro.sql ``` #### Before: `757ae07` ```bash git -C gluten checkout --detach 757ae0759c3c5452d9e8469933f15f2d71d5b2f9 ``` Then [step3] and [step4]: ``` 26/10/27 22:32:56 WARN GlutenFallbackReporter: Validation failed for plan: Project[QueryId=4], due to: - Validation failed with exception from: ProjectExecTransformer, reason: LAST_WIN policy is not supported yet in native to deduplicate map keys ... 1 [{"key":1,"value":"c"},{"key":0,"value":"b"}] 2 [{"key":0,"value":"c"},{"key":1,"value":"b"}] ... == Physical Plan == Project [map_entries(transform_keys(map(id#18L, a, (id#18L + 1), b, (id#18L + 2), c), lambdafunction((lambda k#30L % 2), lambda k#30L, lambda v#31, false))) AS m#24] +- VeloxColumnarToRow +- ^(1) FileFileSourceScanExecTransformer parquet spark_catalog.default.t[id#18L] Batched: true, DataFilters: [], Format: Parquet, Location: InMemoryFileIndex(1 paths)[file:/work/spark-warehouse/t], PartitionFilters: [], PushedFilters: [], ReadSchema: struct<id:bigint> NativeFilters: [] ... org.apache.spark.SparkException: Job aborted due to stage failure: Task 1 in stage 5.0 failed 1 times, most recent failure: Lost task 1.0 in stage 5.0 (TID 8) (localhost executor driver): org.apache.gluten.exception.GlutenException: org.apache.gluten.exception.GlutenException: Error during calling Java code from native code: org.apache.gluten.exception.GlutenException: org.apache.gluten.exception.GlutenException: Exception: VeloxUserError Error Source: USER Error Code: INVALID_ARGUMENT Reason: map key cannot be indeterminate: {1, null} Retriable: False Expression: !transformedKeys->containsNullAt(keyRow) Context: Top-level Expression: transform_keys(map(n0_0, a:VARCHAR), (k, v) -> array(k, null:BIGINT)) ``` #### After: this pull request on `757ae07` ```bash git -C gluten cherry-pick --no-commit pr ``` Then [step3] and [step4]: ``` 1 [{"key":1,"value":"c"},{"key":0,"value":"b"}] 2 [{"key":0,"value":"c"},{"key":1,"value":"b"}] ... == Physical Plan == VeloxColumnarToRow +- ^(1) ProjectExecTransformer [map_entries(transform_keys(map(id#18L, a, (id#18L + 1), b, (id#18L + 2), c), lambdafunction((lambda k#30L % 2), lambda k#30L, lambda v#31, false))) AS m#24] +- ^(1) FileFileSourceScanExecTransformer parquet spark_catalog.default.t[id#18L] Batched: true, DataFilters: [], Format: Parquet, Location: InMemoryFileIndex(1 paths)[file:/work/spark-warehouse/t], PartitionFilters: [], PushedFilters: [], ReadSchema: struct<id:bigint> NativeFilters: [] ... 1 {[1,null]:"a"} 2 {[2,null]:"a"} ``` [step4] without the Gluten jars and `--conf` flags prints the same rows from vanilla Spark 3.5.5. ### Tests [`ScalarFunctionsValidateSuite`](https://github.com/apache/gluten/blob/main/backends-velox/src/test/scala/org/apache/gluten/functions/ScalarFunctionsValidateSuite.scala) gains a case that runs `transform_keys` with a repeated key under both policies and asserts that it stays in Velox and matches vanilla Spark, key order included. ``` ./build/mvn test -Pbackends-velox -Pspark-3.5 -Pfast-build -pl backends-velox -Dtest=none -Dsurefire.failIfNoSpecifiedTests=false "-Dsuites=org.apache.gluten.functions.ScalarFunctionsValidateSuite transform_keys" -DtagsToExclude=org.apache.spark.tags.SkipTest ``` ``` ScalarFunctionsValidateSuite: - transform_keys - transform_keys offloads to Velox under both mapKeyDedupPolicy values ... Run completed in 10 seconds, 442 milliseconds. Suites: completed 1, aborted 0 Tests: succeeded 2, failed 0, canceled 0, ignored 0, pending 0 All tests passed. ``` ``` ./build/mvn test -Pbackends-velox -Pspark-3.5 -Pspark-ut -Pfast-build -pl gluten-ut/spark35 -Dtest=none -Dsurefire.failIfNoSpecifiedTests=false "-Dsuites=org.apache.spark.sql.catalyst.expressions.GlutenHigherOrderFunctionsSuite TransformKeys,org.apache.spark.sql.GlutenDataFrameFunctionsSuite transform keys" -DtagsToExclude=org.apache.spark.tags.SkipTest ``` ``` GlutenHigherOrderFunctionsSuite: - TransformKeys ... GlutenDataFrameFunctionsSuite: - transform keys function - primitive data types - transform keys function - Invalid lambda functions and exceptions ... Run completed in 14 seconds, 121 milliseconds. Suites: completed 2, aborted 0 Tests: succeeded 3, failed 0, canceled 0, ignored 0, pending 0 All tests passed. ``` ## Was this patch authored or co-authored using generative AI tooling? Generated-by: Co-authored with Claude Code (claude-opus-5-5) <!-- Thank you for submitting a pull request! Here are some tips: 1. For first-time contributors, please read our contributing guide: https://github.com/apache/gluten/blob/main/CONTRIBUTING.md 2. If necessary, create a GitHub issue for discussion beforehand to avoid duplicate work. 3. If the PR is specific to a single backend, include [VL] or [CH] in the PR title to indicate the Velox or ClickHouse backend, respectively. 4. If the PR is not ready for review, please mark it as a draft. --> ## What changes are proposed in this pull request? <!-- Provide a clear and concise description of the changes introduced in this PR. Ensure the PR description aligns with the code changes, especially after updates. If applicable, include "Fixes #<GitHub_Issue_ID>" to automatically close the corresponding issue when the PR is merged. --> ## How was this patch tested? <!-- Describe how the changes were tested, if applicable. Include new tests to validate the functionality, if necessary. For UI-related changes, attach screenshots to demonstrate the updates. --> ## Was this patch authored or co-authored using generative AI tooling? <!-- If generative AI tooling has been used in the process of authoring this patch, please include the phrase: 'Generated-by: ' followed by the name of the tool and its version. If no, write 'No'. Please refer to the [ASF Generative Tooling Guidance](https://www.apache.org/legal/generative-tooling.html) for details. --> -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
