diff --git a/src/paimon/common/utils/rapidjson_util.h b/src/paimon/common/utils/rapidjson_util.h
index 08824661..7a6c4400 100644
--- a/src/paimon/common/utils/rapidjson_util.h
+++ b/src/paimon/common/utils/rapidjson_util.h
@@ -35,16 +35,13 @@
 #include "rapidjson/writer.h"
 
 namespace paimon {
+
 class RapidJsonUtil {
  public:
     RapidJsonUtil() = delete;
     ~RapidJsonUtil() = delete;
 
-    // supports vector and map and optional, if T is custom type, T must have ToJson()
-    // noted that rapidjson does not support map with non string key (will
-    // trigger assert in rapidjson: Assertion `name.IsString()' failed)
-    // therefore, RapidJsonUtil convert key to string in serialize and convert string to key type in
-    // deserialize
+    // if T is custom type, T must have ToJson()
     template <typename T>
     static inline Status ToJsonString(const T& obj, std::string* json_str) {
         rapidjson::Document doc;
@@ -53,6 +50,9 @@ class RapidJsonUtil {
         try {
             if constexpr (is_pointer<T>::value) {
                 value = obj->ToJson(&allocator);
+            } else if constexpr (std::is_same_v<T, std::map<std::string, std::string>>) {
+                *json_str = MapToJsonString(obj);
+                return Status::OK();
             } else {
                 value = obj.ToJson(&allocator);
             }
@@ -67,19 +67,27 @@ class RapidJsonUtil {
         return Status::OK();
     }
 
-    // supports vector and map, if T is custom type, T must have FromJson()
+    // if T is custom type, T must have FromJson()
     template <typename T>
     static inline Status FromJsonString(const std::string& json_str, T* obj) {
-        rapidjson::Document doc;
-        if (!obj || !FromJson(json_str, &doc)) {
-            return Status::Invalid("deserialize failed: ", json_str);
+        if (!obj) {
+            return Status::Invalid("deserialize failed: obj is nullptr");
         }
-        try {
-            obj->FromJson(doc);
-        } catch (const std::invalid_argument& e) {
-            return Status::Invalid("deserialize failed, possibly type incompatible: ", e.what());
-        } catch (...) {
-            return Status::Invalid("deserialize failed, reason unknown: ", json_str);
+        if constexpr (std::is_same_v<T, std::map<std::string, std::string>>) {
+            PAIMON_ASSIGN_OR_RAISE(*obj, MapFromJsonString(json_str));
+        } else {
+            rapidjson::Document doc;
+            if (!FromJson(json_str, &doc)) {
+                return Status::Invalid("deserialize failed: ", json_str);
+            }
+            try {
+                obj->FromJson(doc);
+            } catch (const std::invalid_argument& e) {
+                return Status::Invalid("deserialize failed, possibly type incompatible: ",
+                                       e.what());
+            } catch (...) {
+                return Status::Invalid("deserialize failed, reason unknown: ", json_str);
+            }
         }
         return Status::OK();
     }
@@ -140,6 +148,42 @@ class RapidJsonUtil {
 
     template <typename T>
     static T GetValue(const rapidjson::Value& value);
+
+    static std::string MapToJsonString(const std::map<std::string, std::string>& map) {
+        rapidjson::Document d;
+        d.SetObject();
+        rapidjson::Document::AllocatorType& allocator = d.GetAllocator();
+
+        for (const auto& kv : map) {
+            d.AddMember(rapidjson::Value(kv.first.c_str(), allocator),
+                        rapidjson::Value(kv.second.c_str(), allocator), allocator);
+        }
+
+        rapidjson::StringBuffer buffer;
+        rapidjson::Writer<rapidjson::StringBuffer> writer(buffer);
+        d.Accept(writer);
+
+        return buffer.GetString();
+    }
+    static Result<std::map<std::string, std::string>> MapFromJsonString(
+        const std::string& json_str) {
+        rapidjson::Document doc;
+        doc.Parse(json_str.c_str());
+        if (doc.HasParseError() || !doc.IsObject()) {
+            return Status::Invalid("deserialize failed: parse error or not JSON object: ",
+                                   json_str);
+        }
+
+        std::map<std::string, std::string> result;
+        for (auto it = doc.MemberBegin(); it != doc.MemberEnd(); ++it) {
+            if (!it->name.IsString() || !it->value.IsString()) {
+                return Status::Invalid(
+                    "deserialize failed: non-string key or value in JSON object: ", json_str);
+            }
+            result[it->name.GetString()] = it->value.GetString();
+        }
+        return result;
+    }
 };
 
 template <typename T>
diff --git a/src/paimon/common/utils/rapidjson_util_test.cpp b/src/paimon/common/utils/rapidjson_util_test.cpp
index 57d09471..b143653b 100644
--- a/src/paimon/common/utils/rapidjson_util_test.cpp
+++ b/src/paimon/common/utils/rapidjson_util_test.cpp
@@ -22,6 +22,7 @@
 #include <vector>
 
 #include "gtest/gtest.h"
+#include "paimon/testing/utils/testharness.h"
 #include "rapidjson/allocators.h"
 #include "rapidjson/document.h"
 #include "rapidjson/rapidjson.h"
@@ -129,4 +130,15 @@ TEST(RapidJsonUtilTest, TestSerializeAndDeserialize) {
     ASSERT_EQ(2.333, non_exist_value);
 }
 
+TEST(RapidJsonUtilTest, TestMapJsonString) {
+    std::map<std::string, std::string> m1 = {{"key1", "value1"}, {"key2", "value2"}};
+    std::string result;
+    ASSERT_OK(RapidJsonUtil::ToJsonString(m1, &result));
+    ASSERT_EQ(result, "{\"key1\":\"value1\",\"key2\":\"value2\"}");
+
+    std::map<std::string, std::string> m2;
+    ASSERT_OK(RapidJsonUtil::FromJsonString(result, &m2));
+    ASSERT_EQ(m1, m2);
+}
+
 }  // namespace paimon::test
diff --git a/src/paimon/common/utils/string_utils.cpp b/src/paimon/common/utils/string_utils.cpp
index c387c024..88312d03 100644
--- a/src/paimon/common/utils/string_utils.cpp
+++ b/src/paimon/common/utils/string_utils.cpp
@@ -40,6 +40,16 @@ std::string StringUtils::Replace(const std::string& text, const std::string& sea
     return str;
 }
 
+std::string StringUtils::ReplaceLast(const std::string& text, const std::string& old_str,
+                                     const std::string& new_str) {
+    std::string str = text;
+    size_t pos = str.rfind(old_str);
+    if (pos != std::string::npos) {
+        str.replace(pos, old_str.size(), new_str);
+    }
+    return str;
+}
+
 bool StringUtils::StartsWith(const std::string& str, const std::string& prefix, size_t start_pos) {
     return (str.size() >= prefix.size()) && (str.compare(start_pos, prefix.size(), prefix) == 0);
 }
diff --git a/src/paimon/common/utils/string_utils.h b/src/paimon/common/utils/string_utils.h
index 0f9f478f..f0b01090 100644
--- a/src/paimon/common/utils/string_utils.h
+++ b/src/paimon/common/utils/string_utils.h
@@ -98,6 +98,9 @@ class PAIMON_EXPORT StringUtils {
     static std::string Replace(const std::string& text, const std::string& search_string,
                                const std::string& replacement, int32_t max);
 
+    static std::string ReplaceLast(const std::string& text, const std::string& old_str,
+                                   const std::string& new_str);
+
     static bool StartsWith(const std::string& str, const std::string& prefix, size_t start_pos = 0);
 
     static bool EndsWith(const std::string& str, const std::string& suffix);
diff --git a/src/paimon/common/utils/string_utils_test.cpp b/src/paimon/common/utils/string_utils_test.cpp
index 3b421f76..18bb44e7 100644
--- a/src/paimon/common/utils/string_utils_test.cpp
+++ b/src/paimon/common/utils/string_utils_test.cpp
@@ -113,6 +113,28 @@ TEST_F(StringUtilsTest, TestReplaceAll) {
     }
 }
 
+TEST_F(StringUtilsTest, TestReplaceLast) {
+    {
+        std::string origin = "a/b/c//";
+        std::string expect = "a/b/c/_";
+        std::string actual = StringUtils::ReplaceLast(origin, "/", "_");
+        ASSERT_EQ(expect, actual);
+    }
+    {
+        std::string origin = "a/b/c//";
+        std::string expect = "a/b/c//";
+        std::string actual = StringUtils::ReplaceLast(origin, "_", "/");
+        ASSERT_EQ(expect, actual);
+    }
+
+    {
+        std::string origin = "how is is you";
+        std::string expect = "how is are you";
+        std::string actual = StringUtils::ReplaceLast(origin, "is", "are");
+        ASSERT_EQ(expect, actual);
+    }
+}
+
 TEST_F(StringUtilsTest, TestReplaceWithMaxCount) {
     {
         std::string origin = "how is is you";