This is an automated email from the ASF dual-hosted git repository.
lxy-9602 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/paimon-cpp.git
The following commit(s) were added to refs/heads/main by this push:
new 200e4850 perf(write): write input changelog and data files in parallel
(#350)
200e4850 is described below
commit 200e485041a82bc3feb10d7eac3e0997fc681168
Author: lszskye <[email protected]>
AuthorDate: Wed Sep 16 22:15:02 2026 -0700
perf(write): write input changelog and data files in parallel (#350)
---
include/paimon/defs.h | 5 +
src/paimon/common/defs.cpp | 2 +
src/paimon/core/core_options.cpp | 9 +
src/paimon/core/core_options.h | 1 +
src/paimon/core/core_options_test.cpp | 3 +
src/paimon/core/mergetree/merge_tree_writer.cpp | 252 ++++++++++++++-------
src/paimon/core/mergetree/merge_tree_writer.h | 3 +
.../core/mergetree/merge_tree_writer_test.cpp | 50 +++-
8 files changed, 236 insertions(+), 89 deletions(-)
diff --git a/include/paimon/defs.h b/include/paimon/defs.h
index cfa08fa0..f51eb452 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 de1d4577..88085f78 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 37fadc8e..bc1bb994 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<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;
}
diff --git a/src/paimon/core/core_options.h b/src/paimon/core/core_options.h
index 86771e6a..6dadc680 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<std::string>& GetChangelogRowDeduplicateIgnoreFields()
const;
+ bool InputChangelogParallelWriteEnabled() const;
std::string ChangelogFilePrefix() const;
std::shared_ptr<FileFormat> GetChangelogFileFormat() const;
std::optional<std::string> GetChangelogFileCompression() const;
diff --git a/src/paimon/core/core_options_test.cpp
b/src/paimon/core/core_options_test.cpp
index 3e489a4c..be9fa823 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<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());
diff --git a/src/paimon/core/mergetree/merge_tree_writer.cpp
b/src/paimon/core/mergetree/merge_tree_writer.cpp
index 88df1b16..e5ac1af6 100644
--- a/src/paimon/core/mergetree/merge_tree_writer.cpp
+++ b/src/paimon/core/mergetree/merge_tree_writer.cpp
@@ -20,6 +20,7 @@
#include <algorithm>
#include <cassert>
+#include <future>
#include <limits>
#include <map>
#include <unordered_set>
@@ -49,6 +50,55 @@
namespace paimon {
+namespace {
+
+Status ValidateReaders(const
std::vector<std::unique_ptr<KeyValueRecordReader>>& readers) {
+ if (readers.empty()) {
+ return Status::Invalid("sorted readers must not be empty");
+ }
+ for (const std::unique_ptr<KeyValueRecordReader>& reader : readers) {
+ if (reader == nullptr) {
+ return Status::Invalid("sorted readers must not contain null
reader");
+ }
+ }
+ return Status::OK();
+}
+
+std::unique_ptr<AsyncKeyValueProducerAndConsumer<KeyValue, KeyValueBatch>>
+CreateSortMergeProducerAndConsumer(
+ std::unique_ptr<SortMergeReader>&& sort_merge_reader, int32_t
write_batch_size,
+ const AsyncKeyValueProducerAndConsumer<KeyValue,
KeyValueBatch>::ConsumerCreator&
+ create_consumer) {
+ assert(sort_merge_reader != nullptr);
+ std::unique_ptr<AsyncKeyValueBatchProducer> producer =
+
std::make_unique<SortMergeReaderBatchProducer>(std::move(sort_merge_reader),
+ write_batch_size);
+ return std::make_unique<AsyncKeyValueProducerAndConsumer<KeyValue,
KeyValueBatch>>(
+ std::move(producer), create_consumer, /*consumer_thread_num=*/1);
+}
+
+Result<std::vector<std::shared_ptr<DataFileMeta>>> WriteKeyValueBatchesToFiles(
+ AsyncKeyValueProducerAndConsumer<KeyValue, KeyValueBatch>*
producer_and_consumer,
+ RollingFileWriter<KeyValueBatch, std::shared_ptr<DataFileMeta>>*
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<std::shared_ptr<DataFileMeta>>
flushed_files,
+ rolling_writer->GetResult());
+ return flushed_files;
+}
+
+} // namespace
+
Result<std::shared_ptr<MergeTreeWriter>> MergeTreeWriter::Create(
int64_t last_sequence_number, const std::vector<std::string>&
trimmed_primary_keys,
const std::shared_ptr<DataFilePathFactory>& 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<KeyValueRecordReader>& 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<SortMergeReaderWithLoserTree>(
@@ -187,28 +230,18 @@ Status MergeTreeWriter::WriteSortedReadersToFiles(
-> Result<std::unique_ptr<RowToArrowArrayConverter<KeyValue,
KeyValueBatch>>> {
return KeyValueMetaProjectionConsumer::Create(target_schema, pool);
};
- // consumer batch size is WriteBatchSize
- std::unique_ptr<AsyncKeyValueBatchProducer> producer =
-
std::make_unique<SortMergeReaderBatchProducer>(std::move(sort_merge_reader),
-
options_.GetWriteBatchSize());
- auto async_key_value_producer_consumer =
- std::make_unique<AsyncKeyValueProducerAndConsumer<KeyValue,
KeyValueBatch>>(
- std::move(producer), create_consumer, /*projection_thread_num=*/1);
- ScopeGuard async_readers_guard([&]() -> void {
async_key_value_producer_consumer->Close(); });
+ std::unique_ptr<AsyncKeyValueProducerAndConsumer<KeyValue, KeyValueBatch>>
+ producer_and_consumer = CreateSortMergeProducerAndConsumer(
+ std::move(sort_merge_reader), options_.GetWriteBatchSize(),
create_consumer);
+ ScopeGuard producer_guard([&]() -> void { producer_and_consumer->Close();
});
std::unique_ptr<RollingFileWriter<KeyValueBatch,
std::shared_ptr<DataFileMeta>>> 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<std::shared_ptr<DataFileMeta>>
flushed_files,
- rolling_writer->GetResult());
+ PAIMON_ASSIGN_OR_RAISE(
+ std::vector<std::shared_ptr<DataFileMeta>> 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<DataFileMeta>& 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<AsyncKeyValueProducerAndConsumer<KeyValue,
KeyValueBatch>>
- async_changelog_producer_consumer;
- std::unique_ptr<RollingFileWriter<KeyValueBatch,
std::shared_ptr<DataFileMeta>>>
- changelog_writer;
- std::vector<std::shared_ptr<DataFileMeta>> 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<std::unique_ptr<KeyValueRecordReader>>
raw_readers,
write_buffer_->CreateRawReaders());
+ ScopeGuard raw_readers_guard([&]() -> void {
+ for (std::unique_ptr<KeyValueRecordReader>& reader :
raw_readers) {
+ if (reader != nullptr) {
+ reader->Close();
+ }
+ }
+ });
+
PAIMON_ASSIGN_OR_RAISE(std::vector<std::unique_ptr<KeyValueRecordReader>>
readers,
+ write_buffer_->CreateReaders());
+ ScopeGuard readers_guard([&]() -> void {
+ for (std::unique_ptr<KeyValueRecordReader>& 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<SortMergeReaderWithMinHeap>(
std::move(raw_readers), key_comparator_,
user_defined_seq_comparator_,
/*merge_function_wrapper=*/nullptr);
- std::unique_ptr<AsyncKeyValueBatchProducer> producer =
-
std::make_unique<SortMergeReaderBatchProducer>(std::move(raw_sort_merge_reader),
-
options_.GetWriteBatchSize());
- async_changelog_producer_consumer =
- std::make_unique<AsyncKeyValueProducerAndConsumer<KeyValue,
KeyValueBatch>>(
- std::move(producer), create_consumer,
/*projection_thread_num=*/1);
+ raw_readers_guard.Release();
+ auto sort_merge_reader =
std::make_unique<SortMergeReaderWithLoserTree>(
+ 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<AsyncKeyValueProducerAndConsumer<KeyValue,
KeyValueBatch>>
+ changelog_producer_and_consumer =
CreateSortMergeProducerAndConsumer(
+ std::move(raw_sort_merge_reader), write_batch_size,
create_consumer);
+ std::unique_ptr<AsyncKeyValueProducerAndConsumer<KeyValue,
KeyValueBatch>>
+ 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<RollingFileWriter<KeyValueBatch,
std::shared_ptr<DataFileMeta>>>
+ 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<RollingFileWriter<KeyValueBatch,
std::shared_ptr<DataFileMeta>>>
+ data_writer;
+ PAIMON_ASSIGN_OR_RAISE(data_writer, CreateRollingRowWriter());
+ ScopeGuard data_writer_guard([&]() -> void { data_writer->Abort();
});
+
+ std::vector<std::shared_ptr<DataFileMeta>> data_files;
+ std::vector<std::shared_ptr<DataFileMeta>> changelog_files;
+ if (options_.InputChangelogParallelWriteEnabled()) {
+
std::future<Result<std::vector<std::shared_ptr<DataFileMeta>>>>
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<std::vector<std::shared_ptr<DataFileMeta>>> data_result
=
+
WriteKeyValueBatchesToFiles(data_producer_and_consumer.get(),
+ data_writer.get());
+ Result<std::vector<std::shared_ptr<DataFileMeta>>>
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<std::unique_ptr<KeyValueRecordReader>>
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<DataFileMeta>& 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<std::unique_ptr<KeyValueRecordReader>>
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<CommitIncrement> MergeTreeWriter::DrainIncrement()
{
Result<std::unique_ptr<RollingFileWriter<KeyValueBatch,
std::shared_ptr<DataFileMeta>>>>
MergeTreeWriter::CreateRollingRowWriter() const {
- PAIMON_ASSIGN_OR_RAISE(
- std::shared_ptr<KeyValueDataFileWriterFactories::WriterFactory>
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<RollingFileWriter<KeyValueBatch,
std::shared_ptr<DataFileMeta>>>(
- options_.GetTargetFileSize(/*has_primary_key=*/true),
options_.GetTargetFileRowNum(),
- factory);
+ return CreateRollingWriter(/*is_changelog=*/false,
options_.GetTargetFileRowNum());
}
Result<std::unique_ptr<RollingFileWriter<KeyValueBatch,
std::shared_ptr<DataFileMeta>>>>
MergeTreeWriter::CreateRollingChangelogWriter() const {
- PAIMON_ASSIGN_OR_RAISE(
- std::shared_ptr<KeyValueDataFileWriterFactories::WriterFactory>
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<int64_t>::max());
+}
+
+Result<std::unique_ptr<RollingFileWriter<KeyValueBatch,
std::shared_ptr<DataFileMeta>>>>
+MergeTreeWriter::CreateRollingWriter(bool is_changelog, int64_t
target_file_row_num) const {
+
PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<KeyValueDataFileWriterFactories::WriterFactory>
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<RollingFileWriter<KeyValueBatch,
std::shared_ptr<DataFileMeta>>>(
- options_.GetTargetFileSize(/*has_primary_key=*/true),
- /*target_file_row_num=*/std::numeric_limits<int64_t>::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 9f73e43b..1a2c2c3d 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<std::unique_ptr<RollingFileWriter<KeyValueBatch,
std::shared_ptr<DataFileMeta>>>>
CreateRollingChangelogWriter() const;
+ Result<std::unique_ptr<RollingFileWriter<KeyValueBatch,
std::shared_ptr<DataFileMeta>>>>
+ CreateRollingWriter(bool is_changelog, int64_t target_file_row_num) const;
+
Status TrySyncLatestCompaction(bool blocking);
Status UpdateCompactResult(const std::shared_ptr<CompactResult>&
compact_result);
Status UpdateCompactDeletionFile(const
std::shared_ptr<CompactDeletionFile>& 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 8a152e2f..e557af32 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<DataFilePathFactory>();
+ 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<arrow::Array> 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"},
