feat: Support Spark expression str_to_map (#3654)

Author: unknowntpo

docs/source/user-guide/latest/expressions.md

@@ -264,6 +264,7 @@ Comet supports using the following aggregate functions within window contexts wi
 | MapEntries    |
 | MapValues     |
 | MapFromArrays |
+| StringToMap   |
 
 ## Struct Expressions
 

native/core/src/execution/jni_api.rs

@@ -54,6 +54,7 @@ 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::hex::SparkHex;
 use datafusion_spark::function::math::width_bucket::SparkWidthBucket;
@@ -565,6 +566,7 @@ fn register_datafusion_spark_function(session_ctx: &SessionContext) {
     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(SparkBin::default()));
+    session_ctx.register_udf(ScalarUDF::new_from_impl(SparkStrToMap::default()));
 }
 
 /// Prepares arrow arrays for output.

spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala

@@ -135,7 +135,8 @@ object QueryPlanSerde extends Logging with CometExprShim with CometTypeShim {
     classOf[MapValues] -> CometMapValues,
     classOf[MapFromArrays] -> CometMapFromArrays,
     classOf[MapContainsKey] -> CometMapContainsKey,
-    classOf[MapFromEntries] -> CometMapFromEntries)
+    classOf[MapFromEntries] -> CometMapFromEntries,
+    classOf[StringToMap] -> CometStrToMap)
 
   private[comet] val structExpressions: Map[Class[_ <: Expression], CometExpressionSerde[_]] =
     Map(

spark/src/main/scala/org/apache/comet/serde/maps.scala

@@ -159,3 +159,10 @@ object CometMapFromEntries extends CometScalarFunction[MapFromEntries]("map_from
     Compatible(None)
   }
 }
+
+object CometStrToMap extends CometScalarFunction[StringToMap]("str_to_map") {
+
+  override def getSupportLevel(expr: StringToMap): SupportLevel = {
+    Compatible(None)
+  }
+}

spark/src/test/resources/sql-tests/expressions/map/str_to_map.sql

@@ -0,0 +1,108 @@
+-- Licensed to the Apache Software Foundation (ASF) under one
+-- or more contributor license agreements.  See the NOTICE file
+-- distributed with this work for additional information
+-- regarding copyright ownership.  The ASF licenses this file
+-- to you under the Apache License, Version 2.0 (the
+-- "License"); you may not use this file except in compliance
+-- with the License.  You may obtain a copy of the License at
+--
+--   http://www.apache.org/licenses/LICENSE-2.0
+--
+-- Unless required by applicable law or agreed to in writing,
+-- software distributed under the License is distributed on an
+-- "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+-- KIND, either express or implied.  See the License for the
+-- specific language governing permissions and limitations
+-- under the License.
+
+-- Tests for Spark-compatible str_to_map function
+-- https://spark.apache.org/docs/latest/api/sql/index.html#str_to_map
+--
+-- Test cases derived from Spark test("StringToMap"):
+-- https://github.com/apache/spark/blob/v4.0.0/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/ComplexTypeSuite.scala#L525-L618
+
+-- ConfigMatrix: parquet.enable.dictionary=false,true
+
+statement
+CREATE TABLE test_str_to_map(
+  s STRING,
+  pair_delim STRING,
+  key_value_delim STRING,
+  s_char CHAR(16),
+  pair_delim_char CHAR(4),
+  key_value_delim_char CHAR(4),
+  s_varchar VARCHAR(16),
+  pair_delim_varchar VARCHAR(4),
+  key_value_delim_varchar VARCHAR(4)
+) USING parquet
+
+statement
+INSERT INTO test_str_to_map VALUES
+  ('a:1,b:2,c:3', ',', ':', 'a:1,b:2,c:3', ',', ':', 'a:1,b:2,c:3', ',', ':'),
+  ('x=9;y=8', ';', '=', 'x=9;y=8', ';', '=', 'x=9;y=8', ';', '='),
+  (NULL, ',', ':', NULL, ',', ':', NULL, ',', ':')
+
+-- s0: Basic test with default delimiters
+query
+SELECT str_to_map('a:1,b:2,c:3')
+
+-- s1: Preserve spaces in values
+query
+SELECT str_to_map('a: ,b:2')
+
+-- s2: Custom key-value delimiter '='
+query
+SELECT str_to_map('a=1,b=2,c=3', ',', '=')
+
+-- s3: Empty string returns map with empty key and NULL value
+query
+SELECT str_to_map('', ',', '=')
+
+-- s4: Custom pair delimiter '_'
+query
+SELECT str_to_map('a:1_b:2_c:3', '_', ':')
+
+-- s5: Single key without value returns NULL value
+query
+SELECT str_to_map('a')
+
+-- s6: Custom delimiters '&' and '='
+query
+SELECT str_to_map('a=1&b=2&c=3', '&', '=')
+
+-- Duplicate keys: EXCEPTION policy (Spark 3.0+ default)
+-- TODO: Add LAST_WIN policy tests when spark.sql.mapKeyDedupPolicy config is supported
+-- query
+-- SELECT str_to_map('a:1,b:2,a:3')
+
+-- NULL input returns NULL
+query
+SELECT str_to_map(NULL, ',', ':')
+
+-- Explicit 3-arg form
+query
+SELECT str_to_map('a:1,b:2,c:3', ',', ':')
+
+-- Missing key-value delimiter results in NULL value
+query
+SELECT str_to_map('a,b:2', ',', ':')
+
+-- Multi-row test
+query
+SELECT str_to_map(s) FROM test_str_to_map
+
+-- Rows with per-row delimiters
+query
+SELECT str_to_map(s, pair_delim, key_value_delim) FROM test_str_to_map
+
+-- STRING input with literal delimiters
+query
+SELECT str_to_map(s, ',', ':') FROM test_str_to_map
+
+-- CHAR input and delimiters with per-row delimiter values
+query
+SELECT str_to_map(s_char, pair_delim_char, key_value_delim_char) FROM test_str_to_map
+
+-- VARCHAR input and delimiters with per-row delimiter values
+query
+SELECT str_to_map(s_varchar, pair_delim_varchar, key_value_delim_varchar) FROM test_str_to_map