[AURON #2154] Implement native function of str_to_map

auron-core/src/main/java/org/apache/auron/jni/JniBridge.java

@@ -23,6 +23,8 @@
 import java.nio.ByteBuffer;
 import java.util.List;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.regex.Pattern;
+import java.util.regex.PatternSyntaxException;
 import org.apache.auron.configuration.AuronConfiguration;
 import org.apache.auron.configuration.ConfigOption;
 import org.apache.auron.functions.AuronUDFWrapperContext;
@@ -39,6 +41,7 @@
 @SuppressWarnings("unused")
 public class JniBridge {
     private static final ConcurrentHashMap<String, Object> resourcesMap = new ConcurrentHashMap<>();
+    private static final ConcurrentHashMap<String, Pattern> regexCache = new ConcurrentHashMap<>();
 
     private static final List<BufferPoolMXBean> directMXBeans =
             ManagementFactory.getPlatformMXBeans(BufferPoolMXBean.class);
@@ -134,6 +137,38 @@ public static String getEngineName() {
         return AuronAdaptor.getInstance().getEngineName();
     }
 
+    public static String[] strToMapSplit(String text, String pairDelim, String keyValueDelim) {
+        Pattern pairPattern = getCachedPattern(pairDelim, "pairDelim");
+        Pattern keyValuePattern = getCachedPattern(keyValueDelim, "keyValueDelim");
+
+        String[] entries = pairPattern.split(text, -1);
+        String[] flattened = new String[entries.length * 2];
+        for (int i = 0; i < entries.length; i++) {
+            String[] kv = keyValuePattern.split(entries[i], 2);
+            flattened[i * 2] = kv[0];
+            flattened[i * 2 + 1] = kv.length > 1 ? kv[1] : null;
+        }
+        return flattened;
+    }
+
+    private static Pattern getCachedPattern(String pattern, String argName) {
+        Pattern cached = regexCache.get(pattern);
+        if (cached != null) {
+            return cached;
+        }
+
+        final Pattern compiled;
+        try {
+            compiled = Pattern.compile(pattern);
+        } catch (PatternSyntaxException e) {
+            throw new RuntimeException(
+                    "str_to_map " + argName + " arg must be a valid Java regex: " + e.getMessage(), e);
+        }
+
+        Pattern existing = regexCache.putIfAbsent(pattern, compiled);
+        return existing != null ? existing : compiled;
+    }
+
     static <T> T getConfValue(String confKey) {
         Class<? extends AuronConfiguration> confClass =
                 AuronAdaptor.getInstance().getAuronConfiguration().getClass();

native-engine/auron-jni-bridge/src/jni_bridge.rs

@@ -642,6 +642,8 @@ pub struct JniBridge<'a> {
     pub method_booleanConf_ret: ReturnType,
     pub method_stringConf: JStaticMethodID,
     pub method_stringConf_ret: ReturnType,
+    pub method_strToMapSplit: JStaticMethodID,
+    pub method_strToMapSplit_ret: ReturnType,
     pub method_getEngineName: JStaticMethodID,
     pub method_getEngineName_ret: ReturnType,
 }
@@ -757,6 +759,12 @@ impl<'a> JniBridge<'a> {
                 "(Ljava/lang/String;)Ljava/lang/String;",
             )?,
             method_stringConf_ret: ReturnType::Object,
+            method_strToMapSplit: env.get_static_method_id(
+                class,
+                "strToMapSplit",
+                "(Ljava/lang/String;Ljava/lang/String;Ljava/lang/String;)[Ljava/lang/String;",
+            )?,
+            method_strToMapSplit_ret: ReturnType::Object,
             method_getEngineName: env.get_static_method_id(
                 class,
                 "getEngineName",

native-engine/datafusion-ext-functions/Cargo.toml

@@ -29,11 +29,13 @@ arrow = { workspace = true }
 auron-jni-bridge = { workspace = true }
 datafusion = { workspace = true }
 datafusion-ext-commons = { workspace = true }
+jni = { workspace = true }
 
 itertools = { workspace = true }
 log = { workspace = true }
 num = { workspace = true }
 paste = { workspace = true }
+regex = "1.12.3"
 serde_json = { workspace = true }
 sonic-rs = { workspace = true }
 chrono = "0.4.44"

native-engine/datafusion-ext-functions/src/lib.rs

@@ -67,6 +67,7 @@ pub fn create_auron_ext_function(
         "Spark_MapConcat" => Arc::new(spark_map::map_concat),
         "Spark_MapFromArrays" => Arc::new(spark_map::map_from_arrays),
         "Spark_MapFromEntries" => Arc::new(spark_map::map_from_entries),
+        "Spark_StrToMap" => Arc::new(spark_map::str_to_map),
         "Spark_StringSpace" => Arc::new(spark_strings::string_space),
         "Spark_StringRepeat" => Arc::new(spark_strings::string_repeat),
         "Spark_StringSplit" => Arc::new(spark_strings::string_split),