refactor(utils): consolidate string and option handling

src/paimon/common/data/variant/variant_access_utils.cpp

@@ -26,6 +26,7 @@
 #include "fmt/format.h"
 #include "paimon/common/data/variant/variant_defs.h"
 #include "paimon/common/types/data_field.h"
+#include "paimon/common/utils/string_utils.h"
 
 namespace paimon {
 
@@ -61,7 +62,7 @@ std::vector<std::string> SplitDescription(const std::string& description) {
 }
 
 bool HasAccessDescription(const std::shared_ptr<arrow::Field>& field) {
-    return GetDescription(field).rfind(VariantAccessUtils::kMetadataKey, 0) == 0;
+    return StringUtils::StartsWith(GetDescription(field), VariantAccessUtils::kMetadataKey);
 }
 
 }  // namespace

src/paimon/common/types/data_type_json_parser.cpp

@@ -19,7 +19,6 @@
 
 #include "paimon/common/types/data_type_json_parser.h"
 
-#include <algorithm>
 #include <cctype>
 #include <cstddef>
 #include <cstdint>
@@ -331,10 +330,7 @@ std::vector<Token> Tokenize(const std::string& chars) {
                 builder.clear();
                 cursor = ConsumeIdentifier(chars, cursor, builder);
                 auto token = builder.str();
-                auto normalized_token = token;
-                std::transform(normalized_token.begin(), normalized_token.end(),
-                               normalized_token.begin(),
-                               [](unsigned char c) { return std::toupper(c); });
+                std::string normalized_token = StringUtils::ToUpperCase(token);
                 if (Keywords().find(normalized_token) != Keywords().end()) {
                     tokens.emplace_back(TokenType::KEYWORD, cursor, normalized_token);
                 } else {

src/paimon/common/utils/options_utils.h

@@ -89,13 +89,29 @@ class OptionsUtils {
         return value.status();
     }
 
+    static Result<std::string> GetNonEmptyValueFromMap(
+        const std::map<std::string, std::string>& key_value_map, const std::string& key) {
+        Result<std::string> value = GetValueFromMap<std::string>(key_value_map, key);
+        if (!value.ok()) {
+            return value.status();
+        }
+        if (value.value().empty()) {
+            return Status::Invalid(fmt::format("value for key {} must not be empty", key));
+        }
+        return value.value();
+    }
+
     /// Fetch options with specific prefix and remove prefix for key.
+    /// @param prefix Prefix used to select options and removed from the returned keys.
+    /// @param options Options to select from.
+    /// @return Options whose keys start with and are longer than `prefix`, with the prefix removed
+    /// from each key.
     static std::map<std::string, std::string> FetchOptionsWithPrefix(
         const std::string& prefix, const std::map<std::string, std::string>& options) {
         std::map<std::string, std::string> options_with_prefix;
-        int64_t prefix_len = prefix.size();
+        const std::string::size_type prefix_len = prefix.size();
         for (const auto& [key, value] : options) {
-            if (StringUtils::StartsWith(key, prefix)) {
+            if (key.size() > prefix_len && StringUtils::StartsWith(key, prefix)) {
                 options_with_prefix[key.substr(prefix_len)] = value;
             }
         }

src/paimon/common/utils/options_utils_test.cpp

@@ -84,9 +84,19 @@ TEST(OptionsUtilsTest, TestGetOptionalValueFromMap) {
 }
 
 TEST(OptionsUtilsTest, TestFetchOptionsWithPrefix) {
-    std::map<std::string, std::string> options = {{"key1", "value1"}, {"test.key2", "value2"}};
+    std::map<std::string, std::string> options = {
+        {"key1", "value1"}, {"test.", "empty-key"}, {"test.key2", "value2"}};
     auto new_options = OptionsUtils::FetchOptionsWithPrefix("test.", options);
     std::map<std::string, std::string> expected = {{"key2", "value2"}};
     ASSERT_EQ(expected, new_options);
 }
+
+TEST(OptionsUtilsTest, TestGetNonEmptyValueFromMap) {
+    std::map<std::string, std::string> options = {{"present", "value"}, {"empty", ""}};
+    ASSERT_OK_AND_ASSIGN(std::string value,
+                         OptionsUtils::GetNonEmptyValueFromMap(options, "present"));
+    ASSERT_EQ("value", value);
+    ASSERT_TRUE(OptionsUtils::GetNonEmptyValueFromMap(options, "missing").status().IsNotExist());
+    ASSERT_TRUE(OptionsUtils::GetNonEmptyValueFromMap(options, "empty").status().IsInvalid());
+}
 }  // namespace paimon::test

src/paimon/common/utils/string_utils.cpp

@@ -30,8 +30,28 @@
 #include "paimon/status.h"
 
 namespace paimon {
+namespace {
+
+bool IsTrimCharacter(unsigned char c) {
+    // Match the characters removed by Java String::trim for the ASCII strings handled here.
+    return c <= 0x20;
+}
+
+char ToAsciiLower(unsigned char c) {
+    return c >= 'A' && c <= 'Z' ? static_cast<char>(c + ('a' - 'A')) : static_cast<char>(c);
+}
+
+char ToAsciiUpper(unsigned char c) {
+    return c >= 'a' && c <= 'z' ? static_cast<char>(c - ('a' - 'A')) : static_cast<char>(c);
+}
+
+}  // namespace
+
 std::string StringUtils::Replace(const std::string& text, const std::string& search_string,
                                  const std::string& replacement, int32_t max) {
+    if (text.empty() || search_string.empty() || max == 0) {
+        return text;
+    }
     std::string str = text;
     size_t pos = str.find(search_string);
     int32_t count = 0;
@@ -45,6 +65,9 @@ std::string StringUtils::Replace(const std::string& text, const std::string& sea
 
 std::string StringUtils::ReplaceLast(const std::string& text, const std::string& old_str,
                                      const std::string& new_str) {
+    if (text.empty() || old_str.empty()) {
+        return text;
+    }
     std::string str = text;
     size_t pos = str.rfind(old_str);
     if (pos != std::string::npos) {
@@ -54,7 +77,8 @@ std::string StringUtils::ReplaceLast(const std::string& text, const std::string&
 }
 
 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);
+    return start_pos <= str.size() && prefix.size() <= str.size() - start_pos &&
+           str.compare(start_pos, prefix.size(), prefix) == 0;
 }
 bool StringUtils::EndsWith(const std::string& str, const std::string& suffix) {
     size_t s1 = str.size();
@@ -74,26 +98,45 @@ bool StringUtils::IsNullOrWhitespaceOnly(const std::string& str) {
 }
 
 void StringUtils::Trim(std::string* str) {
-    str->erase(str->find_last_not_of(' ') + 1);
-    str->erase(0, str->find_first_not_of(' '));
+    auto first = std::find_if_not(str->begin(), str->end(),
+                                  [](unsigned char c) { return IsTrimCharacter(c); });
+    auto last = std::find_if_not(str->rbegin(), str->rend(), [](unsigned char c) {
+                    return IsTrimCharacter(c);
+                }).base();
+    if (first >= last) {
+        str->clear();
+        return;
+    }
+    *str = std::string(first, last);
 }
 
 std::string StringUtils::ToLowerCase(const std::string& str) {
     std::string result;
     result.reserve(str.length());
-    std::transform(str.begin(), str.end(), std::back_inserter(result),
-                   [](unsigned char c) { return std::tolower(c); });
+    std::transform(str.begin(), str.end(), std::back_inserter(result), ToAsciiLower);
     return result;
 }
 
 std::string StringUtils::ToUpperCase(const std::string& str) {
     std::string result;
     result.reserve(str.length());
-    std::transform(str.begin(), str.end(), std::back_inserter(result),
-                   [](unsigned char c) { return std::toupper(c); });
+    std::transform(str.begin(), str.end(), std::back_inserter(result), ToAsciiUpper);
     return result;
 }
 
+bool StringUtils::EqualsIgnoreCase(const std::string& left, const std::string& right) {
+    if (left.size() != right.size()) {
+        return false;
+    }
+    for (size_t i = 0; i < left.size(); ++i) {
+        if (ToAsciiLower(static_cast<unsigned char>(left[i])) !=
+            ToAsciiLower(static_cast<unsigned char>(right[i]))) {
+            return false;
+        }
+    }
+    return true;
+}
+
 std::vector<std::string> StringUtils::Split(const std::string& text, const std::string& sep_str,
                                             bool ignore_empty) {
     std::vector<std::string> vec;

src/paimon/common/utils/string_utils.h

@@ -50,24 +50,18 @@ class PAIMON_EXPORT StringUtils {
  public:
     /// Replaces all occurrences of a string within another string.
     ///
-    /// A `null` reference passed to this method is a no-op.
-    ///
     /// <pre>
-    /// StringUtils::Replace(null, *, *)        = null
     /// StringUtils::Replace("", *, *)          = ""
-    /// StringUtils::Replace("any", null, *)    = "any"
-    /// StringUtils::Replace("any", *, null)    = "any"
     /// StringUtils::Replace("any", "", *)      = "any"
-    /// StringUtils::Replace("aba", "a", null)  = "aba"
     /// StringUtils::Replace("aba", "a", "")    = "b"
     /// StringUtils::Replace("aba", "a", "z")   = "zbz"
     /// </pre>
     ///
     /// @see #replace(string text, string search_string, string replacement, int max)
-    /// @param text text to search and replace in, may be null
-    /// @param search_string the String to search for, may be null
-    /// @param replacement the String to replace it with, may be null
-    /// @return the text with any replacements processed, `null` if null string input
+    /// @param text text to search and replace in
+    /// @param search_string the String to search for
+    /// @param replacement the String to replace it with
+    /// @return the text with any replacements processed
     static std::string Replace(const std::string& text, const std::string& search_string,
                                const std::string& replacement) {
         return Replace(text, search_string, replacement, -1);
@@ -76,28 +70,22 @@ class PAIMON_EXPORT StringUtils {
     /// Replaces a String with another String inside a larger String, for the first `max` values of
     /// the search String.
     ///
-    /// A `null` reference passed to this method is a no-op.
-    ///
     /// <pre>
-    /// StringUtils::Replace(null, *, *, *)         = null
     /// StringUtils::Replace("", *, *, *)           = ""
-    /// StringUtils::Replace("any", null, *, *)     = "any"
-    /// StringUtils::Replace("any", *, null, *)     = "any"
     /// StringUtils::Replace("any", "", *, *)       = "any"
     /// StringUtils::Replace("any", *, *, 0)        = "any"
-    /// StringUtils::Replace("abaa", "a", null, -1) = "abaa"
     /// StringUtils::Replace("abaa", "a", "", -1)   = "b"
     /// StringUtils::Replace("abaa", "a", "z", 0)   = "abaa"
     /// StringUtils::Replace("abaa", "a", "z", 1)   = "zbaa"
     /// StringUtils::Replace("abaa", "a", "z", 2)   = "zbza"
     /// StringUtils::Replace("abaa", "a", "z", -1)  = "zbzz"
     /// </pre>
     ///
-    /// @param text text to search and replace in, may be null
-    /// @param search_string the String to search for, may be null
-    /// @param replacement the String to replace it with, may be null
+    /// @param text text to search and replace in
+    /// @param search_string the String to search for
+    /// @param replacement the String to replace it with
     /// @param max maximum number of values to replace, or `-1` if no maximum
-    /// @return the text with any replacements processed, `null` if null string input
+    /// @return the text with any replacements processed
     static std::string Replace(const std::string& text, const std::string& search_string,
                                const std::string& replacement, int32_t max);
 
@@ -115,6 +103,9 @@ class PAIMON_EXPORT StringUtils {
     static std::string ToLowerCase(const std::string& str);
     static std::string ToUpperCase(const std::string& str);
 
+    /// Compares two strings using ASCII case folding.
+    static bool EqualsIgnoreCase(const std::string& left, const std::string& right);
+
     template <typename T>
     static std::string VectorToString(const std::vector<T>& vec) {
         std::vector<std::string> strs;

src/paimon/common/utils/string_utils_test.cpp

@@ -73,6 +73,8 @@ void StringUtilsTest::CheckOverFlowAndUnderFlow(const std::string& over_flow,
 }
 
 TEST_F(StringUtilsTest, TestReplaceAll) {
+    ASSERT_EQ("abc", StringUtils::Replace("abc", "", "x"));
+    ASSERT_EQ("", StringUtils::Replace("", "a", "b"));
     {
         std::string origin = "how is is you";
         std::string expect = "how are are you";
@@ -118,6 +120,8 @@ TEST_F(StringUtilsTest, TestReplaceAll) {
 }
 
 TEST_F(StringUtilsTest, TestReplaceLast) {
+    ASSERT_EQ("abc", StringUtils::ReplaceLast("abc", "", "x"));
+    ASSERT_EQ("", StringUtils::ReplaceLast("", "a", "b"));
     {
         std::string origin = "a/b/c//";
         std::string expect = "a/b/c/_";
@@ -140,6 +144,7 @@ TEST_F(StringUtilsTest, TestReplaceLast) {
 }
 
 TEST_F(StringUtilsTest, TestReplaceWithMaxCount) {
+    ASSERT_EQ("abc", StringUtils::Replace("abc", "a", "b", 0));
     {
         std::string origin = "how is is you";
         std::string expect = "how are is you";
@@ -236,6 +241,13 @@ TEST_F(StringUtilsTest, TestToUpperCase) {
     }
 }
 
+TEST_F(StringUtilsTest, TestEqualsIgnoreCase) {
+    ASSERT_TRUE(StringUtils::EqualsIgnoreCase("", ""));
+    ASSERT_TRUE(StringUtils::EqualsIgnoreCase("AbC-123", "aBc-123"));
+    ASSERT_FALSE(StringUtils::EqualsIgnoreCase("abc", "abcd"));
+    ASSERT_FALSE(StringUtils::EqualsIgnoreCase("abc", "abx"));
+}
+
 TEST_F(StringUtilsTest, TestStartsWith) {
     {
         std::string str = "abcde";
@@ -261,6 +273,26 @@ TEST_F(StringUtilsTest, TestStartsWith) {
         std::string str = "";
         ASSERT_TRUE(StringUtils::StartsWith(str, ""));
     }
+    {
+        std::string str = "abc";
+        ASSERT_TRUE(StringUtils::StartsWith(str, "", /*start_pos=*/3));
+        ASSERT_FALSE(StringUtils::StartsWith(str, "", /*start_pos=*/4));
+        ASSERT_FALSE(StringUtils::StartsWith(str, "a", /*start_pos=*/4));
+    }
+}
+
+TEST_F(StringUtilsTest, TestTrim) {
+    std::string value = " \tabc\r\n";
+    StringUtils::Trim(&value);
+    ASSERT_EQ("abc", value);
+
+    value = "\t\r\n";
+    StringUtils::Trim(&value);
+    ASSERT_TRUE(value.empty());
+
+    value.clear();
+    StringUtils::Trim(&value);
+    ASSERT_TRUE(value.empty());
 }
 TEST_F(StringUtilsTest, TestEndsWith) {
     {