fix: enforce null-key rejection and mapKeyDedupPolicy in native map construction - #5854
Open
peterxcli wants to merge 2 commits into
Open
fix: enforce null-key rejection and mapKeyDedupPolicy in native map construction#5854peterxcli wants to merge 2 commits into
peterxcli wants to merge 2 commits into
Conversation
…onstruction `map_from_arrays` and `map_from_entries` built their maps without the entry checks Spark's `ArrayBasedMapBuilder` performs, so a `NULL` key inside the keys array produced a map with a `NULL` key instead of raising `NULL_MAP_KEY`, and `spark.sql.mapKeyDedupPolicy=LAST_WIN` fell the whole expression back to Spark. DataFusion 55 added `datafusion.spark.map_key_dedup_policy` and taught the `datafusion-spark` map kernels to follow it, which is the missing half. Forward Spark's `spark.sql.mapKeyDedupPolicy` to it across JNI, and pass the session's `ConfigOptions` into `ScalarFunctionExpr` so a kernel that reads a setting sees the session's value rather than DataFusion's defaults. New `SparkMapFromArrays` / `SparkMapFromEntries` / `SparkStrToMap` wrappers add the checks the upstream kernels do not perform and restate their errors as the Spark error classes `SparkErrorConverter` turns back into `QueryExecutionErrors`: a `NULL` key raises `NULL_MAP_KEY` ahead of any duplicate-key check, key and value arrays of different lengths raise `MAP_KEY_VALUE_DIFF_SIZES`, and a duplicate key under `EXCEPTION` raises `DUPLICATED_MAP_KEY` naming the key. `CometMapFromArrays` now emits `map_from_arrays`, which is null intolerant like Spark's, so the `CaseWhen` guard against NULL input arrays is no longer needed. A floating-point map key stays a documented difference: Spark normalizes `-0.0` to `+0.0` and canonicalizes `NaN` before storing a key, while the native builders compare the raw Arrow values. `spark.comet.exec.strictFloatingPoint` declines those key types. Closes apache#4680 Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01REK8xCiYKTcqw1NGQniHXj
peterxcli
marked this pull request as ready for review
September 11, 2026 09:34
…d-dedup-policy # Conflicts: # native/spark-expr/src/comet_scalar_funcs.rs # native/spark-expr/src/lib.rs # native/spark-expr/src/map_funcs/mod.rs
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Which issue does this PR close?
Rationale for this change
Spark builds every map through
ArrayBasedMapBuilder, which refuses aNULLkey and resolves duplicate keys according tospark.sql.mapKeyDedupPolicy. Comet'smap_from_arraysandmap_from_entriesdid neither. ANULLinside the keys array produced a map with aNULLkey instead of an error, and setting the policy toLAST_WINpushed the whole expression back to Spark.DataFusion 55 supplies what was missing. Its
datafusion.spark.map_key_dedup_policyoption takes the sameEXCEPTIONandLAST_WINvalues as the Spark config, and thedatafusion-sparkmap kernels already follow it. Once Comet passes the setting through,LAST_WINruns natively, and the remaining checks thatArrayBasedMapBuilderperforms cost only a few lines on top.What changes are included in this PR?
spark.sql.mapKeyDedupPolicynow crosses JNI.CometExecIterator.serializeCometSQLConfssends it explicitly, sincecometSqlConfscarries only keys underspark.comet., andprepare_datafusion_session_contextapplies it to the session asdatafusion.spark.map_key_dedup_policy.A second change was needed before that setting could reach a kernel at all.
create_scalar_function_exprhanded everyScalarFunctionExpra freshConfigOptions::default(), so any kernel reading a session option saw DataFusion's defaults. It now passes the session's ownConfigOptions.native/spark-expr/src/map_funcs/map_builders.rsadds three wrappers,SparkMapFromArrays,SparkMapFromEntriesandSparkStrToMap. Each calls the matchingdatafusion-sparkkernel, adds the checks that kernel skips, and translates its errors into the Spark error classes thatSparkErrorConverterconverts back intoQueryExecutionErrors:NULLkey raisesNULL_MAP_KEY, before any check for duplicates, matching the order Spark applies them;MAP_KEY_VALUE_DIFF_SIZES;EXCEPTIONraisesDUPLICATED_MAP_KEYand names the key.str_to_mapneeds only the last of these, because splitting a string never yields aNULLkey. Passing the config through also fixed itsLAST_WINcase, which used to raise an error where Spark returns a map.CometMapFromArraysnow emitsmap_from_arrays. It used to emit the genericmapwrapped inCaseWhen(IsNotNull(left) AND IsNotNull(right), ...)so that a NULL input array yielded a NULL map; the Spark kernel already behaves that way, so the wrapper came out. Both serdes also drop theirLAST_WINIncompatiblebranch.One difference with Spark remains.
ArrayBasedMapBuildernormalizes a floating point key before storing it, so-0.0becomes+0.0and everyNaNcollapses into one. The native builders compare the raw Arrow values, so a map built from both-0.0and+0.0keeps two entries where Spark reports a duplicate key. The compatibility notes record this, andspark.comet.exec.strictFloatingPointmakes Comet decline a floating point key type for anyone who needs the guarantee.How are these changes tested?
The 21 native unit tests cover the wrappers. Two of them pin the exact wording DataFusion uses when it reports a duplicate key, because the wrapper reads that message to recover the key it should name. If DataFusion rewords the message, those tests fail rather than the error quietly degrading into a generic execution failure.
Seven new tests in
CometMapExpressionSuiterun each case through both engines and compare the exception type, error class and SQLSTATE, along with the answers each engine returns underLAST_WIN.Among the SQL fixtures, the two
*_dedup_policy.sqlfiles used to assert theLAST_WINfallback and now assert native execution.map_from_arrays.sql,map_from_entries.sqlandstr_to_map.sqlgained theEXCEPTIONerror cases, andstr_to_map_dedup_policy.sqlis new. That also retires theTODO: Add LAST_WIN policy tests when spark.sql.mapKeyDedupPolicy config is supportednote instr_to_map.sql.The test for mismatched array lengths compares the two engines against each other instead of naming an error condition. Spark still reports that case through a
_LEGACY_ERROR_TEMP_*condition whose number moves between Spark versions, soCometTestBase.checkSparkErrornow builds on a newcheckSparkErrorParityhelper.The
ConfigOptionschange affects every scalar function, so the full 487-fixture suite ran green as well.