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

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

@@ -45,16 +45,21 @@
 ## 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)).
 - 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`, which `CometExecIterator` forwards 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).
+- Known limitation: `ArrayBasedMapBuilder` normalizes a floating-point key before storing it (`-0.0` becomes `+0.0`, every `NaN` collapses to one), while the native builder compares the raw Arrow values, so a map built from both `-0.0` and `+0.0` keeps two entries where Spark reports a duplicate key. Gated only 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.
 
 ## 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`, which `CometExecIterator` forwards 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).
+- Known limitation: `ArrayBasedMapBuilder` normalizes a floating-point key before storing it (`-0.0` becomes `+0.0`, every `NaN` collapses to one), while the native builder compares the raw Arrow values, so a map built from both `-0.0` and `+0.0` keeps two entries where Spark reports a duplicate key. Gated only 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
@@ -74,7 +79,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

@@ -57,8 +57,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;
@@ -112,7 +110,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 datafusion_comet_proto::spark_operator::operator::OpStruct;
@@ -715,6 +713,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
@@ -754,15 +761,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

@@ -3661,6 +3661,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,
@@ -3787,7 +3795,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

@@ -31,7 +31,8 @@ use crate::{
     EvalMode, SparkArrayPositionFunc, SparkArraySlice, SparkArraysOverlap, SparkContains,
     SparkDateDiff, SparkDateFromUnixDate, SparkDateTrunc, SparkFlatten, SparkIcebergBucket,
     SparkIcebergTemporalTransform, SparkIcebergTruncate, SparkMakeDate, SparkMakeInterval,
-    SparkMakeTime, SparkMapExtract, SparkNextDay, SparkSecondsToTimestamp, SparkSizeFunc,
+    SparkMakeTime, SparkMapExtract, SparkMapFromArrays, SparkMapFromEntries, SparkNextDay,
+    SparkSecondsToTimestamp, SparkSizeFunc, SparkStrToMap,
 };
 use arrow::datatypes::DataType;
 use datafusion::common::{DataFusionError, Result as DataFusionResult};
@@ -325,9 +326,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;