diff --git a/include/paimon/defs.h b/include/paimon/defs.h index cfa08fa00..f51eb4520 100644 --- a/include/paimon/defs.h +++ b/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[]; 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..bc1bb994a 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,10 @@ 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 +1635,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..e5ac1af68 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(); +} + +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); + std::unique_ptr producer = + std::make_unique(std::move(sort_merge_reader), + write_batch_size); + 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()); + 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()); + 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,18 @@ 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> + 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(); }); - 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, + 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) { @@ -336,60 +369,109 @@ 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)); + 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); + 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::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()); - while (true) { - PAIMON_ASSIGN_OR_RAISE(KeyValueBatch key_value_batch, - async_changelog_producer_consumer->NextBatch()); - if (key_value_batch.batch == nullptr) { - break; + 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, [producer_and_consumer = + changelog_producer_and_consumer.get(), + writer = changelog_writer.get()]() { + return WriteKeyValueBatchesToFiles(producer_and_consumer, writer); + }); + Result>> data_result = + WriteKeyValueBatchesToFiles(data_producer_and_consumer.get(), + data_writer.get()); + Result>> changelog_result = + changelog_future.get(); + if (!data_result.ok()) { + return data_result.status(); } - PAIMON_RETURN_NOT_OK(changelog_writer->Write(std::move(key_value_batch))); + if (!changelog_result.ok()) { + return changelog_result.status(); + } + data_files = std::move(data_result).value(); + changelog_files = std::move(changelog_result).value(); + } else { + 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())); } - 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) { + 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(), + 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 +499,24 @@ 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 { - 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_)); + 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, 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..e557af325 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,48 @@ 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, "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"},