fix: enforce null-key rejection and mapKeyDedupPolicy in native map construction - #5854
fix: enforce null-key rejection and mapKeyDedupPolicy in native map construction#5854peterxcli 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
…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
| self.inner | ||
| .invoke_with_args(args) | ||
| .map_err(|error| as_spark_error(error, DuplicateKeyFormat::Bare)) |
There was a problem hiding this comment.
This can return a key from a preceding row. With keys [[10], [20]] and values [[100], [200]], slicing both to the second row returns {10: 200} instead of {20: 200}. The previous MapFunc returns {20: 200}.
The helper applies a zero-based keys_mask to the unsliced flat_keys, while value indices include the starting offset. I reproduced this through a native GlobalLimitExec -> ProjectionExec component test on DataFusion 55.0.0; the relevant kernels are unchanged in 55.1.0.
Please fix the offset handling in the helper or normalize the inputs before delegation, and add a sliced-list regression test. The newly enabled LAST_WIN path for map_from_entries is affected too.
| if let Some(nulls) = &key_nulls { | ||
| if nulls.slice(start, end - start).null_count() > 0 { | ||
| return Err(SparkError::NullMapKey.into()); | ||
| } |
There was a problem hiding this comment.
For keys [1, 1, NULL] under EXCEPTION, Spark 4.1.3 reports DUPLICATED_MAP_KEY, but this pre-scan reports NULL_MAP_KEY. Spark inserts entries in order and fails on the second key before reaching the null.
Please preserve that check order in both builders and update the comments claiming null-key errors always take precedence.
andygrove
left a comment
There was a problem hiding this comment.
MapBuilderSupport.keySupport only looks at the floating point gate, but MapKeySupport.keySupport a few lines above declines a non-default string collation for map lookups. Spark's ArrayBasedMapBuilder picks a collation-aware TreeMap for any StringType that is not supportsBinaryEquality, so under UTF8_LCASE the keys 'a' and 'A' are the same key, while the native builder compares raw Arrow values.
SELECT map_from_arrays(array(CAST('a' AS STRING COLLATE UTF8_LCASE), CAST('A' AS STRING COLLATE UTF8_LCASE)), array(i, i)) FROM t looks like it reaches the native path, since both casts have Literal children so CometCast folds them and supportedDataType accepts a collated StringType. Spark 4 raises DUPLICATED_MAP_KEY there and Comet returns a two-entry map. The LAST_WIN side worries me more, because the old isLastWin branch declined and sent that case back to Spark, so it used to be correct and is not after this change. Would it make sense for MapBuilderSupport.keySupport to call hasNonDefaultStringCollation the way MapKeySupport.keySupport does, with a fixture next to element_at_map_collation.sql to pin it?
The floating point note also reads as though the only difference is duplicate detection, and I think it is off in both directions. Spark only normalizes map keys from 4.0.0 onwards. spark.sql.legacy.disableMapKeyNormalization is marked .version("4.0.0") and the 3.5 ArrayBasedMapBuilder has no keyNormalizer at all, so on 3.4 and 3.5 the native builder already matches and spark.comet.exec.strictFloatingPoint declines for nothing. On 4.0 and later MapFromArrays calls mapBuilder.from(...), which reuses the original key array whenever the row has no duplicates, so a lone -0.0 key stays -0.0 in Spark too and "a -0.0 key is stored as +0.0" is not observable through map_from_arrays. MapFromEntries is the one that stores the normalized key, because it goes through put and build(), so SELECT map_from_entries(array(struct(-0.0D AS key, 1 AS value))) gives {0.0 -> 1} on Spark 4 and {-0.0 -> 1} on Comet with no duplicate anywhere. The NaN half overstates it too, since ScalarValue compares floats by to_bits, so two double('NaN') keys do collapse natively. Could the note be reworded around what actually differs per function and scoped to Spark 4.0 and later? A fixture for a floating point map key would help, since neither the default path nor the new strict decline is exercised today.
One smaller thing. The comment on checkSparkErrorParity says the mismatched-length case goes through a _LEGACY_ERROR_TEMP_* condition whose number moves between versions, but reading it out of the shipped jars it is _LEGACY_ERROR_TEMP_2128 on 3.4.3, 3.5.8, 4.0.1, 4.1.3 and 4.2.0 alike. If that holds then checkSparkError(df, "_LEGACY_ERROR_TEMP_2128") and expect_error(_LEGACY_ERROR_TEMP_2128) pin it directly and the new helper is not needed. Is there a version where the number actually differs?
On sequencing, #5846 also rewrites CometMapFromArrays.convert and adds its own SparkMapFromArrays re-export, and #5844 adds the CodegenDispatchFallback for LAST_WIN that this PR would make unnecessary. I have commented on both pointing here. Worth agreeing an order with @sunchao and @LinSimon-901101 so the same lines are not landed twice.
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.