From f304e7e3eb65f1ca65a07935d11811038ac386eb Mon Sep 17 00:00:00 2001 From: "lisizhuo.lsz" Date: Tue, 15 Sep 2026 18:32:22 +0800 Subject: [PATCH 1/4] perf(write): write input changelog and data files in parallel --- include/paimon/defs.h | 4 + src/paimon/common/defs.cpp | 2 + src/paimon/core/core_options.cpp | 10 + src/paimon/core/core_options.h | 1 + src/paimon/core/core_options_test.cpp | 3 + .../core/mergetree/merge_tree_writer.cpp | 229 +++++++++++------- src/paimon/core/mergetree/merge_tree_writer.h | 3 + .../core/mergetree/merge_tree_writer_test.cpp | 51 +++- 8 files changed, 218 insertions(+), 85 deletions(-) diff --git a/include/paimon/defs.h b/include/paimon/defs.h index cfa08fa00..33d83cd6d 100644 --- a/include/paimon/defs.h +++ b/include/paimon/defs.h @@ -421,6 +421,10 @@ 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". + 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[]; diff --git a/src/paimon/common/defs.cpp b/src/paimon/common/defs.cpp index de1d45773..88085f78d 100644 --- a/src/paimon/common/defs.cpp +++ b/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"; diff --git a/src/paimon/core/core_options.cpp b/src/paimon/core/core_options.cpp index 37fadc8e1..17472d072 100644 --- a/src/paimon/core/core_options.cpp +++ b/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,11 @@ struct CoreOptions::Impl { PAIMON_RETURN_NOT_OK(parser.ParseList( 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(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(Options::PARTIAL_UPDATE_REMOVE_RECORD_ON_DELETE, &partial_update_remove_record_on_delete)); @@ -1630,6 +1636,10 @@ const std::vector& 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; } diff --git a/src/paimon/core/core_options.h b/src/paimon/core/core_options.h index 86771e6aa..6dadc680e 100644 --- a/src/paimon/core/core_options.h +++ b/src/paimon/core/core_options.h @@ -210,6 +210,7 @@ class PAIMON_EXPORT CoreOptions { ChangelogProducer GetChangelogProducer() const; bool ChangelogRowDeduplicate() const; const std::vector& GetChangelogRowDeduplicateIgnoreFields() const; + bool InputChangelogParallelWriteEnabled() const; std::string ChangelogFilePrefix() const; std::shared_ptr GetChangelogFileFormat() const; std::optional GetChangelogFileCompression() const; diff --git a/src/paimon/core/core_options_test.cpp b/src/paimon/core/core_options_test.cpp index 3e489a4c9..be9fa8234 100644 --- a/src/paimon/core/core_options_test.cpp +++ b/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({"f0", "f2"}), core_options.GetChangelogRowDeduplicateIgnoreFields()); + ASSERT_FALSE(core_options.InputChangelogParallelWriteEnabled()); ASSERT_EQ("test-changelog-", core_options.ChangelogFilePrefix()); ASSERT_EQ(std::optional("lz4"), core_options.GetChangelogFileCompression()); ASSERT_TRUE(core_options.NeedLookup()); diff --git a/src/paimon/core/mergetree/merge_tree_writer.cpp b/src/paimon/core/mergetree/merge_tree_writer.cpp index 88df1b16e..2b97ba169 100644 --- a/src/paimon/core/mergetree/merge_tree_writer.cpp +++ b/src/paimon/core/mergetree/merge_tree_writer.cpp @@ -20,6 +20,7 @@ #include #include +#include #include #include #include @@ -49,6 +50,55 @@ namespace paimon { +namespace { + +Status ValidateReaders(const std::vector>& readers) { + if (readers.empty()) { + return Status::Invalid("sorted readers must not be empty"); + } + for (const std::unique_ptr& reader : readers) { + if (reader == nullptr) { + return Status::Invalid("sorted readers must not contain null reader"); + } + } + return Status::OK(); +} + +Result>> WriteSortMergeReaderToFiles( + std::unique_ptr&& sort_merge_reader, + RollingFileWriter>* rolling_writer, + int32_t write_batch_size, + const AsyncKeyValueProducerAndConsumer::ConsumerCreator& + create_consumer) { + assert(sort_merge_reader != nullptr); + assert(rolling_writer != nullptr); + + std::unique_ptr producer = + std::make_unique(std::move(sort_merge_reader), + write_batch_size); + auto producer_and_consumer = + std::make_unique>( + std::move(producer), create_consumer, /*consumer_thread_num=*/1); + ScopeGuard producer_guard([&]() -> void { producer_and_consumer->Close(); }); + + while (true) { + PAIMON_ASSIGN_OR_RAISE(KeyValueBatch key_value_batch, producer_and_consumer->NextBatch()); + if (key_value_batch.batch == nullptr) { + break; + } + PAIMON_RETURN_NOT_OK(rolling_writer->Write(std::move(key_value_batch))); + } + + PAIMON_RETURN_NOT_OK(rolling_writer->Close()); + PAIMON_ASSIGN_OR_RAISE(std::vector> flushed_files, + rolling_writer->GetResult()); + producer_and_consumer->Close(); + producer_guard.Release(); + return flushed_files; +} + +} // namespace + Result> MergeTreeWriter::Create( int64_t last_sequence_number, const std::vector& trimmed_primary_keys, const std::shared_ptr& path_factory, @@ -169,14 +219,7 @@ Status MergeTreeWriter::WriteSortedReadersToFiles( } } }); - if (readers.empty()) { - return Status::Invalid("sorted readers must not be empty"); - } - for (const std::unique_ptr& reader : readers) { - if (reader == nullptr) { - return Status::Invalid("sorted readers must not contain null reader"); - } - } + PAIMON_RETURN_NOT_OK(ValidateReaders(readers)); // prepare loser tree sort merge reader auto sort_merge_reader = std::make_unique( @@ -187,28 +230,13 @@ Status MergeTreeWriter::WriteSortedReadersToFiles( -> Result>> { return KeyValueMetaProjectionConsumer::Create(target_schema, pool); }; - // consumer batch size is WriteBatchSize - std::unique_ptr producer = - std::make_unique(std::move(sort_merge_reader), - options_.GetWriteBatchSize()); - auto async_key_value_producer_consumer = - std::make_unique>( - std::move(producer), create_consumer, /*projection_thread_num=*/1); - ScopeGuard async_readers_guard([&]() -> void { async_key_value_producer_consumer->Close(); }); std::unique_ptr>> rolling_writer; PAIMON_ASSIGN_OR_RAISE(rolling_writer, CreateRollingRowWriter()); ScopeGuard abort_writer_guard([&]() -> void { rolling_writer->Abort(); }); - while (true) { - PAIMON_ASSIGN_OR_RAISE(KeyValueBatch key_value_batch, - async_key_value_producer_consumer->NextBatch()); - if (key_value_batch.batch == nullptr) { - break; - } - PAIMON_RETURN_NOT_OK(rolling_writer->Write(std::move(key_value_batch))); - } - PAIMON_RETURN_NOT_OK(rolling_writer->Close()); - PAIMON_ASSIGN_OR_RAISE(std::vector> flushed_files, - rolling_writer->GetResult()); + PAIMON_ASSIGN_OR_RAISE( + std::vector> flushed_files, + WriteSortMergeReaderToFiles(std::move(sort_merge_reader), rolling_writer.get(), + options_.GetWriteBatchSize(), create_consumer)); abort_writer_guard.Release(); for (const std::shared_ptr& flushed_file : flushed_files) { @@ -336,60 +364,98 @@ Status MergeTreeWriter::FlushWriteBuffer(bool wait_for_latest_compaction, return KeyValueMetaProjectionConsumer::Create(target_schema, pool); }; - std::unique_ptr> - async_changelog_producer_consumer; - std::unique_ptr>> - changelog_writer; - std::vector> flushed_changelog_files; - ScopeGuard changelog_write_guard([&]() -> void { - if (changelog_writer) { - changelog_writer->Abort(); - } - if (async_changelog_producer_consumer) { - async_changelog_producer_consumer->Close(); - } - }); if (options_.GetChangelogProducer() == ChangelogProducer::INPUT) { PAIMON_ASSIGN_OR_RAISE(std::vector> raw_readers, write_buffer_->CreateRawReaders()); + ScopeGuard raw_readers_guard([&]() -> void { + for (std::unique_ptr& reader : raw_readers) { + if (reader != nullptr) { + reader->Close(); + } + } + }); + PAIMON_ASSIGN_OR_RAISE(std::vector> readers, + write_buffer_->CreateReaders()); + ScopeGuard readers_guard([&]() -> void { + for (std::unique_ptr& reader : readers) { + if (reader != nullptr) { + reader->Close(); + } + } + }); + PAIMON_RETURN_NOT_OK(ValidateReaders(raw_readers)); + PAIMON_RETURN_NOT_OK(ValidateReaders(readers)); + + std::unique_ptr>> + changelog_writer; + PAIMON_ASSIGN_OR_RAISE(changelog_writer, CreateRollingChangelogWriter()); + ScopeGuard changelog_writer_guard([&]() -> void { changelog_writer->Abort(); }); + std::unique_ptr>> + data_writer; + PAIMON_ASSIGN_OR_RAISE(data_writer, CreateRollingRowWriter()); + ScopeGuard data_writer_guard([&]() -> void { data_writer->Abort(); }); + auto raw_sort_merge_reader = std::make_unique( std::move(raw_readers), key_comparator_, user_defined_seq_comparator_, /*merge_function_wrapper=*/nullptr); - std::unique_ptr producer = - std::make_unique(std::move(raw_sort_merge_reader), - options_.GetWriteBatchSize()); - async_changelog_producer_consumer = - std::make_unique>( - std::move(producer), create_consumer, /*projection_thread_num=*/1); - PAIMON_ASSIGN_OR_RAISE(changelog_writer, CreateRollingChangelogWriter()); - while (true) { - PAIMON_ASSIGN_OR_RAISE(KeyValueBatch key_value_batch, - async_changelog_producer_consumer->NextBatch()); - if (key_value_batch.batch == nullptr) { - break; + raw_readers_guard.Release(); + auto sort_merge_reader = std::make_unique( + std::move(readers), key_comparator_, user_defined_seq_comparator_, + merge_function_wrapper_); + readers_guard.Release(); + + const int32_t write_batch_size = options_.GetWriteBatchSize(); + std::vector> data_files; + std::vector> changelog_files; + if (options_.InputChangelogParallelWriteEnabled()) { + std::future>>> changelog_future = + std::async(std::launch::async, + [raw_sort_merge_reader = std::move(raw_sort_merge_reader), + changelog_writer = changelog_writer.get(), write_batch_size, + create_consumer]() mutable { + return WriteSortMergeReaderToFiles( + std::move(raw_sort_merge_reader), changelog_writer, + write_batch_size, create_consumer); + }); + Result>> data_result = + WriteSortMergeReaderToFiles(std::move(sort_merge_reader), data_writer.get(), + write_batch_size, create_consumer); + Result>> changelog_result = + changelog_future.get(); + if (!data_result.ok()) { + return data_result.status(); + } + if (!changelog_result.ok()) { + return changelog_result.status(); } - PAIMON_RETURN_NOT_OK(changelog_writer->Write(std::move(key_value_batch))); + data_files = std::move(data_result).value(); + changelog_files = std::move(changelog_result).value(); + } else { + PAIMON_ASSIGN_OR_RAISE(changelog_files, + WriteSortMergeReaderToFiles( + std::move(raw_sort_merge_reader), changelog_writer.get(), + write_batch_size, create_consumer)); + PAIMON_ASSIGN_OR_RAISE( + data_files, + WriteSortMergeReaderToFiles(std::move(sort_merge_reader), data_writer.get(), + write_batch_size, create_consumer)); } - PAIMON_RETURN_NOT_OK(changelog_writer->Close()); - PAIMON_ASSIGN_OR_RAISE(flushed_changelog_files, changelog_writer->GetResult()); - } - - // 1. flush write buffer to get sorted readers - // Flush write buffer to get sorted and merged data readers. - PAIMON_ASSIGN_OR_RAISE(std::vector> readers, - write_buffer_->CreateReaders()); - PAIMON_RETURN_NOT_OK(WriteSortedReadersToFiles(std::move(readers))); - if (async_changelog_producer_consumer) { - async_changelog_producer_consumer->Close(); - } - - new_changelog_files_.insert(new_changelog_files_.end(), flushed_changelog_files.begin(), - flushed_changelog_files.end()); - - changelog_write_guard.Release(); - if (changelog_writer) { + changelog_writer_guard.Release(); + data_writer_guard.Release(); + new_changelog_files_.insert(new_changelog_files_.end(), changelog_files.begin(), + changelog_files.end()); + new_files_.insert(new_files_.end(), data_files.begin(), data_files.end()); + for (const std::shared_ptr& flushed_file : data_files) { + PAIMON_RETURN_NOT_OK(compact_manager_->AddNewFile(flushed_file)); + } + metrics_->Merge(data_writer->GetMetrics()); metrics_->Merge(changelog_writer->GetMetrics()); + } else { + // Flush write buffer to get sorted and merged data readers. + PAIMON_ASSIGN_OR_RAISE(std::vector> readers, + write_buffer_->CreateReaders()); + PAIMON_RETURN_NOT_OK(WriteSortedReadersToFiles(std::move(readers))); } } PAIMON_RETURN_NOT_OK(TrySyncLatestCompaction(wait_for_latest_compaction)); @@ -417,28 +483,25 @@ Result MergeTreeWriter::DrainIncrement() { Result>>> MergeTreeWriter::CreateRollingRowWriter() const { - PAIMON_ASSIGN_OR_RAISE( - std::shared_ptr factory, - KeyValueDataFileWriterFactories::Create(options_, schema_id_, write_schema_, /*level=*/0, - FileSource::Append(), trimmed_primary_keys_, - path_factory_, /*create_stats_extractor=*/true, - /*is_changelog=*/false, pool_)); - return std::make_unique>>( - options_.GetTargetFileSize(/*has_primary_key=*/true), options_.GetTargetFileRowNum(), - factory); + return CreateRollingWriter(/*is_changelog=*/false, options_.GetTargetFileRowNum()); } Result>>> MergeTreeWriter::CreateRollingChangelogWriter() const { + return CreateRollingWriter(/*is_changelog=*/true, + /*target_file_row_num=*/std::numeric_limits::max()); +} + +Result>>> +MergeTreeWriter::CreateRollingWriter(bool is_changelog, int64_t target_file_row_num) const { PAIMON_ASSIGN_OR_RAISE( std::shared_ptr factory, KeyValueDataFileWriterFactories::Create(options_, schema_id_, write_schema_, /*level=*/0, FileSource::Append(), trimmed_primary_keys_, path_factory_, /*create_stats_extractor=*/true, - /*is_changelog=*/true, pool_)); + is_changelog, pool_)); return std::make_unique>>( - options_.GetTargetFileSize(/*has_primary_key=*/true), - /*target_file_row_num=*/std::numeric_limits::max(), factory); + options_.GetTargetFileSize(/*has_primary_key=*/true), target_file_row_num, factory); } } // namespace paimon diff --git a/src/paimon/core/mergetree/merge_tree_writer.h b/src/paimon/core/mergetree/merge_tree_writer.h index 9f73e43be..1a2c2c3dd 100644 --- a/src/paimon/core/mergetree/merge_tree_writer.h +++ b/src/paimon/core/mergetree/merge_tree_writer.h @@ -108,6 +108,9 @@ class MergeTreeWriter : public BatchWriter { Result>>> CreateRollingChangelogWriter() const; + Result>>> + CreateRollingWriter(bool is_changelog, int64_t target_file_row_num) const; + Status TrySyncLatestCompaction(bool blocking); Status UpdateCompactResult(const std::shared_ptr& compact_result); Status UpdateCompactDeletionFile(const std::shared_ptr& new_deletion_file); diff --git a/src/paimon/core/mergetree/merge_tree_writer_test.cpp b/src/paimon/core/mergetree/merge_tree_writer_test.cpp index 8a152e2f7..ec10b2610 100644 --- a/src/paimon/core/mergetree/merge_tree_writer_test.cpp +++ b/src/paimon/core/mergetree/merge_tree_writer_test.cpp @@ -396,8 +396,12 @@ TEST_P(MergeTreeWriterTest, TestInputChangelog) { const DataIncrement& data_increment = commit_increment.GetNewFilesIncrement(); ASSERT_EQ(1, data_increment.NewFiles().size()); ASSERT_EQ(1, data_increment.ChangelogFiles().size()); - ASSERT_EQ("data-" + uuid + "-1.orc", data_increment.NewFiles()[0]->file_name); - ASSERT_EQ("changes-" + uuid + "-0.parquet", data_increment.ChangelogFiles()[0]->file_name); + ASSERT_TRUE( + StringUtils::StartsWith(data_increment.NewFiles()[0]->file_name, "data-" + uuid + "-")); + ASSERT_TRUE(StringUtils::EndsWith(data_increment.NewFiles()[0]->file_name, ".orc")); + ASSERT_TRUE(StringUtils::StartsWith(data_increment.ChangelogFiles()[0]->file_name, + "changes-" + uuid + "-")); + ASSERT_TRUE(StringUtils::EndsWith(data_increment.ChangelogFiles()[0]->file_name, ".parquet")); ASSERT_EQ(2, data_increment.NewFiles()[0]->row_count); ASSERT_EQ(4, data_increment.ChangelogFiles()[0]->row_count); @@ -423,6 +427,49 @@ TEST_P(MergeTreeWriterTest, TestInputChangelog) { expected_changelog, "parquet"); } +TEST_P(MergeTreeWriterTest, TestInputChangelogWithParallelWriteDisabled) { + ASSERT_OK_AND_ASSIGN( + CoreOptions options, + CoreOptions::FromMap( + {{Options::FILE_FORMAT, "orc"}, + {Options::CHANGELOG_PRODUCER, "input"}, + {Options::CHANGELOG_PRODUCER_INPUT_PARALLEL_WRITE_ENABLED, "false"}})); + + auto dir = UniqueTestDirectory::Create(); + ASSERT_TRUE(dir); + auto path_factory = std::make_shared(); + ASSERT_OK(path_factory->Init(dir->Str(), "orc", options.DataFilePrefix(), nullptr)); + std::string uuid = path_factory->uuid_; + + ASSERT_OK_AND_ASSIGN(auto merge_writer, + CreateMergeWriter(/*last_sequence_number=*/-1, dir->Str(), path_factory, + /*schema_id=*/1, options)); + std::shared_ptr array = + arrow::ipc::internal::json::ArrayFromJSON(value_type_, R"([ + ["Bob", 20, 0, 20.0], + ["Alice", 10, 0, 10.0], + ["Alice", 11, 0, 11.0], + ["Bob", 21, 0, 21.0] + ])") + .ValueOrDie(); + WriteBatch(array, /*row_kinds=*/{}, merge_writer.get()); + if (GetParam()) { + ASSERT_OK(merge_writer->FlushMemory()); + } + + ASSERT_OK_AND_ASSIGN(CommitIncrement commit_increment, + merge_writer->PrepareCommit(/*wait_compaction=*/false)); + ASSERT_OK(merge_writer->Close()); + + const DataIncrement& data_increment = commit_increment.GetNewFilesIncrement(); + ASSERT_EQ(1, data_increment.NewFiles().size()); + ASSERT_EQ(1, data_increment.ChangelogFiles().size()); + ASSERT_EQ("changelog-" + uuid + "-0.orc", data_increment.ChangelogFiles()[0]->file_name); + ASSERT_EQ("data-" + uuid + "-1.orc", data_increment.NewFiles()[0]->file_name); + ASSERT_EQ(4, data_increment.ChangelogFiles()[0]->row_count); + ASSERT_EQ(2, data_increment.NewFiles()[0]->row_count); +} + TEST_P(MergeTreeWriterTest, TestInputChangelogIgnoresTargetFileRowNum) { ASSERT_OK_AND_ASSIGN(CoreOptions options, CoreOptions::FromMap({{Options::FILE_FORMAT, "orc"}, From 8d0cea59be1701c2a2373257c3f366f670dc7af5 Mon Sep 17 00:00:00 2001 From: "lisizhuo.lsz" Date: Wed, 16 Sep 2026 10:08:19 +0800 Subject: [PATCH 2/4] fix ci --- src/paimon/core/mergetree/merge_tree_writer_test.cpp | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/paimon/core/mergetree/merge_tree_writer_test.cpp b/src/paimon/core/mergetree/merge_tree_writer_test.cpp index ec10b2610..94bb40b3b 100644 --- a/src/paimon/core/mergetree/merge_tree_writer_test.cpp +++ b/src/paimon/core/mergetree/merge_tree_writer_test.cpp @@ -433,7 +433,7 @@ TEST_P(MergeTreeWriterTest, TestInputChangelogWithParallelWriteDisabled) { CoreOptions::FromMap( {{Options::FILE_FORMAT, "orc"}, {Options::CHANGELOG_PRODUCER, "input"}, - {Options::CHANGELOG_PRODUCER_INPUT_PARALLEL_WRITE_ENABLED, "false"}})); + {Options::CHANGELOG_PRODUCER_INPUT_PARALLEL_WRITE, "false"}})); auto dir = UniqueTestDirectory::Create(); ASSERT_TRUE(dir); From 0bb240e1283df509e5c463c4f2d8b4cc18a50502 Mon Sep 17 00:00:00 2001 From: "lisizhuo.lsz" Date: Wed, 16 Sep 2026 14:06:48 +0800 Subject: [PATCH 3/4] fix ci --- .../core/mergetree/merge_tree_writer.cpp | 98 +++++++++++-------- 1 file changed, 57 insertions(+), 41 deletions(-) diff --git a/src/paimon/core/mergetree/merge_tree_writer.cpp b/src/paimon/core/mergetree/merge_tree_writer.cpp index 2b97ba169..5be825334 100644 --- a/src/paimon/core/mergetree/merge_tree_writer.cpp +++ b/src/paimon/core/mergetree/merge_tree_writer.cpp @@ -64,22 +64,24 @@ Status ValidateReaders(const std::vector>& return Status::OK(); } -Result>> WriteSortMergeReaderToFiles( - std::unique_ptr&& sort_merge_reader, - RollingFileWriter>* rolling_writer, - int32_t write_batch_size, +std::unique_ptr> +CreateSortMergeProducerAndConsumer( + std::unique_ptr&& sort_merge_reader, int32_t write_batch_size, const AsyncKeyValueProducerAndConsumer::ConsumerCreator& create_consumer) { assert(sort_merge_reader != nullptr); - assert(rolling_writer != nullptr); - std::unique_ptr producer = std::make_unique(std::move(sort_merge_reader), write_batch_size); - auto producer_and_consumer = - std::make_unique>( - std::move(producer), create_consumer, /*consumer_thread_num=*/1); - ScopeGuard producer_guard([&]() -> void { producer_and_consumer->Close(); }); + return std::make_unique>( + std::move(producer), create_consumer, /*consumer_thread_num=*/1); +} + +Result>> WriteKeyValueBatchesToFiles( + AsyncKeyValueProducerAndConsumer* producer_and_consumer, + RollingFileWriter>* rolling_writer) { + assert(producer_and_consumer != nullptr); + assert(rolling_writer != nullptr); while (true) { PAIMON_ASSIGN_OR_RAISE(KeyValueBatch key_value_batch, producer_and_consumer->NextBatch()); @@ -92,8 +94,6 @@ Result>> WriteSortMergeReaderToFiles( PAIMON_RETURN_NOT_OK(rolling_writer->Close()); PAIMON_ASSIGN_OR_RAISE(std::vector> flushed_files, rolling_writer->GetResult()); - producer_and_consumer->Close(); - producer_guard.Release(); return flushed_files; } @@ -230,13 +230,18 @@ Status MergeTreeWriter::WriteSortedReadersToFiles( -> Result>> { return KeyValueMetaProjectionConsumer::Create(target_schema, pool); }; + std::unique_ptr> + producer_and_consumer = CreateSortMergeProducerAndConsumer( + std::move(sort_merge_reader), options_.GetWriteBatchSize(), create_consumer); + ScopeGuard producer_guard([&]() -> void { producer_and_consumer->Close(); }); std::unique_ptr>> rolling_writer; PAIMON_ASSIGN_OR_RAISE(rolling_writer, CreateRollingRowWriter()); ScopeGuard abort_writer_guard([&]() -> void { rolling_writer->Abort(); }); PAIMON_ASSIGN_OR_RAISE( std::vector> flushed_files, - WriteSortMergeReaderToFiles(std::move(sort_merge_reader), rolling_writer.get(), - options_.GetWriteBatchSize(), create_consumer)); + WriteKeyValueBatchesToFiles(producer_and_consumer.get(), rolling_writer.get())); + producer_and_consumer->Close(); + producer_guard.Release(); abort_writer_guard.Release(); for (const std::shared_ptr& flushed_file : flushed_files) { @@ -386,15 +391,6 @@ Status MergeTreeWriter::FlushWriteBuffer(bool wait_for_latest_compaction, PAIMON_RETURN_NOT_OK(ValidateReaders(raw_readers)); PAIMON_RETURN_NOT_OK(ValidateReaders(readers)); - std::unique_ptr>> - changelog_writer; - PAIMON_ASSIGN_OR_RAISE(changelog_writer, CreateRollingChangelogWriter()); - ScopeGuard changelog_writer_guard([&]() -> void { changelog_writer->Abort(); }); - std::unique_ptr>> - data_writer; - PAIMON_ASSIGN_OR_RAISE(data_writer, CreateRollingRowWriter()); - ScopeGuard data_writer_guard([&]() -> void { data_writer->Abort(); }); - auto raw_sort_merge_reader = std::make_unique( std::move(raw_readers), key_comparator_, user_defined_seq_comparator_, /*merge_function_wrapper=*/nullptr); @@ -405,21 +401,40 @@ Status MergeTreeWriter::FlushWriteBuffer(bool wait_for_latest_compaction, readers_guard.Release(); const int32_t write_batch_size = options_.GetWriteBatchSize(); + std::unique_ptr> + changelog_producer_and_consumer = CreateSortMergeProducerAndConsumer( + std::move(raw_sort_merge_reader), write_batch_size, create_consumer); + std::unique_ptr> + data_producer_and_consumer = CreateSortMergeProducerAndConsumer( + std::move(sort_merge_reader), write_batch_size, create_consumer); + ScopeGuard producers_guard([&]() -> void { + data_producer_and_consumer->Close(); + changelog_producer_and_consumer->Close(); + }); + + // Writers retain key rows backed by the producers' source readers. On error, this + // declaration order destroys the writers before the producer guard closes the readers. + std::unique_ptr>> + changelog_writer; + PAIMON_ASSIGN_OR_RAISE(changelog_writer, CreateRollingChangelogWriter()); + ScopeGuard changelog_writer_guard([&]() -> void { changelog_writer->Abort(); }); + std::unique_ptr>> + data_writer; + PAIMON_ASSIGN_OR_RAISE(data_writer, CreateRollingRowWriter()); + ScopeGuard data_writer_guard([&]() -> void { data_writer->Abort(); }); + std::vector> data_files; std::vector> changelog_files; if (options_.InputChangelogParallelWriteEnabled()) { std::future>>> changelog_future = - std::async(std::launch::async, - [raw_sort_merge_reader = std::move(raw_sort_merge_reader), - changelog_writer = changelog_writer.get(), write_batch_size, - create_consumer]() mutable { - return WriteSortMergeReaderToFiles( - std::move(raw_sort_merge_reader), changelog_writer, - write_batch_size, create_consumer); - }); + std::async(std::launch::async, [producer_and_consumer = + changelog_producer_and_consumer.get(), + writer = changelog_writer.get()]() { + return WriteKeyValueBatchesToFiles(producer_and_consumer, writer); + }); Result>> data_result = - WriteSortMergeReaderToFiles(std::move(sort_merge_reader), data_writer.get(), - write_batch_size, create_consumer); + WriteKeyValueBatchesToFiles(data_producer_and_consumer.get(), + data_writer.get()); Result>> changelog_result = changelog_future.get(); if (!data_result.ok()) { @@ -431,16 +446,17 @@ Status MergeTreeWriter::FlushWriteBuffer(bool wait_for_latest_compaction, data_files = std::move(data_result).value(); changelog_files = std::move(changelog_result).value(); } else { - PAIMON_ASSIGN_OR_RAISE(changelog_files, - WriteSortMergeReaderToFiles( - std::move(raw_sort_merge_reader), changelog_writer.get(), - write_batch_size, create_consumer)); - PAIMON_ASSIGN_OR_RAISE( - data_files, - WriteSortMergeReaderToFiles(std::move(sort_merge_reader), data_writer.get(), - write_batch_size, create_consumer)); + PAIMON_ASSIGN_OR_RAISE(changelog_files, WriteKeyValueBatchesToFiles( + changelog_producer_and_consumer.get(), + changelog_writer.get())); + PAIMON_ASSIGN_OR_RAISE(data_files, + WriteKeyValueBatchesToFiles(data_producer_and_consumer.get(), + data_writer.get())); } + data_producer_and_consumer->Close(); + changelog_producer_and_consumer->Close(); + producers_guard.Release(); changelog_writer_guard.Release(); data_writer_guard.Release(); new_changelog_files_.insert(new_changelog_files_.end(), changelog_files.begin(), From a57420e0199a3cc5d919a8f40940b84d27f3c79b Mon Sep 17 00:00:00 2001 From: "lisizhuo.lsz" Date: Thu, 17 Sep 2026 10:54:54 +0800 Subject: [PATCH 4/4] fix pre-commit & xinyu comment --- include/paimon/defs.h | 3 ++- src/paimon/core/core_options.cpp | 5 ++--- src/paimon/core/mergetree/merge_tree_writer.cpp | 11 +++++------ src/paimon/core/mergetree/merge_tree_writer_test.cpp | 7 +++---- 4 files changed, 12 insertions(+), 14 deletions(-) diff --git a/include/paimon/defs.h b/include/paimon/defs.h index 33d83cd6d..f51eb4520 100644 --- a/include/paimon/defs.h +++ b/include/paimon/defs.h @@ -422,7 +422,8 @@ struct PAIMON_EXPORT Options { 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". + /// 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 diff --git a/src/paimon/core/core_options.cpp b/src/paimon/core/core_options.cpp index 17472d072..bc1bb994a 100644 --- a/src/paimon/core/core_options.cpp +++ b/src/paimon/core/core_options.cpp @@ -755,9 +755,8 @@ struct CoreOptions::Impl { &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(Options::CHANGELOG_PRODUCER_INPUT_PARALLEL_WRITE, - &input_changelog_parallel_write_enabled)); + PAIMON_RETURN_NOT_OK(parser.Parse(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(Options::PARTIAL_UPDATE_REMOVE_RECORD_ON_DELETE, &partial_update_remove_record_on_delete)); diff --git a/src/paimon/core/mergetree/merge_tree_writer.cpp b/src/paimon/core/mergetree/merge_tree_writer.cpp index 5be825334..e5ac1af68 100644 --- a/src/paimon/core/mergetree/merge_tree_writer.cpp +++ b/src/paimon/core/mergetree/merge_tree_writer.cpp @@ -510,12 +510,11 @@ MergeTreeWriter::CreateRollingChangelogWriter() const { Result>>> MergeTreeWriter::CreateRollingWriter(bool is_changelog, int64_t target_file_row_num) const { - PAIMON_ASSIGN_OR_RAISE( - std::shared_ptr factory, - KeyValueDataFileWriterFactories::Create(options_, schema_id_, write_schema_, /*level=*/0, - FileSource::Append(), trimmed_primary_keys_, - path_factory_, /*create_stats_extractor=*/true, - is_changelog, pool_)); + PAIMON_ASSIGN_OR_RAISE(std::shared_ptr factory, + KeyValueDataFileWriterFactories::Create( + options_, schema_id_, write_schema_, /*level=*/0, + FileSource::Append(), trimmed_primary_keys_, path_factory_, + /*create_stats_extractor=*/true, is_changelog, pool_)); return std::make_unique>>( options_.GetTargetFileSize(/*has_primary_key=*/true), target_file_row_num, factory); } diff --git a/src/paimon/core/mergetree/merge_tree_writer_test.cpp b/src/paimon/core/mergetree/merge_tree_writer_test.cpp index 94bb40b3b..e557af325 100644 --- a/src/paimon/core/mergetree/merge_tree_writer_test.cpp +++ b/src/paimon/core/mergetree/merge_tree_writer_test.cpp @@ -430,10 +430,9 @@ TEST_P(MergeTreeWriterTest, TestInputChangelog) { TEST_P(MergeTreeWriterTest, TestInputChangelogWithParallelWriteDisabled) { ASSERT_OK_AND_ASSIGN( CoreOptions options, - CoreOptions::FromMap( - {{Options::FILE_FORMAT, "orc"}, - {Options::CHANGELOG_PRODUCER, "input"}, - {Options::CHANGELOG_PRODUCER_INPUT_PARALLEL_WRITE, "false"}})); + CoreOptions::FromMap({{Options::FILE_FORMAT, "orc"}, + {Options::CHANGELOG_PRODUCER, "input"}, + {Options::CHANGELOG_PRODUCER_INPUT_PARALLEL_WRITE, "false"}})); auto dir = UniqueTestDirectory::Create(); ASSERT_TRUE(dir);