feat(functions): map functions and spark.sql.mapKeyDedupPolicy

crates/krishiv-cli/src/sql.rs

@@ -484,6 +484,24 @@ fn cell(a: &dyn Array, i: usize, null: &str, tz: i64) -> String {
                 .collect();
             format!("{{{}}}", items.join(", "))
         }
+        DataType::Map(_, _) => {
+            let m = a.as_map();
+            let offsets = m.value_offsets();
+            let (from, to) = (
+                usize::try_from(offsets[i]).unwrap_or(0),
+                usize::try_from(offsets[i + 1]).unwrap_or(0),
+            );
+            let items: Vec<String> = (from..to)
+                .map(|j| {
+                    format!(
+                        "{} -> {}",
+                        cell(m.keys().as_ref(), j, "NULL", tz),
+                        cell(m.values().as_ref(), j, "NULL", tz)
+                    )
+                })
+                .collect();
+            format!("{{{}}}", items.join(", "))
+        }
         other => format!("<{other}>"),
     }
 }
@@ -591,6 +609,20 @@ mod tests {
             Arc::new(Field::new("x", DataType::Int32, true)),
             arr(Int32Array::from(vec![7])),
         )]);
+        let mut mb = krishiv::arrow::array::builder::MapBuilder::new(
+            None,
+            krishiv::arrow::array::builder::StringBuilder::new(),
+            krishiv::arrow::array::builder::Int32Builder::new(),
+        );
+        mb.keys().append_value("a");
+        mb.values().append_value(1);
+        mb.keys().append_value("b");
+        mb.values().append_null();
+        mb.append(true).expect("map entry");
+        mb.append(true).expect("empty map");
+        let map = mb.finish();
+        assert_eq!(cell(&map, 0, "NULL", 0), "{a -> 1, b -> NULL}");
+        assert_eq!(cell(&map, 1, "NULL", 0), "{}");
         let cases: Vec<(ArrayRef, &str)> = vec![
             (arr(BooleanArray::from(vec![true])), "true"),
             (arr(UInt8Array::from(vec![5])), "5"),

crates/krishiv-common/src/config/registry.rs

@@ -13,6 +13,7 @@ pub static REGISTRY: &[&ConfigEntry] = &[
     &ANSI_ENABLED,
     &SESSION_TIME_ZONE,
     &STORE_ASSIGNMENT_POLICY,
+    &MAP_KEY_DEDUP_POLICY,
     &TARGET_PARTITION_BYTES,
     &EXEC_BATCH_SIZE,
     &EXEC_AGG_ROW_LAYOUT,
@@ -93,13 +94,15 @@ pub static INDEX: &[(&str, &ConfigEntry)] = &[
     ("spark.sql.session.timeZone", &SESSION_TIME_ZONE),
     ("krishiv.exec.join.radixBits", &EXEC_JOIN_RADIX_BITS),
     ("krishiv.exec.join.rowLayout", &EXEC_JOIN_ROW_LAYOUT),
+    ("spark.sql.mapKeyDedupPolicy", &MAP_KEY_DEDUP_POLICY),
     ("krishiv.sql.session.timeZone", &SESSION_TIME_ZONE),
     ("spark.sql.shuffle.partitions", &SHUFFLE_PARTITIONS_HINT),
     ("krishiv.exec.sort.hashBuckets", &EXEC_SORT_HASH_BUCKETS),
     ("krishiv.io.metadataCacheBytes", &IO_METADATA_CACHE_BYTES),
     ("krishiv.io.multipartPartBytes", &IO_MULTIPART_PART_BYTES),
     ("krishiv.runtimeFilter.enabled", &RUNTIME_FILTER_ENABLED),
     ("krishiv.spill.dropBehindBytes", &SPILL_DROP_BEHIND_BYTES),
+    ("krishiv.sql.mapKeyDedupPolicy", &MAP_KEY_DEDUP_POLICY),
     ("krishiv.io.lateMaterialization", &IO_LATE_MATERIALIZATION),
     (
         "krishiv.memory.spillWaitTimeout",
@@ -263,6 +266,19 @@ pub static STORE_ASSIGNMENT_POLICY: ConfigEntry = ConfigEntry {
     spark_alias: Some("spark.sql.storeAssignmentPolicy"),
 };
 
+/// Map key deduplication policy.
+pub static MAP_KEY_DEDUP_POLICY: ConfigEntry = ConfigEntry {
+    key: "krishiv.sql.mapKeyDedupPolicy",
+    value_type: ValueType::Enum(&["EXCEPTION", "LAST_WIN"]),
+    default: Some(ConfigValue::String(Cow::Borrowed("EXCEPTION"))),
+    doc: "How duplicate map keys are handled by the map functions (`map`, `map_concat`, \
+          `map_from_arrays`, `map_from_entries`, `str_to_map`): `EXCEPTION` fails the query, \
+          `LAST_WIN` keeps the value inserted last.",
+    scope: Scope::Session,
+    version_added: "0.1.0",
+    spark_alias: Some("spark.sql.mapKeyDedupPolicy"),
+};
+
 /// Target partition size, replacing a fixed shuffle partition count.
 pub static TARGET_PARTITION_BYTES: ConfigEntry = ConfigEntry {
     key: "krishiv.exec.targetPartitionBytes",

crates/krishiv-common/src/config/tests.rs

@@ -163,6 +163,31 @@ fn spill_entries_parse_and_default() {
     assert!(c.set("krishiv.spill.maxBytes", "lots").is_err());
 }
 
+#[test]
+fn map_key_dedup_policy_parses_and_defaults() {
+    let mut c = Config::new();
+    assert_eq!(
+        c.get(&MAP_KEY_DEDUP_POLICY).and_then(ConfigValue::as_str),
+        Some("EXCEPTION")
+    );
+    c.set("krishiv.sql.mapKeyDedupPolicy", "LAST_WIN").unwrap();
+    assert_eq!(
+        c.get(&MAP_KEY_DEDUP_POLICY).and_then(ConfigValue::as_str),
+        Some("LAST_WIN")
+    );
+    // The Spark alias sets the same entry.
+    c.set("spark.sql.mapKeyDedupPolicy", "EXCEPTION").unwrap();
+    assert_eq!(
+        c.get(&MAP_KEY_DEDUP_POLICY).and_then(ConfigValue::as_str),
+        Some("EXCEPTION")
+    );
+    assert!(c.set("spark.sql.mapKeyDedupPolicy", "FIRST_WIN").is_err());
+    assert_eq!(
+        c.get(&MAP_KEY_DEDUP_POLICY).and_then(ConfigValue::as_str),
+        Some("EXCEPTION")
+    );
+}
+
 #[test]
 fn invalid_values_are_typed_errors() {
     let mut c = Config::new();

crates/krishiv-exec/src/physical/mod.rs

@@ -204,6 +204,10 @@ pub struct ExprEnv {
     pub query_start_micros: i64,
     /// The session user (`current_user()`).
     pub user: Arc<str>,
+    /// `spark.sql.mapKeyDedupPolicy = LAST_WIN`: duplicate map keys keep
+    /// the value inserted last. A plan that sets it needs the
+    /// `map_key_dedup_policy` codec feature.
+    pub map_key_dedup_last_win: bool,
 }
 
 /// A node of a physical plan. Children come first in each variant;
@@ -546,7 +550,14 @@ plan_data!(
         Partitioned,
     }
 );
-plan_data!(struct ExprEnv { ansi, legacy_statistical_aggregate, zone, query_start_micros, user });
+plan_data!(struct ExprEnv {
+    ansi,
+    legacy_statistical_aggregate,
+    zone,
+    query_start_micros,
+    user,
+    map_key_dedup_last_win
+});
 plan_data!(enum PhysNode {
     Scan(scan),
     OneRow,

crates/krishiv-exec/src/planner/eval/func.rs

@@ -8,6 +8,7 @@ use arrow_array::RecordBatch;
 use arrow_schema::DataType as ArrowType;
 use krishiv_functions::scalar::{Arg, BindArg, BindOptions, ScalarFunction, bind_scalar};
 use krishiv_kernels::cast::FixedOffset;
+use krishiv_kernels::map::DedupPolicy;
 use krishiv_plan::{Expr, ExprNode, ResolvedFunction};
 
 use super::strings::to_view;
@@ -54,6 +55,11 @@ pub(super) fn compile(
         .with_kernel(input.env.kernels)
         .with_time_zone(Arc::new(zone))
         .with_query_start(input.env.query_start_micros)
+        .with_map_key_dedup_policy(if input.env.map_key_dedup_last_win {
+            DedupPolicy::LastWin
+        } else {
+            DedupPolicy::Exception
+        })
         .with_user(Arc::clone(&input.env.user))
         .with_seed_salt(Arc::clone(&input.env.seed_salt));
     let function = bind_scalar(name, &bind_args, &opts).map_err(|err| {

crates/krishiv-exec/src/planner/eval/hof.rs

@@ -447,6 +447,7 @@ mod tests {
             kernels: KernelCtx::new(),
             query_start_micros: 0,
             user: Arc::from("test"),
+            map_key_dedup_last_win: false,
             seed_salt: Arc::new(std::sync::atomic::AtomicU64::new(0)),
         }
     }

crates/krishiv-exec/src/planner/eval/mod.rs

@@ -40,7 +40,7 @@ use arrow_array::cast::AsArray;
 use arrow_array::{Array, ArrayRef, BooleanArray, RecordBatch, UInt32Array, new_null_array};
 use arrow_schema::{DataType as ArrowType, SchemaRef};
 use krishiv_common::config::{
-    ANSI_ENABLED, Config, LEGACY_STATISTICAL_AGGREGATE, SESSION_TIME_ZONE,
+    ANSI_ENABLED, Config, LEGACY_STATISTICAL_AGGREGATE, MAP_KEY_DEDUP_POLICY, SESSION_TIME_ZONE,
 };
 use krishiv_kernels::KernelCtx;
 use krishiv_plan::{
@@ -178,6 +178,10 @@ pub(crate) struct EvalEnv {
     pub(crate) query_start_micros: i64,
     /// The session user (`current_user()`).
     pub(crate) user: Arc<str>,
+    /// `spark.sql.mapKeyDedupPolicy = LAST_WIN`: duplicate map keys keep
+    /// the value inserted last. It travels in the plan (`ExprEnv`), so a
+    /// worker instantiating a plan without the feature refuses it.
+    pub(crate) map_key_dedup_last_win: bool,
     /// Counts the unseeded random binds of one plan instantiation (see
     /// [`BindOptions::seed_salt`](krishiv_functions::scalar::BindOptions::seed_salt)).
     pub(crate) seed_salt: Arc<AtomicU64>,
@@ -197,6 +201,10 @@ impl EvalEnv {
             .get(&SESSION_TIME_ZONE)
             .and_then(|v| v.as_str())
             .and_then(|z| krishiv_kernels::cast::FixedOffset::parse(z).map(|o| o.0));
+        let map_key_dedup_last_win = config
+            .get(&MAP_KEY_DEDUP_POLICY)
+            .and_then(|v| v.as_str())
+            .is_some_and(|s| s.eq_ignore_ascii_case("LAST_WIN"));
         let query_start_micros = std::time::SystemTime::now()
             .duration_since(std::time::UNIX_EPOCH)
             .ok()
@@ -209,6 +217,7 @@ impl EvalEnv {
             kernels: KernelCtx::new(),
             query_start_micros,
             user,
+            map_key_dedup_last_win,
             seed_salt: Arc::new(AtomicU64::new(0)),
         }
     }
@@ -224,6 +233,7 @@ impl EvalEnv {
             kernels: KernelCtx::new(),
             query_start_micros: env.query_start_micros,
             user: Arc::clone(&env.user),
+            map_key_dedup_last_win: env.map_key_dedup_last_win,
             seed_salt: Arc::new(AtomicU64::new(0)),
         }
     }
@@ -236,6 +246,7 @@ impl EvalEnv {
             zone: self.zone,
             query_start_micros: self.query_start_micros,
             user: Arc::clone(&self.user),
+            map_key_dedup_last_win: self.map_key_dedup_last_win,
         }
     }
 }

crates/krishiv-exec/src/profile/tests.rs

@@ -21,6 +21,7 @@ fn env() -> ExprEnv {
         zone: Some(0),
         query_start_micros: 0,
         user: Arc::from("u"),
+        map_key_dedup_last_win: false,
     }
 }
 

crates/krishiv-exec/tests/planner/physical.rs

@@ -167,6 +167,7 @@ fn env_data() -> ExprEnv {
         zone: Some(0),
         query_start_micros: 0,
         user: Arc::from("u"),
+        map_key_dedup_last_win: false,
     }
 }
 

crates/krishiv-exec/tests/planner/write.rs

@@ -127,6 +127,7 @@ fn query_results_are_written_committed_and_read_back() {
             zone: None,
             query_start_micros: 0,
             user: Arc::from("u"),
+            map_key_dedup_last_win: false,
         },
     );
     let worker = WorkerContext::new(Arc::new(MemoryTables::new())).with_write_sink(other);

crates/krishiv-functions/CLAUDE.md

@@ -23,8 +23,8 @@ Spark-compatible function library: scalar, aggregate, window and generator funct
   `Arc<dyn ScalarFunction>` (`return_type`, `invoke(&[Arg], rows)`), IR-free (Arrow types only) so the physical planner
   (krishiv-exec) calls it per batch. Arguments arrive after the analyzer's implicit casts; strings/binary in any Arrow
   carrier (read as `Utf8`/`Binary`), results in plain carriers. One family module per kernel family (`string`, `chars`, `codec`,
-  `datetime`, `math`, `arith`, `hash`, `session`, `regex`), each with `NAMES` and a `bind` over the argument shape. `BindOptions` holds ANSI, `KernelCtx` (Spark
-  version), session zone, the query start instant (`current_date`, `current_timestamp`, `localtimestamp`), the session user (`current_user`, `user`, `session_user`; family `session`) and `decimalOperations.allowPrecisionLoss`. `round`/`ceil`/`floor` bind
+  `datetime`, `math`, `arith`, `hash`, `session`, `regex`, `map`), each with `NAMES` and a `bind` over the argument shape. `BindOptions` holds ANSI, `KernelCtx` (Spark
+  version), session zone, the query start instant (`current_date`, `current_timestamp`, `localtimestamp`), the session user (`current_user`, `user`, `session_user`; family `session`), `decimalOperations.allowPrecisionLoss` and `spark.sql.mapKeyDedupPolicy` (the map family). `round`/`ceil`/`floor` bind
   their scale from a `BindArg::Literal`. Operators, `CAST` and aggregates are not bound here.
 
 - `python_wrappers`: generates `python/krishiv/sql/_generated_functions.py` and `.pyi` (PySpark names and parameters) from the
@@ -41,7 +41,7 @@ Paths are relative to this crate.
 - `src/registry/conditional.rs` conditional evaluators; `src/registry/hashing.rs` hash signatures
 - `src/registry/catalog.rs` catalog rows loaded from `vendored/spark-functions.tsv`
 - `src/scalar/mod.rs` `bind_scalar`; `src/scalar/args.rs` argument handling shared by bindings
-- `src/scalar/{string,chars,codec,datetime,math,arith,hash,session,regex}.rs` family bindings (`NAMES` and `bind`); `src/scalar/regex/java.rs` Java-to-`fancy_regex` translation
+- `src/scalar/{string,chars,codec,datetime,math,arith,hash,map,session,regex}.rs` family bindings (`NAMES` and `bind`); `src/scalar/regex/java.rs` Java-to-`fancy_regex` translation
 - `vendored/spark-functions.tsv` generated from Spark runs; never edit by hand
 
 ## Extension points
@@ -106,6 +106,6 @@ Paths are relative to this crate.
   (`spark.sql.legacy.javaCharsets` and `...codingErrorAction` are not read) and report unmappable input. `split_part`/`substring_index`
   with `Int.MIN` return the algorithm's value where the JVM raises; `initcap`/`mask` classify non-ASCII with the toolchain's Unicode
   data (pinned by `toolchain_unicode_is_pinned`). A string is always valid UTF-8 here, so the validators only fail on `binary` input.
-  Not implemented (`KnownUnsupported`): `format_string`/`printf`, `parse_url`/`try_parse_url`, `sentences`, `split`, `str_to_map`
-  (array/map results or a Java formatter), `collate`/`collation`/`collations`, `randstr` (no seeded random source), and the
+  Not implemented (`KnownUnsupported`): `format_string`/`printf`, `parse_url`/`try_parse_url`, `sentences`, `split`
+  (array results or a Java formatter), `collate`/`collation`/`collations`, `randstr` (no seeded random source), and the
   `decode(expr, search, result, ...)` form (analysis error `KRISHIV_UNSUPPORTED_FEATURE`).

crates/krishiv-functions/Cargo.toml

@@ -34,3 +34,6 @@ ignored = ["regex-automata", "regex-syntax"]
 
 [lints]
 workspace = true
+
+[dev-dependencies]
+arrow-buffer.workspace = true

crates/krishiv-functions/src/python_wrappers/mod.rs

@@ -299,6 +299,19 @@ fn render_function(row: &Row, params: &[Param]) -> (String, String) {
     if let Some(message) = row.deprecated {
         body.push_str(&format!("    _deprecated(\"{message}\")\n"));
     }
+    if row.py == "str_to_map" {
+        // PySpark sends the default pair delimiter when only `keyValueDelim`
+        // is given, so the third argument keeps its position.
+        body.push_str("    if pairDelim is None and keyValueDelim is not None:\n");
+        let defaulted = vec![
+            format!("\"{}\"", row.registry),
+            "_named(text, \"text\")".to_owned(),
+            "_lit(\",\", \"pairDelim\")".to_owned(),
+            "_named(keyValueDelim, \"keyValueDelim\")".to_owned(),
+        ];
+        body.push_str(&wrap("        return _call(", &defaulted, ")", 8));
+        body.push('\n');
+    }
     body.push_str(&wrap("    return _call(", &args, ")", 4));
     body.push('\n');
     let def = format!("{}\n{body}", wrap(&head, &declarations, ") -> Column:", 0));

crates/krishiv-functions/src/python_wrappers/table.rs

@@ -147,6 +147,10 @@ pub(super) fn rows() -> Vec<Row> {
         "log1p",
         "log2",
         "lower",
+        "map_entries",
+        "map_from_entries",
+        "map_keys",
+        "map_values",
         "max",
         "md5",
         "min",
@@ -284,6 +288,7 @@ pub(super) fn rows() -> Vec<Row> {
         "array",
         "coalesce",
         "hash",
+        "map_concat",
         "named_struct",
         "struct",
         "xxhash64",
@@ -305,6 +310,13 @@ pub(super) fn rows() -> Vec<Row> {
             "col: ColumnOrName, errMsg: Column | str = None",
         ),
         Row::new("raise_error", "errMsg: Column | str"),
+        Row::new("create_map", "*cols: ColumnOrName").to("map"),
+        Row::new("map_contains_key", "col: ColumnOrName, value: Any"),
+        Row::new("map_from_arrays", "col1: ColumnOrName, col2: ColumnOrName"),
+        Row::new(
+            "str_to_map",
+            "text: ColumnOrName, pairDelim: ColumnOrName = None, keyValueDelim: ColumnOrName = None",
+        ),
         Row::new("bitwise_not", C).to("~"),
         Row::new("mean", C).to("avg"),
         Row::new("btrim", "str: ColumnOrName, trim: ColumnOrName = None"),

crates/krishiv-functions/src/registry/defs.rs

@@ -843,6 +843,7 @@ pub(super) fn specs() -> Vec<Spec> {
     out.extend(aggregates());
     out.extend(windows());
     out.extend(super::conditional::specs());
+    out.extend(super::map::specs());
     out.extend(super::math::specs());
     out.extend(super::hashing::specs());
     out.extend(super::bitwise::specs());