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

docs/source/contributor-guide/expression-audits/map_funcs.md

@@ -45,16 +45,23 @@
 ## map_from_arrays
 
 - Spark 3.4.3 (audited 2026-05-27): identical to 3.5.8.
-- Spark 3.5.8 (audited 2026-05-27): baseline. `MapFromArrays(left, right) extends BinaryExpression with NullIntolerant`; Spark uses `ArrayBasedMapBuilder` to detect duplicate keys (subject to `spark.sql.mapKeyDedupPolicy`) and rejects null keys with `RuntimeException("Cannot use null as map key")`. Comet `CometMapFromArrays` wraps the inputs in `CaseWhen(IsNotNull(left) AND IsNotNull(right), map(left, right), null)` so NULL-array inputs return NULL rather than triggering the previously reported native crash ([#3327](https://github.com/apache/datafusion-comet/issues/3327)).
+- Spark 3.5.8 (audited 2026-05-27): baseline. `MapFromArrays(left, right) extends BinaryExpression with NullIntolerant`; Spark uses `ArrayBasedMapBuilder` to detect duplicate keys (subject to `spark.sql.mapKeyDedupPolicy`) and rejects null keys with `RuntimeException("Cannot use null as map key")`. Comet `CometMapFromArrays` wires the native `map_from_arrays` from `datafusion-spark`, which is null intolerant the same way, so NULL-array inputs return NULL rather than triggering the previously reported native crash ([#3327](https://github.com/apache/datafusion-comet/issues/3327)). The serde still nests `CASE WHEN left IS NOT NULL THEN (CASE WHEN right IS NOT NULL THEN map_from_arrays(left, right) END) END` around the call: `BinaryExpression.eval` never evaluates `right` for a row whose `left` is NULL, and DataFusion evaluates a THEN branch only on the rows its WHEN selected, so a failing cast in the values array does not run for such a row. A single `left IS NOT NULL AND right IS NOT NULL` guard does not give that, since DataFusion's `AND` evaluates its right side on the whole batch unless the left side is false on all or most rows.
 - Spark 4.0.1 (audited 2026-05-27): semantics unchanged; `NullIntolerant` trait replaced by `nullIntolerant: Boolean`.
 - Spark 4.1.1 (audited 2026-05-27): identical to 4.0.1.
+- `ArrayBasedMapBuilder` semantics, reproduced natively rather than falling back ([#4680](https://github.com/apache/datafusion-comet/issues/4680)): a `NULL` key element raises `NULL_MAP_KEY`, ahead of any duplicate-key check, matching the order Spark applies them in; a duplicate key follows `spark.sql.mapKeyDedupPolicy`, forwarded to the native session as `datafusion.spark.map_key_dedup_policy` (`EXCEPTION` raises `DUPLICATED_MAP_KEY` naming the key, `LAST_WIN` keeps the last value for the key). `CometExecIterator.serializeCometSQLConfs` reads the setting when it builds the native plan for a task, so materializing or explaining a plan does not fix it and a Dataset re-executed after a change to the setting uses the new value.
+- Known limitation: on Spark 4.0+, `ArrayBasedMapBuilder` normalizes a floating-point key before comparing it (`keyNormalizer`, added in 4.0 with `spark.sql.legacy.disableMapKeyNormalization`), so `-0.0` and `+0.0` are one key and all `NaN`s are one key; the native builder compares the raw Arrow values and keeps them apart. `from` returns the input arrays untouched when no key repeated, so the stored keys match Spark either way and only duplicate detection diverges. Spark 3.4 and 3.5 do not normalize, so they already match. Gated under `spark.comet.exec.strictFloatingPoint`, which marks the expression `Incompatible` for a floating-point key type.
+- Spark raises `MAP_KEY_VALUE_DIFF_SIZES` when a row's key and value arrays differ in length; the native path raises the same error.
+- Known limitation: Spark reads `spark.sql.mapKeyDedupPolicy` into `ArrayBasedMapBuilder`, a lazy field of the map expression, so _when_ it reads it depends on how the projection runs. Outside whole-stage codegen (the flag off, or a projection wider than `spark.sql.codegen.maxFields`) the projection is rebuilt in every task and the setting is read again on each action, which is what Comet does. Inside whole-stage codegen Spark creates the builder once on the driver, in the first action, and keeps it, so a Dataset re-executed after a change to the setting still builds its maps under the policy it started with, where Comet uses the new one. Comet cannot tell the two apart: it replaces the operator before `CollapseCodegenStages` runs, so the plan it sees carries no record of which path Spark would have taken. Matching the whole-stage case instead would mean returning a map where Spark raises `DUPLICATED_MAP_KEY` in the other three configurations, so the loud divergence is preferred over the silent one. Only a Dataset that is executed more than once across a change to the setting is affected.
+- Known limitation: the two null guards serialize each child a second time inside the `map_from_arrays` call, so a nondeterministic child such as `monotonically_increasing_id()` would advance independently in each copy and the result would drift from Spark ([#5781](https://github.com/apache/datafusion-comet/issues/5781)). `CometMapFromArrays` declines such a child as `Unsupported` through `NullGuardSupport` and the projection falls back to Spark; [#5867](https://github.com/apache/datafusion-comet/pull/5867) routes the same decline through the JVM codegen dispatcher and applies it to `size`, `array_append` and `arrays_zip` as well.
 
 ## map_from_entries
 
 - Spark 3.4.3 (audited 2026-05-27): identical to 3.5.8.
 - Spark 3.5.8 (audited 2026-05-27): baseline. `MapFromEntries(child) extends UnaryExpression with NullIntolerant`; expects an array of structs and produces a map. Wired as `CometScalarFunction("map_from_entries")`.
 - Spark 4.0.1 (audited 2026-05-27): semantics unchanged; trait refactor.
 - Spark 4.1.1 (audited 2026-05-27): identical to 4.0.1.
+- `ArrayBasedMapBuilder` semantics, reproduced natively rather than falling back ([#4680](https://github.com/apache/datafusion-comet/issues/4680)): a `NULL` key element raises `NULL_MAP_KEY`, ahead of any duplicate-key check, matching the order Spark applies them in; a duplicate key follows `spark.sql.mapKeyDedupPolicy`, forwarded to the native session as `datafusion.spark.map_key_dedup_policy` (`EXCEPTION` raises `DUPLICATED_MAP_KEY` naming the key, `LAST_WIN` keeps the last value for the key). `CometExecIterator.serializeCometSQLConfs` reads the setting when it builds the native plan for a task, so materializing or explaining a plan does not fix it and a Dataset re-executed after a change to the setting uses the new value.
+- Known limitation: on Spark 4.0+, `ArrayBasedMapBuilder` normalizes a floating-point key before comparing it (`keyNormalizer`, added in 4.0 with `spark.sql.legacy.disableMapKeyNormalization`), so `-0.0` and `+0.0` are one key and all `NaN`s are one key; the native builder compares the raw Arrow values and keeps them apart. Unlike `map_from_arrays`, this expression always calls `build()`, so Spark stores the normalized key and returns `+0.0` for a `-0.0` key where Comet returns `-0.0`. Spark 3.4 and 3.5 do not normalize, so they already match. Gated under `spark.comet.exec.strictFloatingPoint`, which marks the expression `Incompatible` for a floating-point key type.
 - Known limitation: input arrays where the struct's key or value type contains `BinaryType` are marked `Incompatible` and fall back unless `spark.comet.expression.MapFromEntries.allowIncompatible=true`.
 
 ## map_keys
@@ -79,7 +86,7 @@
 ## str_to_map
 
 - Spark 3.4.3 (audited 2026-05-27): identical to 3.5.8.
-- Spark 3.5.8 (audited 2026-05-27): baseline. `StringToMap(text, pairDelim, keyValueDelim) extends TernaryExpression`; splits `text` on `pairDelim`, then each pair on `keyValueDelim` (default `","` and `":"`). Uses `ArrayBasedMapBuilder` for duplicate-key handling. Wired as `CometScalarFunction("str_to_map")`.
+- Spark 3.5.8 (audited 2026-05-27): baseline. `StringToMap(text, pairDelim, keyValueDelim) extends TernaryExpression`; splits `text` on `pairDelim`, then each pair on `keyValueDelim` (default `","` and `":"`). Uses `ArrayBasedMapBuilder` for duplicate-key handling. Wired as `CometScalarFunction("str_to_map")`. The native `str_to_map` reads the duplicate-key policy from `datafusion.spark.map_key_dedup_policy`, which `CometExecIterator` forwards from `spark.sql.mapKeyDedupPolicy`.
 - Spark 4.0.1 (audited 2026-05-27): `inputTypes` widened to `StringTypeNonCSAICollation`; uses `CollationAwareUTF8String.splitSQL` with a `collationId`. Runtime unchanged for `UTF8_BINARY`.
 - Spark 4.1.1 (audited 2026-05-27): adds the `legacySplitTruncate` flag (driven by `spark.sql.legacy.truncateForEmptyRegexSplit`) to both `splitSQL` calls. The Comet native impl always behaves as if the flag were false, so `CometStrToMap` reads the config by string key and reports `Incompatible` when it is enabled; the `CodegenDispatchFallback` trait then routes the expression through the JVM codegen dispatcher rather than falling the whole projection back to Spark. Non-UTF8_BINARY collations on the input or the delimiters are handled the same way.
 

native/core/src/execution/jni_api.rs

@@ -59,8 +59,6 @@ use datafusion_spark::function::datetime::to_utc_timestamp::SparkToUtcTimestamp;
 use datafusion_spark::function::hash::crc32::SparkCrc32;
 use datafusion_spark::function::hash::sha1::SparkSha1;
 use datafusion_spark::function::hash::sha2::SparkSha2;
-use datafusion_spark::function::map::map_from_entries::MapFromEntries;
-use datafusion_spark::function::map::str_to_map::SparkStrToMap;
 use datafusion_spark::function::math::expm1::SparkExpm1;
 use datafusion_spark::function::math::factorial::SparkFactorial;
 use datafusion_spark::function::math::hex::SparkHex;
@@ -115,7 +113,7 @@ use crate::execution::memory_pools::logging_pool::LoggingMemoryPool;
 use crate::execution::spark_config::{
     SparkConfig, COMET_DEBUG_ENABLED, COMET_DEBUG_MEMORY, COMET_EXPLAIN_NATIVE_ENABLED,
     COMET_MAX_TEMP_DIRECTORY_SIZE, COMET_PARQUET_ROW_FILTER_PUSHDOWN_ENABLED,
-    COMET_TRACING_ENABLED, SPARK_EXECUTOR_CORES,
+    COMET_TRACING_ENABLED, SPARK_EXECUTOR_CORES, SPARK_MAP_KEY_DEDUP_POLICY,
 };
 use crate::parquet::encryption_support::{CometEncryptionFactory, ENCRYPTION_FACTORY_ID};
 use crate::parquet::parquet_support::CometObjectStoreRegistry;
@@ -761,6 +759,15 @@ fn prepare_datafusion_session_context(
             session_config.set_str("datafusion.execution.parquet.reorder_filters", "true");
     }
 
+    // `map_from_arrays`, `map_from_entries` and `str_to_map` build their maps with the
+    // duplicate-key policy Spark's `ArrayBasedMapBuilder` uses. DataFusion spells the same
+    // setting `datafusion.spark.map_key_dedup_policy` and takes the same `EXCEPTION` /
+    // `LAST_WIN` values. Set before the `spark.comet.datafusion.*` testing escape hatch
+    // pass-through below, so an explicit override of the DataFusion key still wins.
+    if let Some(policy) = spark_config.get(SPARK_MAP_KEY_DEDUP_POLICY) {
+        session_config = session_config.set_str("datafusion.spark.map_key_dedup_policy", policy);
+    }
+
     // Pass through DataFusion configs from Spark.
     // e.g: spark-shell --conf spark.comet.datafusion.sql_parser.parse_float_as_decimal=true
     // becomes datafusion.sql_parser.parse_float_as_decimal=true
@@ -802,15 +809,13 @@ fn register_datafusion_spark_function(session_ctx: &SessionContext) {
     session_ctx.register_udf(ScalarUDF::new_from_impl(SparkBitwiseNot::default()));
     session_ctx.register_udf(ScalarUDF::new_from_impl(SparkHex::default()));
     session_ctx.register_udf(ScalarUDF::new_from_impl(SparkWidthBucket::default()));
-    session_ctx.register_udf(ScalarUDF::new_from_impl(MapFromEntries::default()));
     session_ctx.register_udf(ScalarUDF::new_from_impl(SparkCrc32::default()));
     session_ctx.register_udf(ScalarUDF::new_from_impl(SparkLuhnCheck::default()));
     session_ctx.register_udf(ScalarUDF::new_from_impl(SparkSpace::default()));
     session_ctx.register_udf(ScalarUDF::new_from_impl(SparkBitCount::default()));
     session_ctx.register_udf(ScalarUDF::new_from_impl(SparkArrayContains::default()));
     session_ctx.register_udf(ScalarUDF::new_from_impl(SparkArrayRepeat::default()));
     session_ctx.register_udf(ScalarUDF::new_from_impl(SparkBin::default()));
-    session_ctx.register_udf(ScalarUDF::new_from_impl(SparkStrToMap::default()));
     session_ctx.register_udf(ScalarUDF::new_from_impl(SparkUrlDecode::default()));
     session_ctx.register_udf(ScalarUDF::new_from_impl(SparkUrlEncode::default()));
     session_ctx.register_udf(ScalarUDF::new_from_impl(SparkTryUrlDecode::default()));

native/core/src/execution/planner.rs

@@ -3666,6 +3666,14 @@ impl PhysicalPlanner {
         }
     }
 
+    /// The session's `ConfigOptions`, so a kernel that reads one sees what
+    /// `prepare_datafusion_session_context` set rather than DataFusion's defaults. The map
+    /// builders read `datafusion.spark.map_key_dedup_policy` this way, which Comet forwards from
+    /// `spark.sql.mapKeyDedupPolicy`.
+    fn session_config_options(&self) -> Arc<ConfigOptions> {
+        Arc::clone(self.session_ctx.copied_config().options())
+    }
+
     fn create_scalar_function_expr(
         &self,
         expr: &ScalarFunc,
@@ -3792,7 +3800,7 @@ impl PhysicalPlanner {
             fun_expr,
             args.to_vec(),
             Arc::new(Field::new(fun_name, data_type.clone(), true)),
-            Arc::new(ConfigOptions::default()),
+            self.session_config_options(),
         ));
 
         // DF53 changed some UDFs (e.g. md5) to return StringViewArray at execution

native/core/src/execution/spark_config.rs

@@ -25,6 +25,8 @@ pub(crate) const COMET_DEBUG_MEMORY: &str = "spark.comet.debug.memory";
 pub(crate) const COMET_PARQUET_ROW_FILTER_PUSHDOWN_ENABLED: &str =
     "spark.comet.parquet.rowFilterPushdown.enabled";
 pub(crate) const SPARK_EXECUTOR_CORES: &str = "spark.executor.cores";
+/// Spark's duplicate map key policy, forwarded to `datafusion.spark.map_key_dedup_policy`.
+pub(crate) const SPARK_MAP_KEY_DEDUP_POLICY: &str = "spark.sql.mapKeyDedupPolicy";
 
 pub(crate) trait SparkConfig {
     fn get_bool(&self, name: &str) -> bool;

native/spark-expr/src/comet_scalar_funcs.rs

@@ -32,8 +32,8 @@ use crate::{
     EvalMode, SparkArrayPositionFunc, SparkArraySlice, SparkArraysOverlap, SparkContains,
     SparkDateDiff, SparkDateFromUnixDate, SparkDateTrunc, SparkDayOfWeek, SparkFlatten,
     SparkIcebergBucket, SparkIcebergTemporalTransform, SparkIcebergTruncate, SparkMakeDate,
-    SparkMakeInterval, SparkMakeTime, SparkMapExtract, SparkNextDay, SparkSecondsToTimestamp,
-    SparkSizeFunc, SparkWeekDay,
+    SparkMakeInterval, SparkMakeTime, SparkMapExtract, SparkMapFromArrays, SparkMapFromEntries,
+    SparkNextDay, SparkSecondsToTimestamp, SparkSizeFunc, SparkStrToMap, SparkWeekDay,
 };
 use arrow::datatypes::DataType;
 use datafusion::common::{DataFusionError, Result as DataFusionResult};
@@ -341,9 +341,12 @@ fn all_scalar_functions() -> Vec<Arc<ScalarUDF>> {
         // returns the value itself rather than a one-element list (#5795). It carries the same
         // `element_at` alias so both registry entries the override replaces point here.
         Arc::new(ScalarUDF::new_from_impl(SparkMapExtract::default())),
+        Arc::new(ScalarUDF::new_from_impl(SparkMapFromArrays::default())),
+        Arc::new(ScalarUDF::new_from_impl(SparkMapFromEntries::default())),
         Arc::new(ScalarUDF::new_from_impl(SparkNextDay::default())),
         Arc::new(ScalarUDF::new_from_impl(SparkSecondsToTimestamp::default())),
         Arc::new(ScalarUDF::new_from_impl(SparkSizeFunc::default())),
+        Arc::new(ScalarUDF::new_from_impl(SparkStrToMap::default())),
         Arc::new(ScalarUDF::new_from_impl(JsonArrayLength::default())),
     ]
 }

native/spark-expr/src/lib.rs

@@ -61,7 +61,9 @@ pub mod jvm_udf;
 mod conditional_funcs;
 mod conversion_funcs;
 mod map_funcs;
-pub use map_funcs::{spark_map_sort, SparkMapExtract};
+pub use map_funcs::{
+    spark_map_sort, SparkMapExtract, SparkMapFromArrays, SparkMapFromEntries, SparkStrToMap,
+};
 mod math_funcs;
 mod nondetermenistic_funcs;
 pub mod url_funcs;