fix: align format options and listagg semantics with Java

examples/read_write_demo.cpp

@@ -56,8 +56,7 @@ arrow::Result<std::shared_ptr<arrow::StructArray>> PrepareData(const arrow::Fiel
 
 paimon::Status Run(const std::string& root_path, const std::string& db_name,
                    const std::string& table_name) {
-    std::map<std::string, std::string> options = {{paimon::Options::MANIFEST_FORMAT, "orc"},
-                                                  {paimon::Options::FILE_FORMAT, "parquet"},
+    std::map<std::string, std::string> options = {{paimon::Options::FILE_FORMAT, "parquet"},
                                                   {paimon::Options::FILE_SYSTEM, "local"}};
 
     // create table

include/paimon/defs.h

@@ -164,10 +164,6 @@ struct PAIMON_EXPORT Options {
     /// Default value is 8MB.
     static const char MANIFEST_TARGET_FILE_SIZE[];
 
-    /// "manifest.format" - Specify the message format of manifest files.
-    /// Default value is avro.
-    static const char MANIFEST_FORMAT[];
-
     /// "manifest.compression" - File compression for manifest, default value is zstd.
     static const char MANIFEST_COMPRESSION[];
 

src/paimon/common/data/shredding/map_shared_shredding_read_plan_factory_test.cpp

@@ -232,7 +232,6 @@ class MapSharedShreddingReadPlanFactoryTest : public ::testing::Test {
     std::map<std::string, std::string> options_ = {
         {Options::FILE_SYSTEM, "local"},
         {Options::FILE_FORMAT, "mock_format"},
-        {Options::MANIFEST_FORMAT, "mock_format"},
         {"fields.tags.map.storage-layout", "shared-shredding"},
         {"fields.tags.map.shared-shredding.max-columns", "2"},
         {"fields.tags.map.shared-shredding.column-placement-policy", "plain"},

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

@@ -47,8 +47,6 @@ class VariantShreddingWritePlanFactoryTest : public ::testing::Test {
     }
 
     Result<CoreOptions> MakeOptions(std::map<std::string, std::string> options) const {
-        // Keep the manifest format resolvable in test binaries without the avro plugin.
-        options.emplace("manifest.format", "parquet");
         return CoreOptions::FromMap(options);
     }
 

src/paimon/common/defs.cpp

@@ -48,7 +48,6 @@ const char Options::FILE_COMPRESSION[] = "file.compression";
 const char Options::FILE_COMPRESSION_ZSTD_LEVEL[] = "file.compression.zstd-level";
 const char Options::FILE_BLOCK_SIZE[] = "file.block-size";
 const char Options::MANIFEST_TARGET_FILE_SIZE[] = "manifest.target-file-size";
-const char Options::MANIFEST_FORMAT[] = "manifest.format";
 const char Options::MANIFEST_COMPRESSION[] = "manifest.compression";
 const char Options::MANIFEST_MERGE_MIN_COUNT[] = "manifest.merge-min-count";
 const char Options::MANIFEST_FULL_COMPACTION_FILE_SIZE[] =

src/paimon/common/utils/binary_row_partition_computer_test.cpp

@@ -251,20 +251,27 @@ TEST(BinaryRowPartitionComputerTest, TestNullOrWhitespaceOnlyStr) {
         arrow::field("f0", arrow::utf8()),
         arrow::field("f1", arrow::utf8()),
         arrow::field("f2", arrow::utf8()),
+        arrow::field("f3", arrow::utf8()),
     };
 
     auto schema = arrow::schema(fields);
-    std::vector<std::string> partition_keys = {"f0", "f1", "f2"};
+    std::vector<std::string> partition_keys = {"f0", "f1", "f2", "f3"};
     ASSERT_OK_AND_ASSIGN(
         std::unique_ptr<BinaryRowPartitionComputer> computer,
         BinaryRowPartitionComputer::Create(partition_keys, schema, "__DEFAULT_PARTITION__",
                                            /*legacy_partition_name_enabled=*/true, pool));
 
-    ASSERT_OK_AND_ASSIGN(auto partition_key_values,
-                         computer->GeneratePartitionVector(BinaryRowGenerator::GenerateRow(
-                             {std::string(" "), std::string(""), std::string("ab ")}, pool.get())));
+    ASSERT_OK_AND_ASSIGN(
+        auto partition_key_values,
+        computer->GeneratePartitionVector(BinaryRowGenerator::GenerateRow(
+            {std::string(" "), std::string(""), std::string("ab "), std::string(u8"\u3000\u2000")},
+            pool.get())));
     std::vector<std::pair<std::string, std::string>> expected = {
-        {"f0", "__DEFAULT_PARTITION__"}, {"f1", "__DEFAULT_PARTITION__"}, {"f2", "ab "}};
+        {"f0", "__DEFAULT_PARTITION__"},
+        {"f1", "__DEFAULT_PARTITION__"},
+        {"f2", "ab "},
+        {"f3", "__DEFAULT_PARTITION__"},
+    };
     ASSERT_EQ(partition_key_values, expected);
 }
 

src/paimon/common/utils/string_utils.cpp

@@ -37,6 +37,14 @@ bool IsTrimCharacter(unsigned char c) {
     return c <= 0x20;
 }
 
+bool IsJavaWhitespace(uint32_t code_point) {
+    return (code_point >= 0x0009 && code_point <= 0x000d) ||
+           (code_point >= 0x001c && code_point <= 0x0020) || code_point == 0x1680 ||
+           (code_point >= 0x2000 && code_point <= 0x2006) ||
+           (code_point >= 0x2008 && code_point <= 0x200a) || code_point == 0x2028 ||
+           code_point == 0x2029 || code_point == 0x205f || code_point == 0x3000;
+}
+
 char ToAsciiLower(unsigned char c) {
     return c >= 'A' && c <= 'Z' ? static_cast<char>(c + ('a' - 'A')) : static_cast<char>(c);
 }
@@ -85,18 +93,58 @@ bool StringUtils::EndsWith(const std::string& str, const std::string& suffix) {
     size_t s2 = suffix.size();
     return (s1 >= s2) && (str.compare(s1 - s2, s2, suffix) == 0);
 }
-bool StringUtils::IsNullOrWhitespaceOnly(const std::string& str) {
-    if (str.empty()) {
-        return true;
-    }
-    for (char c : str) {
-        if (!std::isspace(static_cast<unsigned char>(c))) {
+
+bool StringUtils::IsBlank(std::string_view str) {
+    size_t offset = 0;
+    while (offset < str.size()) {
+        const auto first = static_cast<uint8_t>(str[offset]);
+        uint32_t code_point = 0;
+        size_t length = 0;
+        if (first <= 0x7f) {
+            code_point = first;
+            length = 1;
+        } else if (first >= 0xc2 && first <= 0xdf) {
+            code_point = first & 0x1f;
+            length = 2;
+        } else if (first >= 0xe0 && first <= 0xef) {
+            code_point = first & 0x0f;
+            length = 3;
+        } else if (first >= 0xf0 && first <= 0xf4) {
+            code_point = first & 0x07;
+            length = 4;
+        } else {
+            return false;
+        }
+        if (offset + length > str.size()) {
+            return false;
+        }
+        for (size_t i = 1; i < length; ++i) {
+            const auto continuation = static_cast<uint8_t>(str[offset + i]);
+            if ((continuation & 0xc0) != 0x80) {
+                return false;
+            }
+            code_point = (code_point << 6) | (continuation & 0x3f);
+        }
+        if ((length == 3 && code_point < 0x800) || (length == 4 && code_point < 0x10000) ||
+            (code_point >= 0xd800 && code_point <= 0xdfff) || code_point > 0x10ffff) {
             return false;
         }
+        if (!IsJavaWhitespace(code_point)) {
+            return false;
+        }
+        offset += length;
     }
     return true;
 }
 
+bool StringUtils::IsNullOrWhitespaceOnly(const std::string& str) {
+    return IsBlank(str);
+}
+
+bool StringUtils::IsEmptyAfterTrim(std::string_view str) {
+    return std::all_of(str.begin(), str.end(), [](unsigned char c) { return IsTrimCharacter(c); });
+}
+
 void StringUtils::Trim(std::string* str) {
     auto first = std::find_if_not(str->begin(), str->end(),
                                   [](unsigned char c) { return IsTrimCharacter(c); });

src/paimon/common/utils/string_utils.h

@@ -31,6 +31,7 @@
 #include <set>
 #include <sstream>
 #include <string>
+#include <string_view>
 #include <system_error>
 #include <vector>
 
@@ -96,8 +97,15 @@ class PAIMON_EXPORT StringUtils {
 
     static bool EndsWith(const std::string& str, const std::string& suffix);
 
+    /// Returns true if the string is empty or contains only characters recognized by Java
+    /// Character.isWhitespace.
+    static bool IsBlank(std::string_view str);
+
     static bool IsNullOrWhitespaceOnly(const std::string& str);
 
+    /// Returns true if Java String::trim would produce an empty string.
+    static bool IsEmptyAfterTrim(std::string_view str);
+
     static void Trim(std::string* str);
 
     static std::string ToLowerCase(const std::string& str);

src/paimon/common/utils/string_utils_test.cpp

@@ -21,6 +21,7 @@
 
 #include <limits>
 #include <memory>
+#include <vector>
 
 #include "gtest/gtest.h"
 #include "paimon/status.h"
@@ -209,6 +210,34 @@ TEST_F(StringUtilsTest, TestIsNullOrWhitespaceOnly) {
         auto ret = StringUtils::IsNullOrWhitespaceOnly(str);
         ASSERT_TRUE(ret);
     }
+    ASSERT_TRUE(StringUtils::IsNullOrWhitespaceOnly(u8"\u3000\u2000"));
+}
+
+TEST_F(StringUtilsTest, TestIsBlank) {
+    const std::vector<std::string> blank_strings = {
+        "",     " ",        "   ",      "\t",       "\n",       "\r",
+        "\r\n", " \t\n\r ", u8"\u1680", u8"\u2000", u8"\u3000", u8" \t\u3000\u2000\n"};
+    for (const std::string& blank : blank_strings) {
+        ASSERT_TRUE(StringUtils::IsBlank(blank)) << blank;
+    }
+
+    ASSERT_FALSE(StringUtils::IsBlank("user1"));
+    ASSERT_FALSE(StringUtils::IsBlank(" user1 "));
+    ASSERT_FALSE(StringUtils::IsBlank(u8"\u00a0"));
+    ASSERT_FALSE(StringUtils::IsBlank(std::string("\xc0\x80", 2)));
+
+    // Non-breaking or otherwise excluded by Character.isWhitespace.
+    ASSERT_FALSE(StringUtils::IsBlank(u8"\u2007"));  // FIGURE SPACE
+    ASSERT_FALSE(StringUtils::IsBlank(u8"\u202f"));  // NARROW NO-BREAK SPACE
+    ASSERT_FALSE(StringUtils::IsBlank(u8"\u0085"));  // NEL
+    ASSERT_FALSE(StringUtils::IsBlank(u8"\u180e"));  // Not whitespace since Java 8
+}
+
+TEST_F(StringUtilsTest, TestIsEmptyAfterTrim) {
+    ASSERT_TRUE(StringUtils::IsEmptyAfterTrim(""));
+    ASSERT_TRUE(StringUtils::IsEmptyAfterTrim(" \t\x1c"));
+    ASSERT_FALSE(StringUtils::IsEmptyAfterTrim(" a "));
+    ASSERT_FALSE(StringUtils::IsEmptyAfterTrim(u8"\u3000"));
 }
 
 TEST_F(StringUtilsTest, TestToLowerCase) {