fix: enforce null-key rejection and mapKeyDedupPolicy in native map construction

[peterxcli]

Which issue does this PR close?

• Closes [Bug] map_from_arrays / map_from_entries do not enforce null-key rejection or spark.sql.mapKeyDedupPolicy #4680.
• Closes str_to_map does not honour mapKeyDedupPolicy=LAST_WIN #5857.

Rationale for this change

Spark builds every map through ArrayBasedMapBuilder, which refuses a NULL key and resolves duplicate keys according to spark.sql.mapKeyDedupPolicy. Comet's map_from_arrays and map_from_entries did neither. A NULL inside the keys array produced a map with a NULL key instead of an error, and setting the policy to LAST_WIN pushed the whole expression back to Spark.

DataFusion 55 supplies what was missing. Its datafusion.spark.map_key_dedup_policy option takes the same EXCEPTION and LAST_WIN values as the Spark config, and the datafusion-spark map kernels already follow it. Once Comet passes the setting through, LAST_WIN runs natively, and the remaining checks that ArrayBasedMapBuilder performs cost only a few lines on top.

What changes are included in this PR?

spark.sql.mapKeyDedupPolicy now crosses JNI. CometExecIterator.serializeCometSQLConfs sends it explicitly, since cometSqlConfs carries only keys under spark.comet., and prepare_datafusion_session_context applies it to the session as datafusion.spark.map_key_dedup_policy.

A second change was needed before that setting could reach a kernel at all. create_scalar_function_expr handed every ScalarFunctionExpr a fresh ConfigOptions::default(), so any kernel reading a session option saw DataFusion's defaults. It now passes the session's own ConfigOptions.

native/spark-expr/src/map_funcs/map_builders.rs adds three wrappers, SparkMapFromArrays, SparkMapFromEntries and SparkStrToMap. Each calls the matching datafusion-spark kernel, adds the checks that kernel skips, and translates its errors into the Spark error classes that SparkErrorConverter converts back into QueryExecutionErrors:

• a NULL key raises NULL_MAP_KEY, before any check for duplicates, matching the order Spark applies them;
• key and value arrays of unequal length raise MAP_KEY_VALUE_DIFF_SIZES;
• a duplicate key under EXCEPTION raises DUPLICATED_MAP_KEY and names the key.

str_to_map needs only the last of these, because splitting a string never yields a NULL key. Passing the config through also fixed its LAST_WIN case, which used to raise an error where Spark returns a map.

CometMapFromArrays now emits map_from_arrays. It used to emit the generic map wrapped in CaseWhen(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 their LAST_WIN Incompatible branch.

One difference with Spark remains. ArrayBasedMapBuilder normalizes a floating point key before storing it, so -0.0 becomes +0.0 and every NaN collapses into one. The native builders compare the raw Arrow values, so a map built from both -0.0 and +0.0 keeps two entries where Spark reports a duplicate key. The compatibility notes record this, and spark.comet.exec.strictFloatingPoint makes 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 CometMapExpressionSuite run each case through both engines and compare the exception type, error class and SQLSTATE, along with the answers each engine returns under LAST_WIN.

Among the SQL fixtures, the two *_dedup_policy.sql files used to assert the LAST_WIN fallback and now assert native execution. map_from_arrays.sql, map_from_entries.sql and str_to_map.sql gained the EXCEPTION error cases, and str_to_map_dedup_policy.sql is new. That also retires the TODO: Add LAST_WIN policy tests when spark.sql.mapKeyDedupPolicy config is supported note in str_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, so CometTestBase.checkSparkError now builds on a new checkSparkErrorParity helper.

The ConfigOptions change affects every scalar function, so the full 487-fixture suite ran green as well.

[rich7420]

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.

[rich7420]

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]

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.