perf(write): write input changelog and data files in parallel

include/paimon/defs.h

@@ -421,6 +421,11 @@ struct PAIMON_EXPORT Options {
     /// "changelog-producer.row-deduplicate" is "true".
     static const char CHANGELOG_PRODUCER_ROW_DEDUPLICATE_IGNORE_FIELDS[];
 
+    /// "changelog-producer.input.parallel-write" - Whether data and changelog files are
+    /// written in parallel for input mode changelog producer. Default value is "true".This
+    /// configuration is specific to cpp paimon.
+    static const char CHANGELOG_PRODUCER_INPUT_PARALLEL_WRITE[];
+
     /// "changelog-file.prefix" - Specify the file name prefix of changelog files. Default value is
     /// "changelog-".
     static const char CHANGELOG_FILE_PREFIX[];

src/paimon/common/defs.cpp

@@ -104,6 +104,8 @@ const char Options::CHANGELOG_PRODUCER[] = "changelog-producer";
 const char Options::CHANGELOG_PRODUCER_ROW_DEDUPLICATE[] = "changelog-producer.row-deduplicate";
 const char Options::CHANGELOG_PRODUCER_ROW_DEDUPLICATE_IGNORE_FIELDS[] =
     "changelog-producer.row-deduplicate-ignore-fields";
+const char Options::CHANGELOG_PRODUCER_INPUT_PARALLEL_WRITE[] =
+    "changelog-producer.input.parallel-write";
 const char Options::CHANGELOG_FILE_PREFIX[] = "changelog-file.prefix";
 const char Options::CHANGELOG_FILE_FORMAT[] = "changelog-file.format";
 const char Options::CHANGELOG_FILE_COMPRESSION[] = "changelog-file.compression";

src/paimon/core/core_options.cpp

@@ -447,6 +447,7 @@ struct CoreOptions::Impl {
     bool force_lookup = false;
     bool lookup_wait = true;
     bool changelog_row_deduplicate = false;
+    bool input_changelog_parallel_write_enabled = true;
     bool partial_update_remove_record_on_delete = false;
     bool aggregation_remove_record_on_delete = false;
     bool table_read_sequence_number_enabled = false;
@@ -752,6 +753,10 @@ struct CoreOptions::Impl {
         PAIMON_RETURN_NOT_OK(parser.ParseList<std::string>(
             Options::CHANGELOG_PRODUCER_ROW_DEDUPLICATE_IGNORE_FIELDS, Options::FIELDS_SEPARATOR,
             &changelog_row_deduplicate_ignore_fields, /*need_trim=*/true));
+        // Parse changelog-producer.input.parallel-write - write data and input changelog
+        // files in parallel, default true.
+        PAIMON_RETURN_NOT_OK(parser.Parse<bool>(Options::CHANGELOG_PRODUCER_INPUT_PARALLEL_WRITE,
+                                                &input_changelog_parallel_write_enabled));
         // Parse partial-update.remove-record-on-delete - remove whole row on delete
         PAIMON_RETURN_NOT_OK(parser.Parse<bool>(Options::PARTIAL_UPDATE_REMOVE_RECORD_ON_DELETE,
                                                 &partial_update_remove_record_on_delete));
@@ -1630,6 +1635,10 @@ const std::vector<std::string>& CoreOptions::GetChangelogRowDeduplicateIgnoreFie
     return impl_->changelog_row_deduplicate_ignore_fields;
 }
 
+bool CoreOptions::InputChangelogParallelWriteEnabled() const {
+    return impl_->input_changelog_parallel_write_enabled;
+}
+
 std::string CoreOptions::ChangelogFilePrefix() const {
     return impl_->changelog_file_prefix;
 }

src/paimon/core/core_options.h

@@ -210,6 +210,7 @@ class PAIMON_EXPORT CoreOptions {
     ChangelogProducer GetChangelogProducer() const;
     bool ChangelogRowDeduplicate() const;
     const std::vector<std::string>& GetChangelogRowDeduplicateIgnoreFields() const;
+    bool InputChangelogParallelWriteEnabled() const;
     std::string ChangelogFilePrefix() const;
     std::shared_ptr<FileFormat> GetChangelogFileFormat() const;
     std::optional<std::string> GetChangelogFileCompression() const;

src/paimon/core/core_options_test.cpp

@@ -131,6 +131,7 @@ TEST(CoreOptionsTest, TestDefaultValue) {
     ASSERT_EQ(ChangelogProducer::NONE, core_options.GetChangelogProducer());
     ASSERT_FALSE(core_options.ChangelogRowDeduplicate());
     ASSERT_TRUE(core_options.GetChangelogRowDeduplicateIgnoreFields().empty());
+    ASSERT_TRUE(core_options.InputChangelogParallelWriteEnabled());
     ASSERT_EQ("changelog-", core_options.ChangelogFilePrefix());
     ASSERT_FALSE(core_options.NeedLookup());
     ASSERT_FALSE(core_options.PrepareCommitWaitCompaction());
@@ -258,6 +259,7 @@ TEST(CoreOptionsTest, TestFromMap) {
         {Options::CHANGELOG_PRODUCER, "full-compaction"},
         {Options::CHANGELOG_PRODUCER_ROW_DEDUPLICATE, "true"},
         {Options::CHANGELOG_PRODUCER_ROW_DEDUPLICATE_IGNORE_FIELDS, "f0, f2"},
+        {Options::CHANGELOG_PRODUCER_INPUT_PARALLEL_WRITE, "false"},
         {Options::CHANGELOG_FILE_PREFIX, "test-changelog-"},
         {Options::CHANGELOG_FILE_COMPRESSION, "lz4"},
         {Options::FORCE_LOOKUP, "true"},
@@ -410,6 +412,7 @@ TEST(CoreOptionsTest, TestFromMap) {
     ASSERT_TRUE(core_options.ChangelogRowDeduplicate());
     ASSERT_EQ(std::vector<std::string>({"f0", "f2"}),
               core_options.GetChangelogRowDeduplicateIgnoreFields());
+    ASSERT_FALSE(core_options.InputChangelogParallelWriteEnabled());
     ASSERT_EQ("test-changelog-", core_options.ChangelogFilePrefix());
     ASSERT_EQ(std::optional<std::string>("lz4"), core_options.GetChangelogFileCompression());
     ASSERT_TRUE(core_options.NeedLookup());