lxy-9602 commented on code in PR #248:
URL: https://github.com/apache/paimon-cpp/pull/248#discussion_r3867875338
##########
src/paimon/core/io/async_key_value_producer_and_consumer.h:
##########
@@ -48,24 +79,39 @@ class AsyncKeyValueProducerAndConsumer {
using ConsumerCreator =
std::function<Result<std::unique_ptr<RowToArrowArrayConverter<T,
R>>>()>;
+ // Limits the number of rows sent to one Arrow projection call.
+ static int32_t NormalizeProjectionBatchSize(int32_t batch_size) {
+ return std::min(batch_size, MAX_PROJECTION_BATCH_SIZE);
+ }
+
AsyncKeyValueProducerAndConsumer(std::unique_ptr<SortMergeReader>&&
sort_merge_reader,
ConsumerCreator create_consumer, int32_t
batch_size,
int32_t consumer_thread_num,
const std::shared_ptr<MemoryPool>& pool);
+ /// Creates a single-conversion-thread pipeline whose producer emits
tagged data and changelog
+ /// row batches. Converted batches retain the producer order and are
returned with their tag.
+
AsyncKeyValueProducerAndConsumer(std::unique_ptr<AsyncKeyValueBatchProducer>&&
producer,
+ ConsumerCreator data_consumer_creator,
+ ConsumerCreator
changelog_consumer_creator);
+
~AsyncKeyValueProducerAndConsumer() {
CleanUp();
}
Result<R> NextBatch();
+ Result<AsyncKeyValueResultBatch<R>> NextBatchWithType();
+
std::shared_ptr<Metrics> GetReaderMetrics() const {
- return sort_merge_reader_->GetReaderMetrics();
+ return sort_merge_reader_ ? sort_merge_reader_->GetReaderMetrics() :
nullptr;
}
Review Comment:
Otherwise, could we create an empty metrics object here instead? It’d be
better to avoid returning nullptr if possible.
##########
src/paimon/core/mergetree/compact/first_row_merge_function_wrapper.h:
##########
@@ -62,17 +69,26 @@ class FirstRowMergeFunctionWrapper : public
MergeFunctionWrapper<KeyValue> {
if (contains) {
// empty
Reset();
- return std::optional<KeyValue>();
+ return std::optional<ChangelogResult>(std::move(changelog_result));
}
- // new record, output changelog
- // TODO(xinyu.lxy) support changelog
+ if (value_serializer_) {
+ PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<Bytes> bytes,
+
value_serializer_->SerializeToBytes(*result->value));
+ PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<InternalRow>
changelog_value,
+ value_serializer_->Deserialize(bytes));
+ changelog_result.changelogs.emplace_back(result->value_kind,
result->sequence_number,
+ result->level,
result->key,
+
std::move(changelog_value));
+ }
Review Comment:
Could we add a TODO here? There seems to be a clear hotspot at the moment,
and we’re still doing row-by-row data copying.
##########
src/paimon/core/io/async_key_value_producer_and_consumer.cpp:
##########
@@ -227,7 +292,12 @@ void AsyncKeyValueConsumer<T, R>::CleanUp() {
if (consumer_future_.valid()) {
[[maybe_unused]] Status status = consumer_future_.get();
}
- key_value_consumer_->CleanUp();
+ if (key_value_consumer_) {
+ key_value_consumer_->CleanUp();
+ }
+ if (changelog_consumer_) {
+ changelog_consumer_->CleanUp();
+ }
Review Comment:
This overload introduces two mutually exclusive internal modes: one uses
`sort_merge_reader_` and `batch_size_`, while the other uses `producer_` and
leaves the former members inactive. As a result, `ProduceLoop`,
`GetReaderMetrics`, and `Close` all need to branch on how the object was
constructed. Could we wrap `SortMergeReader` in an `AsyncKeyValueBatchProducer`
and keep a single producer abstraction? The reader-specific batch size,
metrics, and close behavior could then live in that adapter.
##########
src/paimon/core/mergetree/compact/internal_row_equalizer.h:
##########
@@ -0,0 +1,245 @@
+/*
+ * Copyright 2026-present Alibaba Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#pragma once
+
+#include <functional>
+#include <memory>
+#include <set>
+#include <string>
+#include <utility>
+#include <vector>
+
+#include "arrow/api.h"
+#include "fmt/format.h"
+#include "paimon/common/data/data_getters.h"
+#include "paimon/common/data/internal_array.h"
+#include "paimon/common/data/internal_map.h"
+#include "paimon/common/data/internal_row.h"
+#include "paimon/common/utils/checked_cast.h"
+#include "paimon/common/utils/date_time_utils.h"
+#include "paimon/common/utils/fields_comparator.h"
+#include "paimon/result.h"
+#include "paimon/status.h"
+
+namespace paimon {
+
+/// Creates equality functions for internal rows, including nested values.
+class InternalRowEqualizer {
+ public:
+ using Equalizer = std::function<bool(const InternalRow&, const
InternalRow&)>;
+
Review Comment:
`std::function<bool(const InternalRow&, const InternalRow&)>` is actually
already defined as `FieldComparatorFunc` in `fields_comparator.h`. It looks
like we’ve redefined it in several places now.
##########
src/paimon/core/mergetree/compact/internal_row_equalizer.h:
##########
@@ -0,0 +1,245 @@
+/*
+ * Copyright 2026-present Alibaba Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#pragma once
+
+#include <functional>
+#include <memory>
+#include <set>
+#include <string>
+#include <utility>
+#include <vector>
+
+#include "arrow/api.h"
+#include "fmt/format.h"
+#include "paimon/common/data/data_getters.h"
+#include "paimon/common/data/internal_array.h"
+#include "paimon/common/data/internal_map.h"
+#include "paimon/common/data/internal_row.h"
+#include "paimon/common/utils/checked_cast.h"
+#include "paimon/common/utils/date_time_utils.h"
+#include "paimon/common/utils/fields_comparator.h"
+#include "paimon/result.h"
+#include "paimon/status.h"
+
+namespace paimon {
+
+/// Creates equality functions for internal rows, including nested values.
+class InternalRowEqualizer {
+ public:
+ using Equalizer = std::function<bool(const InternalRow&, const
InternalRow&)>;
+
+ static Result<Equalizer> Create(const std::shared_ptr<arrow::Schema>&
schema,
+ const std::vector<std::string>&
ignore_fields) {
+ std::set<std::string> ignored(ignore_fields.begin(),
ignore_fields.end());
+ std::vector<std::pair<int32_t, ValueEqualizer>> equalizers;
+ for (int32_t i = 0; i < schema->num_fields(); ++i) {
+ if (ignored.find(schema->field(i)->name()) != ignored.end()) {
+ continue;
+ }
+ PAIMON_ASSIGN_OR_RAISE(ValueEqualizer equalizer,
+
CreateValueEqualizer(schema->field(i)->type()));
+ equalizers.emplace_back(i, std::move(equalizer));
+ }
+ return Equalizer(
+ [equalizers = std::move(equalizers)](const InternalRow& lhs, const
InternalRow& rhs) {
+ for (const auto& [field_idx, equalizer] : equalizers) {
+ if (!EqualAt(lhs, field_idx, rhs, field_idx, equalizer)) {
+ return false;
+ }
+ }
+ return true;
+ });
+ }
+
+ private:
+ using ValueEqualizer =
+ std::function<bool(const DataGetters&, int32_t, const DataGetters&,
int32_t)>;
+
+ static bool EqualAt(const DataGetters& lhs, int32_t lhs_pos, const
DataGetters& rhs,
+ int32_t rhs_pos, const ValueEqualizer& equalizer) {
Review Comment:
The Java side also compares `rowkind`. It may not affect the current path,
but could we add a comment to clarify that for now?
##########
src/paimon/core/mergetree/compact/merge_tree_compact_rewriter.cpp:
##########
@@ -146,11 +146,40 @@ MergeTreeCompactRewriter::CreateRollingRowWriter(int32_t
level) {
factory = std::make_shared<ShreddingKeyValueDataFileWriterFactory>(
options_, schema_id_, write_schema_, level, FileSource::Compact(),
trimmed_primary_keys_, data_file_path_factory,
/*create_stats_extractor=*/true,
- plan_factory, pool_);
+ plan_factory, /*is_changelog=*/false, pool_);
} else {
factory = std::make_shared<KeyValueDataFileWriterFactory>(
options_, schema_id_, write_schema_, level, FileSource::Compact(),
- trimmed_primary_keys_, data_file_path_factory,
/*create_stats_extractor=*/true, pool_);
+ trimmed_primary_keys_, data_file_path_factory,
/*create_stats_extractor=*/true,
+ /*is_changelog=*/false, pool_);
+ }
+ return
std::make_unique<MergeTreeCompactRewriter::KeyValueRollingFileWriter>(
+ options_.GetTargetFileSize(/*has_primary_key=*/true),
+ /*target_file_row_num=*/std::numeric_limits<int64_t>::max(), factory);
+}
+
+Result<std::unique_ptr<MergeTreeCompactRewriter::KeyValueRollingFileWriter>>
+MergeTreeCompactRewriter::CreateRollingChangelogWriter(int32_t level) {
+ std::shared_ptr<FileFormat> format = options_.GetChangelogFileFormat();
Review Comment:
`CreateRollingRowWriter` and `CreateRollingChangelogWriter` duplicate the
path-factory setup, shredding-plan selection, data-file-writer factory
construction, and rolling-writer construction. The same shredding-vs-normal
factory selection is also repeated in `MergeTreeWriter` and
`PostponeBucketWriter`. Could we extract a shared
`CreateKeyValueDataFileWriterFactory(...)` helper, while keeping format/path,
FileSource, and rolling row-limit decisions at the callers? At minimum, these
two methods could delegate to a private `CreateRollingWriter(level,
is_changelog)` helper.
##########
src/paimon/core/mergetree/compact/lookup_merge_tree_compact_rewriter.h:
##########
@@ -50,16 +53,19 @@ class LookupMergeTreeCompactRewriter : public
ChangelogMergeTreeRewriter {
return lookup_levels_->Close();
}
- static std::shared_ptr<MergeFunctionWrapper<KeyValue>>
CreateFirstRowMergeFunctionWrapper(
- std::unique_ptr<FirstRowMergeFunction>&& merge_func, int32_t
output_level,
- LookupLevels<bool>* lookup_levels);
+ static std::shared_ptr<MergeFunctionWrapper<ChangelogResult>>
+
CreateFirstRowMergeFunctionWrapper(std::unique_ptr<FirstRowMergeFunction>&&
merge_func,
+ int32_t output_level,
LookupLevels<bool>* lookup_levels,
+
std::unique_ptr<RowCompactedSerializer>&& value_serializer);
Review Comment:
How about moving `value_serializer` before `lookup_levels` instead?
##########
src/paimon/core/mergetree/compact/merge_tree_compact_manager_factory.cpp:
##########
@@ -241,29 +270,37 @@
MergeTreeCompactManagerFactory::CreateLookupRewriterWithDeletionVector(
lookup_file_cache_, remote_lookup_file_manager, pool_));
auto merge_function_wrapper_factory =
[data_schema = schema_, options = options_, trimmed_primary_keys,
- lookup_levels_ptr = lookup_levels.get(), lookup_strategy,
- dv_maintainer_ptr = dv_maintainer, pool = pool_,
- user_defined_seq_comparator = user_defined_seq_comparator_](
- int32_t output_level) ->
Result<std::shared_ptr<MergeFunctionWrapper<KeyValue>>> {
+ lookup_levels_ptr = lookup_levels.get(), lookup_strategy,
should_produce_changelog,
+ dv_maintainer_ptr = dv_maintainer,
+ user_defined_seq_comparator = user_defined_seq_comparator_,
+ pool = pool_](int32_t output_level)
+ -> Result<std::shared_ptr<MergeFunctionWrapper<ChangelogResult>>> {
PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<MergeFunction> merge_func,
PrimaryKeyTableUtils::CreateMergeFunction(
data_schema, trimmed_primary_keys,
options, pool));
PAIMON_ASSIGN_OR_RAISE(
- std::shared_ptr<MergeFunctionWrapper<KeyValue>>
merge_function_wrapper,
+ std::unique_ptr<RowCompactedSerializer> value_serializer,
+ CreateChangelogValueSerializer(data_schema,
should_produce_changelog, pool));
+ PAIMON_ASSIGN_OR_RAISE(
+ InternalRowEqualizer::Equalizer value_equalizer,
+ CreateChangelogValueEqualizer(data_schema, options,
should_produce_changelog));
+ PAIMON_ASSIGN_OR_RAISE(
+ std::shared_ptr<MergeFunctionWrapper<ChangelogResult>>
merge_function_wrapper,
Review Comment:
Could we refactor this a bit? It looks like very similar logic has been
implemented multiple times now.
##########
src/paimon/core/mergetree/merge_tree_writer.cpp:
##########
@@ -329,15 +390,38 @@ MergeTreeWriter::CreateRollingRowWriter() const {
factory = std::make_shared<ShreddingKeyValueDataFileWriterFactory>(
options_, schema_id_, write_schema_, /*level=*/0,
FileSource::Append(),
trimmed_primary_keys_, path_factory_,
/*create_stats_extractor=*/true, plan_factory,
- pool_);
+ /*is_changelog=*/false, pool_);
} else {
factory = std::make_shared<KeyValueDataFileWriterFactory>(
options_, schema_id_, write_schema_, /*level=*/0,
FileSource::Append(),
- trimmed_primary_keys_, path_factory_,
/*create_stats_extractor=*/true, pool_);
+ 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);
}
+Result<std::unique_ptr<RollingFileWriter<KeyValueBatch,
std::shared_ptr<DataFileMeta>>>>
+MergeTreeWriter::CreateRollingChangelogWriter() const {
+ std::shared_ptr<SingleFileWriterFactory<KeyValueBatch,
std::shared_ptr<DataFileMeta>>> factory;
+ PAIMON_ASSIGN_OR_RAISE(
Review Comment:
This is very similar to `CreateRollingRowWriter`. Could we extract the
shared parts?
##########
src/paimon/core/mergetree/compact/changelog_merge_tree_rewriter.cpp:
##########
@@ -17,7 +17,152 @@
*/
#include "paimon/core/mergetree/compact/changelog_merge_tree_rewriter.h"
+
+#include <utility>
+
+#include "paimon/common/utils/scope_guard.h"
+#include "paimon/core/io/key_value_meta_projection_consumer.h"
+#include "paimon/core/io/row_to_arrow_array_converter.h"
+#include "paimon/format/file_format.h"
namespace paimon {
+
+namespace {
+
+using SortMergeReaderFactory =
+ std::function<Result<std::unique_ptr<SortMergeReader>>(const
std::vector<SortedRun>& section)>;
+using BoundChangelogMergeFunctionWrapperFactory =
+
std::function<Result<std::shared_ptr<MergeFunctionWrapper<ChangelogResult>>>()>;
+using KeyComparator = std::function<int32_t(const InternalRow&, const
InternalRow&)>;
+using CancellationChecker = std::function<bool()>;
+
+class ChangelogCompactionBatchProducer : public AsyncKeyValueBatchProducer {
+ public:
+ ChangelogCompactionBatchProducer(
+ const std::vector<std::vector<SortedRun>>& sections,
+ std::vector<std::unique_ptr<SortMergeReader>>& reader_holders, int32_t
write_batch_size,
+ SortMergeReaderFactory reader_factory,
Review Comment:
`std::vector<std::unique_ptr<SortMergeReader>>& reader_holders`? Use
`std::vector &&` or just `std::vector`
##########
src/paimon/core/mergetree/compact/merge_tree_compact_manager_factory_test.cpp:
##########
@@ -377,13 +379,12 @@ TEST_F(MergeTreeCompactManagerFactoryWriteTest,
}
TEST_F(MergeTreeCompactManagerFactoryWriteTest,
- TestCreateFileStoreWriteShouldFailWhenLookupChangelogConfigured) {
- ASSERT_NOK_WITH_MSG(
- CreateSingleStringFileStoreWrite({{"bucket", "1"},
- {Options::DELETION_VECTORS_ENABLED,
"true"},
- {Options::CHANGELOG_PRODUCER,
"lookup"}},
- /*with_io_manager=*/true),
- "C++ Paimon does not support changelog-producer yet");
+ TestCreateFileStoreWriteShouldSucceedWhenLookupChangelogConfigured) {
+ ASSERT_OK(CreateSingleStringFileStoreWrite({{"bucket", "1"},
+
{Options::DELETION_VECTORS_ENABLED, "true"},
+ {Options::CHANGELOG_PRODUCER,
"lookup"}},
+ /*with_io_manager=*/true)
+ .status());
Review Comment:
ASSERT_OK(Func())
##########
src/paimon/core/mergetree/compact/lookup_changelog_merge_function_wrapper.h:
##########
@@ -105,30 +119,80 @@ class LookupChangelogMergeFunctionWrapper : public
MergeFunctionWrapper<KeyValue
}
}
if (lookup_high_level) {
+ if (contain_level0 && should_produce_changelog_) {
+ PAIMON_ASSIGN_OR_RAISE(before,
+
CloneKeyValue(lookup_high_level.value(),
+
lookup_high_level.value().value_kind));
+ }
merge_function_->InsertInto(std::move(lookup_high_level),
comparator_);
}
}
// 3. Calculate result
PAIMON_ASSIGN_OR_RAISE(std::optional<KeyValue> result,
merge_function_->GetResult());
- Reset();
+
// 4. Set changelog when there's level-0 records
- // TODO(liancheng.lsz): setChangelog
- return result;
+ ChangelogResult changelog_result;
+ if (contain_level0 && should_produce_changelog_) {
+ PAIMON_RETURN_NOT_OK(
+ SetChangelog(std::move(before), result,
&changelog_result.changelogs));
+ }
+ changelog_result.result = std::move(result);
+ Reset();
+ return std::optional<ChangelogResult>(std::move(changelog_result));
}
private:
LookupChangelogMergeFunctionWrapper(
std::unique_ptr<LookupMergeFunction>&& merge_function,
std::function<Result<std::optional<T>>(const
std::shared_ptr<InternalRow>&)> lookup,
- const LookupStrategy& lookup_strategy,
+ const LookupStrategy& lookup_strategy, bool should_produce_changelog,
const std::shared_ptr<BucketedDvMaintainer>&
deletion_vectors_maintainer,
- const std::shared_ptr<FieldsComparator>& user_defined_seq_comparator)
+ const std::shared_ptr<FieldsComparator>& user_defined_seq_comparator,
+ std::unique_ptr<RowCompactedSerializer>&& value_serializer,
ValueEqualizer value_equalizer)
: merge_function_(std::move(merge_function)),
lookup_(std::move(lookup)),
lookup_strategy_(lookup_strategy),
+ should_produce_changelog_(should_produce_changelog),
deletion_vectors_maintainer_(deletion_vectors_maintainer),
- comparator_(CreateSequenceComparator(user_defined_seq_comparator)) {}
+ comparator_(CreateSequenceComparator(user_defined_seq_comparator)),
+ value_serializer_(std::move(value_serializer)),
+ value_equalizer_(std::move(value_equalizer)) {}
+
+ Result<KeyValue> CloneKeyValue(const KeyValue& from, const RowKind*
value_kind) {
+ PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<Bytes> bytes,
+
value_serializer_->SerializeToBytes(*from.value));
+ PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<InternalRow> value,
+ value_serializer_->Deserialize(bytes));
Review Comment:
Similarly, could we add a comment here to call out the copy overhead and a
TODO for future optimization?
##########
src/paimon/core/mergetree/compact/lookup_changelog_merge_function_wrapper.h:
##########
@@ -105,30 +119,80 @@ class LookupChangelogMergeFunctionWrapper : public
MergeFunctionWrapper<KeyValue
}
}
if (lookup_high_level) {
+ if (contain_level0 && should_produce_changelog_) {
+ PAIMON_ASSIGN_OR_RAISE(before,
+
CloneKeyValue(lookup_high_level.value(),
+
lookup_high_level.value().value_kind));
+ }
merge_function_->InsertInto(std::move(lookup_high_level),
comparator_);
}
}
// 3. Calculate result
PAIMON_ASSIGN_OR_RAISE(std::optional<KeyValue> result,
merge_function_->GetResult());
- Reset();
+
// 4. Set changelog when there's level-0 records
- // TODO(liancheng.lsz): setChangelog
- return result;
+ ChangelogResult changelog_result;
+ if (contain_level0 && should_produce_changelog_) {
+ PAIMON_RETURN_NOT_OK(
+ SetChangelog(std::move(before), result,
&changelog_result.changelogs));
+ }
+ changelog_result.result = std::move(result);
+ Reset();
+ return std::optional<ChangelogResult>(std::move(changelog_result));
}
private:
LookupChangelogMergeFunctionWrapper(
std::unique_ptr<LookupMergeFunction>&& merge_function,
std::function<Result<std::optional<T>>(const
std::shared_ptr<InternalRow>&)> lookup,
- const LookupStrategy& lookup_strategy,
+ const LookupStrategy& lookup_strategy, bool should_produce_changelog,
const std::shared_ptr<BucketedDvMaintainer>&
deletion_vectors_maintainer,
- const std::shared_ptr<FieldsComparator>& user_defined_seq_comparator)
+ const std::shared_ptr<FieldsComparator>& user_defined_seq_comparator,
+ std::unique_ptr<RowCompactedSerializer>&& value_serializer,
ValueEqualizer value_equalizer)
: merge_function_(std::move(merge_function)),
lookup_(std::move(lookup)),
lookup_strategy_(lookup_strategy),
+ should_produce_changelog_(should_produce_changelog),
deletion_vectors_maintainer_(deletion_vectors_maintainer),
- comparator_(CreateSequenceComparator(user_defined_seq_comparator)) {}
+ comparator_(CreateSequenceComparator(user_defined_seq_comparator)),
+ value_serializer_(std::move(value_serializer)),
+ value_equalizer_(std::move(value_equalizer)) {}
+
+ Result<KeyValue> CloneKeyValue(const KeyValue& from, const RowKind*
value_kind) {
+ PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<Bytes> bytes,
+
value_serializer_->SerializeToBytes(*from.value));
+ PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<InternalRow> value,
+ value_serializer_->Deserialize(bytes));
+ return KeyValue(value_kind, from.sequence_number, from.level,
from.key, std::move(value));
Review Comment:
Java's `KeyValue.replace` resets the generated changelog record's level to
`UNKNOWN_LEVEL`, while `CloneKeyValue` preserves from.level. The current Arrow
converter does not serialize this field, so this may not affect files today,
but the intermediate KeyValue semantics differ from Java. Could we use
`KeyValue::UNKNOWN_LEVEL` here?
##########
src/paimon/core/mergetree/merge_tree_writer_test.cpp:
##########
@@ -295,6 +300,182 @@ TEST_P(MergeTreeWriterTest, TestSimple) {
ASSERT_EQ(expected_data_increment,
commit_increment.GetNewFilesIncrement());
}
+TEST_P(MergeTreeWriterTest, TestInputChangelog) {
+ ASSERT_OK_AND_ASSIGN(CoreOptions options,
+ CoreOptions::FromMap({{Options::FILE_FORMAT, "orc"},
+ {Options::CHANGELOG_PRODUCER,
"input"},
+
{Options::CHANGELOG_FILE_PREFIX, "changes-"},
+
{Options::CHANGELOG_FILE_FORMAT, "parquet"}}));
+
+ 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,
+ {RecordBatch::RowKind::INSERT,
RecordBatch::RowKind::UPDATE_BEFORE,
+ RecordBatch::RowKind::UPDATE_AFTER,
RecordBatch::RowKind::DELETE},
+ 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("data-" + uuid + "-1.orc",
data_increment.NewFiles()[0]->file_name);
+ ASSERT_EQ("changes-" + uuid + "-0.parquet",
data_increment.ChangelogFiles()[0]->file_name);
+ ASSERT_EQ(2, data_increment.NewFiles()[0]->row_count);
+ ASSERT_EQ(4, data_increment.ChangelogFiles()[0]->row_count);
+
+ std::shared_ptr<arrow::ChunkedArray> expected_data;
+ auto data_status =
arrow::ipc::internal::json::ChunkedArrayFromJSON(write_type_, {R"([
+ [2, 2, "Alice", 11, 0, 11.0],
+ [3, 3, "Bob", 21, 0, 21.0]
+ ])"},
+
&expected_data);
+ ASSERT_TRUE(data_status.ok());
+ CheckFileContent(dir->Str() + "/" +
data_increment.NewFiles()[0]->file_name, expected_data);
+
+ std::shared_ptr<arrow::ChunkedArray> expected_changelog;
+ auto changelog_status =
arrow::ipc::internal::json::ChunkedArrayFromJSON(write_type_, {R"([
+ [1, 1, "Alice", 10, 0, 10.0],
+ [2, 2, "Alice", 11, 0, 11.0],
+ [0, 0, "Bob", 20, 0, 20.0],
+ [3, 3, "Bob", 21, 0, 21.0]
+ ])"},
+
&expected_changelog);
+ ASSERT_TRUE(changelog_status.ok());
+ CheckFileContent(dir->Str() + "/" +
data_increment.ChangelogFiles()[0]->file_name,
+ expected_changelog, "parquet");
+}
+
+TEST_P(MergeTreeWriterTest, TestInputChangelogIgnoresTargetFileRowNum) {
+ ASSERT_OK_AND_ASSIGN(CoreOptions options,
+ CoreOptions::FromMap({{Options::FILE_FORMAT, "orc"},
+ {Options::CHANGELOG_PRODUCER,
"input"},
+ {Options::TARGET_FILE_ROW_NUM,
"1"},
+ {Options::WRITE_BATCH_SIZE,
"1"}}));
+
Review Comment:
I’m a bit curious: since `TargetFileRowNum` is already defined at batch
granularity, if we only write a single batch, wouldn’t it end up as one file
regardless of whether `ignore` is enabled?
##########
src/paimon/core/io/async_key_value_producer_and_consumer.cpp:
##########
@@ -227,7 +292,12 @@ void AsyncKeyValueConsumer<T, R>::CleanUp() {
if (consumer_future_.valid()) {
[[maybe_unused]] Status status = consumer_future_.get();
}
- key_value_consumer_->CleanUp();
+ if (key_value_consumer_) {
+ key_value_consumer_->CleanUp();
+ }
+ if (changelog_consumer_) {
+ changelog_consumer_->CleanUp();
+ }
Review Comment:
Could we simplify this pipeline by keeping the responsibilities separated as
follows?
- The producer decides which output types are enabled and tags each emitted
batch as DATA or CHANGELOG.
- The consumer only converts `KeyValue` rows into `KeyValueBatch` and
preserves the tag in the converted result.
- `ChangelogMergeTreeRewriter` uses the preserved tag to route the converted
batch to either the compact data writer or the changelog writer.
DATA and CHANGELOG currently use the same `KeyValue` input type, write
schema, and `create_consumer` factory, so maintaining separate
`key_value_consumer_` and `changelog_consumer_` instances and selecting between
them here does not appear necessary. Could we use a single converter and keep
the DATA/CHANGELOG distinction only for downstream writer routing? This would
also remove the nullable consumer state and the need to synchronize producer
output flags with consumer availability.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
