fix: align format options and listagg semantics with Java

include/paimon/defs.h

@@ -164,7 +164,8 @@ struct PAIMON_EXPORT Options {
     /// Default value is 8MB.
     static const char MANIFEST_TARGET_FILE_SIZE[];
 
-    /// "manifest.format" - Specify the message format of manifest files.
+    /// "manifest.format" - Deprecated compatibility option for reading legacy manifests.
+    /// Avro is the only writable manifest format; non-Avro formats are read-only.
     /// Default value is avro.
     static const char MANIFEST_FORMAT[];
 

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,54 @@ 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);
+}
+
 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,6 +97,10 @@ 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);
 
     static void Trim(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,21 @@ 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)));
 }
 
 TEST_F(StringUtilsTest, TestToLowerCase) {

src/paimon/core/append/append_compact_coordinator.cpp

@@ -133,17 +133,17 @@ Result<std::unique_ptr<FileStoreScan>> CreateFileStoreScan(
     const std::shared_ptr<FileStorePathFactory>& path_factory,
     const std::shared_ptr<ScanFilter>& scan_filter, const std::shared_ptr<Executor>& executor,
     const std::shared_ptr<MemoryPool>& pool) {
-    PAIMON_ASSIGN_OR_RAISE(
-        std::shared_ptr<ManifestList> manifest_list,
-        ManifestList::Create(core_options.GetFileSystem(), core_options.GetManifestFormat(),
-                             core_options.GetManifestCompression(), path_factory,
-                             core_options.GetCache(), pool));
-    PAIMON_ASSIGN_OR_RAISE(
-        std::shared_ptr<ManifestFile> manifest_file,
-        ManifestFile::Create(core_options.GetFileSystem(), core_options.GetManifestFormat(),
-                             core_options.GetManifestCompression(), path_factory,
-                             core_options.GetManifestTargetFileSize(), pool, core_options,
-                             partition_schema));
+    PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<FileFormat> manifest_format,
+                           core_options.GetManifestFormat(/*write=*/false));
+    PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<ManifestList> manifest_list,
+                           ManifestList::Create(core_options.GetFileSystem(), manifest_format,
+                                                core_options.GetManifestCompression(), path_factory,
+                                                core_options.GetCache(), pool));
+    PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<ManifestFile> manifest_file,
+                           ManifestFile::Create(core_options.GetFileSystem(), manifest_format,
+                                                core_options.GetManifestCompression(), path_factory,
+                                                core_options.GetManifestTargetFileSize(), pool,
+                                                core_options, partition_schema));
     PAIMON_ASSIGN_OR_RAISE(
         std::unique_ptr<AppendOnlyFileStoreScan> scan,
         AppendOnlyFileStoreScan::Create(snapshot_manager, schema_manager, manifest_list,

src/paimon/core/core_options.cpp

@@ -610,7 +610,7 @@ struct CoreOptions::Impl {
 
     // Parse manifest file configurations: format, compression, merge, and compaction thresholds.
     Status ParseManifestOptions(const ConfigParser& parser) {
-        // Parse manifest.format - manifest file format, default "avro"
+        // Parse manifest.format for reading legacy manifests; only avro is writable.
         PAIMON_RETURN_NOT_OK(parser.ParseObject<FileFormatFactory>(
             Options::MANIFEST_FORMAT, /*default_identifier=*/"avro", &manifest_file_format));
         // Parse manifest.compression - manifest file compression, default "zstd"
@@ -1143,7 +1143,13 @@ std::string CoreOptions::GetPartitionDefaultName() const {
     return impl_->partition_default_name;
 }
 
-std::shared_ptr<FileFormat> CoreOptions::GetManifestFormat() const {
+Result<std::shared_ptr<FileFormat>> CoreOptions::GetManifestFormat(bool write) const {
+    const std::string& identifier = impl_->manifest_file_format->Identifier();
+    if (write && identifier != "avro") {
+        return Status::Invalid(fmt::format(
+            "manifest.format '{}' is read-only; only 'avro' can be used for writing manifests",
+            identifier));
+    }
     return impl_->manifest_file_format;
 }
 

src/paimon/core/core_options.h

@@ -93,7 +93,9 @@ class PAIMON_EXPORT CoreOptions {
     int64_t GetCompactionFileSize(bool has_primary_key) const;
     std::string GetPartitionDefaultName() const;
 
-    std::shared_ptr<FileFormat> GetManifestFormat() const;
+    /// Return the configured manifest format for the requested access mode.
+    /// Non-Avro formats are supported only for reading legacy manifests.
+    Result<std::shared_ptr<FileFormat>> GetManifestFormat(bool write) const;
     const std::string& GetManifestCompression() const;
     int32_t GetManifestMergeMinCount() const;
     int64_t GetManifestFullCompactionThresholdSize() const;

src/paimon/core/core_options_test.cpp

@@ -37,7 +37,9 @@ namespace paimon::test {
 
 TEST(CoreOptionsTest, TestDefaultValue) {
     ASSERT_OK_AND_ASSIGN(CoreOptions core_options, CoreOptions::FromMap({}));
-    ASSERT_EQ(core_options.GetManifestFormat()->Identifier(), "avro");
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<FileFormat> manifest_format,
+                         core_options.GetManifestFormat(/*write=*/false));
+    ASSERT_EQ(manifest_format->Identifier(), "avro");
     ASSERT_EQ(core_options.GetFileFormat()->Identifier(), "parquet");
     ASSERT_EQ(nullptr, core_options.GetChangelogFileFormat());
     ASSERT_EQ(core_options.GetWriteFileFormat(0)->Identifier(), "parquet");
@@ -193,6 +195,23 @@ TEST(CoreOptionsTest, TestDefaultValue) {
     ASSERT_EQ(BucketFunctionType::DEFAULT, core_options.GetBucketFunctionType());
 }
 
+TEST(CoreOptionsTest, GetManifestFormatForReadAndWrite) {
+    ASSERT_OK_AND_ASSIGN(CoreOptions default_options, CoreOptions::FromMap({}));
+    ASSERT_OK(default_options.GetManifestFormat(/*write=*/true));
+
+    ASSERT_OK_AND_ASSIGN(CoreOptions avro_options,
+                         CoreOptions::FromMap({{Options::MANIFEST_FORMAT, "AvRo"}}));
+    ASSERT_OK(avro_options.GetManifestFormat(/*write=*/true));
+
+    ASSERT_OK_AND_ASSIGN(CoreOptions legacy_options,
+                         CoreOptions::FromMap({{Options::MANIFEST_FORMAT, "orc"}}));
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<FileFormat> legacy_manifest_format,
+                         legacy_options.GetManifestFormat(/*write=*/false));
+    ASSERT_EQ(legacy_manifest_format->Identifier(), "orc");
+    ASSERT_NOK_WITH_MSG(legacy_options.GetManifestFormat(/*write=*/true),
+                        "manifest.format 'orc' is read-only");
+}
+
 TEST(CoreOptionsTest, TestFromMap) {
     std::map<std::string, std::string> options = {
         {Options::FILE_SYSTEM, "Local"},
@@ -339,7 +358,8 @@ TEST(CoreOptionsTest, TestFromMap) {
     ASSERT_EQ(core_options.GetWriteFileFormat(1)->Identifier(), "orc");
     ASSERT_EQ(core_options.GetWriteFileFormat(3)->Identifier(), "parquet");
 
-    auto manifest_format = core_options.GetManifestFormat();
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<FileFormat> manifest_format,
+                         core_options.GetManifestFormat(/*write=*/false));
     ASSERT_EQ(manifest_format->Identifier(), "avro");
 
     ASSERT_EQ(3, core_options.GetBucket());

src/paimon/core/global_index/global_index_scan_impl.cpp

@@ -82,9 +82,11 @@ Result<std::unique_ptr<GlobalIndexScanImpl>> GlobalIndexScanImpl::Create(
     std::shared_ptr<IndexPathFactory> path_factory =
         file_store_path_factory->CreateGlobalIndexFileFactory();
 
+    PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<FileFormat> manifest_format,
+                           options.GetManifestFormat(/*write=*/false));
     PAIMON_ASSIGN_OR_RAISE(
         std::unique_ptr<IndexManifestFile> index_manifest_file,
-        IndexManifestFile::Create(options.GetFileSystem(), options.GetManifestFormat(),
+        IndexManifestFile::Create(options.GetFileSystem(), manifest_format,
                                   options.GetManifestCompression(), file_store_path_factory,
                                   options.GetBucket(), pool, options));
     auto index_file_handler = std::make_unique<IndexFileHandler>(

src/paimon/core/index/index_file_handler_test.cpp

@@ -70,11 +70,13 @@ class IndexFileHandlerTest : public testing::Test {
                 global_index_external_path,
                 /*index_file_in_data_file_dir=*/core_options.IndexFileInDataFileDir(),
                 memory_pool_));
-        PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<IndexManifestFile> index_manifest_file,
-                               IndexManifestFile::Create(
-                                   core_options.GetFileSystem(), core_options.GetManifestFormat(),
-                                   core_options.GetManifestCompression(), path_factory,
-                                   core_options.GetBucket(), memory_pool_, core_options));
+        PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<FileFormat> manifest_format,
+                               core_options.GetManifestFormat(/*write=*/false));
+        PAIMON_ASSIGN_OR_RAISE(
+            std::unique_ptr<IndexManifestFile> index_manifest_file,
+            IndexManifestFile::Create(core_options.GetFileSystem(), manifest_format,
+                                      core_options.GetManifestCompression(), path_factory,
+                                      core_options.GetBucket(), memory_pool_, core_options));
         auto path_factories = std::make_shared<IndexFilePathFactories>(path_factory);
         return std::make_unique<IndexFileHandler>(
             core_options.GetFileSystem(), std::move(index_manifest_file), path_factories,

src/paimon/core/mergetree/compact/aggregate/field_listagg_agg.h

@@ -20,10 +20,12 @@
 
 #include <memory>
 #include <string>
+#include <string_view>
 #include <unordered_set>
 #include <utility>
 
 #include "paimon/common/data/data_define.h"
+#include "paimon/common/utils/string_utils.h"
 #include "paimon/core/core_options.h"
 #include "paimon/core/mergetree/compact/aggregate/field_aggregator.h"
 
@@ -60,15 +62,18 @@ class FieldListaggAgg : public FieldAggregator {
                             const VariantType& input_field) override {
         bool accumulator_null = DataDefine::IsVariantNull(accumulator);
         bool input_null = DataDefine::IsVariantNull(input_field);
-        if (accumulator_null || input_null) {
-            return accumulator_null ? input_field : accumulator;
+        if (input_null) {
+            return accumulator;
         }
-        std::string_view acc_str = DataDefine::GetStringView(accumulator);
         std::string_view in_str = DataDefine::GetStringView(input_field);
-        if (in_str.empty()) {
+        if (StringUtils::IsBlank(in_str)) {
             return accumulator;
         }
-        if (acc_str.empty()) {
+        if (accumulator_null) {
+            return input_field;
+        }
+        std::string_view acc_str = DataDefine::GetStringView(accumulator);
+        if (StringUtils::IsBlank(acc_str)) {
             return input_field;
         }
 
@@ -95,9 +100,7 @@ class FieldListaggAgg : public FieldAggregator {
             size_t pos = remaining.find(delimiter_);
             std::string_view token =
                 (pos == std::string_view::npos) ? remaining : remaining.substr(0, pos);
-            if (!token.empty()) {
-                seen.insert(token);
-            }
+            seen.insert(token);
             if (pos == std::string_view::npos) {
                 break;
             }
@@ -113,7 +116,7 @@ class FieldListaggAgg : public FieldAggregator {
             size_t pos = remaining.find(delimiter_);
             std::string_view token =
                 (pos == std::string_view::npos) ? remaining : remaining.substr(0, pos);
-            if (!token.empty() && seen.insert(token).second) {
+            if (!StringUtils::IsBlank(token) && seen.insert(token).second) {
                 result.append(delimiter_);
                 result.append(token);
             }

src/paimon/core/mergetree/compact/aggregate/field_listagg_agg_test.cpp

@@ -21,6 +21,7 @@
 #include <map>
 #include <string>
 #include <string_view>
+#include <vector>
 
 #include "arrow/type_fwd.h"
 #include "gtest/gtest.h"
@@ -88,13 +89,41 @@ TEST_F(FieldListaggAggTest, TestEmptyString) {
         auto ret = agg->Agg(std::string_view(""), std::string_view("world")).value();
         ASSERT_EQ(DataDefine::GetVariantValue<std::string_view>(ret), "world");
     }
-    // both empty -> return input (which is empty)
+    // blank input -> return accumulator (which is empty)
     {
         auto ret = agg->Agg(std::string_view(""), std::string_view("")).value();
         ASSERT_EQ(DataDefine::GetVariantValue<std::string_view>(ret), "");
     }
 }
 
+TEST_F(FieldListaggAggTest, TestBlankStrings) {
+    ASSERT_OK_AND_ASSIGN(auto agg, MakeAgg());
+
+    const std::vector<std::string> blank_strings = {"",
+                                                    " ",
+                                                    "   ",
+                                                    "\t",
+                                                    "\n",
+                                                    "\r",
+                                                    "\r\n",
+                                                    " \t\n\r ",
+                                                    u8"\u3000",
+                                                    u8"\u2000",
+                                                    u8" \t\u3000\u2000\n"};
+    for (const std::string& blank : blank_strings) {
+        auto ret = agg->Agg(std::string_view("user1"), std::string_view(blank)).value();
+        ASSERT_EQ(DataDefine::GetVariantValue<std::string_view>(ret), "user1");
+    }
+
+    // A blank accumulator must not add a leading delimiter.
+    auto ret = agg->Agg(std::string_view(u8"\u3000\t"), std::string_view("user1")).value();
+    ASSERT_EQ(DataDefine::GetVariantValue<std::string_view>(ret), "user1");
+
+    // A blank input must not turn a null accumulator into a non-null value.
+    ret = agg->Agg(NullType(), std::string_view(u8" \t\u3000")).value();
+    ASSERT_TRUE(DataDefine::IsVariantNull(ret));
+}
+
 TEST_F(FieldListaggAggTest, TestMultipleAccumulation) {
     ASSERT_OK_AND_ASSIGN(auto agg, MakeAgg());
 
@@ -113,6 +142,15 @@ TEST_F(FieldListaggAggTest, TestDistinct) {
     ASSERT_EQ(DataDefine::GetVariantValue<std::string_view>(ret), "a;b;c");
 }
 
+TEST_F(FieldListaggAggTest, TestDistinctIgnoresBlankTokens) {
+    ASSERT_OK_AND_ASSIGN(auto agg, MakeAgg(",", true));
+
+    auto ret =
+        agg->Agg(std::string_view("user1"), std::string_view(u8" ,user2,\t,\u3000,user1,\u2000"))
+            .value();
+    ASSERT_EQ(DataDefine::GetVariantValue<std::string_view>(ret), "user1,user2");
+}
+
 TEST_F(FieldListaggAggTest, TestDistinctNoDuplicates) {
     ASSERT_OK_AND_ASSIGN(auto agg, MakeAgg(" ", true));