This is an automated email from the ASF dual-hosted git repository.
JingsongLi 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 2407cf1 feat: Migrate pk compact rewriters (#97)
2407cf1 is described below
commit 2407cf1e8d57a6f57ff24bb3cf3dc59d8c0e7e2f
Author: lxy <[email protected]>
AuthorDate: Mon Jun 22 17:27:57 2026 +0800
feat: Migrate pk compact rewriters (#97)
---
.../compact/changelog_merge_tree_rewriter.cpp | 125 ++++++
.../compact/changelog_merge_tree_rewriter.h | 89 ++++
.../mergetree/compact/early_full_compaction.cpp | 87 ++++
.../core/mergetree/compact/early_full_compaction.h | 50 +++
.../compact/early_full_compaction_test.cpp | 267 ++++++++++++
.../mergetree/compact/force_up_level0_compaction.h | 72 ++++
.../compact/force_up_level0_compaction_test.cpp | 117 ++++++
.../compact/merge_tree_compact_rewriter.cpp | 291 +++++++++++++
.../compact/merge_tree_compact_rewriter.h | 126 ++++++
.../compact/merge_tree_compact_rewriter_test.cpp | 331 +++++++++++++++
.../mergetree/compact/universal_compaction.cpp | 185 ++++++++
.../core/mergetree/compact/universal_compaction.h | 66 +++
.../compact/universal_compaction_test.cpp | 464 +++++++++++++++++++++
13 files changed, 2270 insertions(+)
diff --git
a/src/paimon/core/mergetree/compact/changelog_merge_tree_rewriter.cpp
b/src/paimon/core/mergetree/compact/changelog_merge_tree_rewriter.cpp
new file mode 100644
index 0000000..ecd2af7
--- /dev/null
+++ b/src/paimon/core/mergetree/compact/changelog_merge_tree_rewriter.cpp
@@ -0,0 +1,125 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.
+ */
+
+#include "paimon/core/mergetree/compact/changelog_merge_tree_rewriter.h"
+namespace paimon {
+ChangelogMergeTreeRewriter::ChangelogMergeTreeRewriter(
+ int32_t max_level, bool force_drop_delete, const BinaryRow& partition,
int32_t bucket,
+ int64_t schema_id, const std::vector<std::string>& trimmed_primary_keys,
+ const CoreOptions& options, const std::shared_ptr<arrow::Schema>&
data_schema,
+ const std::shared_ptr<arrow::Schema>& write_schema,
DeletionVector::Factory dv_factory,
+ const std::shared_ptr<FileStorePathFactoryCache>& path_factory_cache,
+ std::unique_ptr<MergeFileSplitRead>&& merge_file_split_read,
+ MergeFunctionWrapperFactory merge_function_wrapper_factory,
+ const std::shared_ptr<CancellationController>& cancellation_controller,
+ const std::shared_ptr<MemoryPool>& pool)
+ : MergeTreeCompactRewriter(
+ partition, bucket, schema_id, trimmed_primary_keys, options,
data_schema, write_schema,
+ std::move(dv_factory), path_factory_cache,
std::move(merge_file_split_read),
+ std::move(merge_function_wrapper_factory), cancellation_controller,
pool),
+ max_level_(max_level),
+ force_drop_delete_(force_drop_delete) {}
+
+Result<CompactResult> ChangelogMergeTreeRewriter::Rewrite(
+ int32_t output_level, bool drop_delete, const
std::vector<std::vector<SortedRun>>& sections) {
+ if (RewriteChangelog(output_level, drop_delete, sections)) {
+ return RewriteOrProduceChangelog(output_level, sections, drop_delete,
+ /*rewrite_compact_file=*/true);
+ } else {
+ return RewriteCompaction(output_level, drop_delete, sections);
+ }
+}
+
+Result<CompactResult> ChangelogMergeTreeRewriter::Upgrade(
+ int32_t output_level, const std::shared_ptr<DataFileMeta>& file) {
+ UpgradeStrategy upgrade_strategy = GenerateUpgradeStrategy(output_level,
file);
+ if (upgrade_strategy.changelog) {
+ return RewriteOrProduceChangelog(output_level,
{{SortedRun::FromSingle(file)}},
+ force_drop_delete_,
upgrade_strategy.rewrite);
+ } else {
+ return MergeTreeCompactRewriter::Upgrade(output_level, file);
+ }
+}
+
+bool ChangelogMergeTreeRewriter::RewriteLookupChangelog(
+ int32_t output_level, const std::vector<std::vector<SortedRun>>& sections)
const {
+ if (output_level == 0) {
+ return false;
+ }
+ for (const auto& runs : sections) {
+ for (const auto& run : runs) {
+ for (const auto& file : run.Files()) {
+ if (file->level == 0) {
+ return true;
+ }
+ }
+ }
+ }
+ return false;
+}
+
+Result<CompactResult> ChangelogMergeTreeRewriter::RewriteOrProduceChangelog(
+ int32_t output_level, const std::vector<std::vector<SortedRun>>& sections,
bool drop_delete,
+ bool rewrite_compact_file) {
+ PAIMON_ASSIGN_OR_RAISE(MergeTreeCompactRewriter::KeyValueConsumerCreator
create_consumer,
+ GenerateKeyValueConsumer());
+
std::vector<std::shared_ptr<MergeTreeCompactRewriter::KeyValueMergeReader>>
reader_holders;
+
+ std::unique_ptr<MergeTreeCompactRewriter::KeyValueRollingFileWriter>
compact_file_writer;
+ if (rewrite_compact_file) {
+ compact_file_writer = CreateRollingRowWriter(output_level);
+ }
+ // TODO(xinyu.lxy): produce changelog
+ ScopeGuard write_guard([&]() -> void {
+ if (compact_file_writer) {
+ compact_file_writer->Abort();
+ }
+ merge_file_split_read_.reset();
+ for (const auto& reader : reader_holders) {
+ reader->Close();
+ }
+ });
+
+ for (const auto& section : sections) {
+ PAIMON_RETURN_NOT_OK(MergeReadAndWrite(output_level, drop_delete,
section, create_consumer,
+ compact_file_writer.get(),
&reader_holders));
+ }
+ if (compact_file_writer) {
+ PAIMON_RETURN_NOT_OK(compact_file_writer->Close());
+ }
+ auto before = ExtractFilesFromSections(sections);
+ std::vector<std::shared_ptr<DataFileMeta>> after;
+ if (compact_file_writer) {
+ PAIMON_ASSIGN_OR_RAISE(after, compact_file_writer->GetResult());
+ } else {
+ after.reserve(before.size());
+ for (const auto& file : before) {
+ PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<DataFileMeta> new_file,
+ file->Upgrade(output_level));
+ after.emplace_back(std::move(new_file));
+ }
+ }
+ if (rewrite_compact_file) {
+ NotifyRewriteCompactBefore(before);
+ }
+ PAIMON_ASSIGN_OR_RAISE(after, NotifyRewriteCompactAfter(after));
+ write_guard.Release();
+ return CompactResult(before, after);
+}
+
+} // namespace paimon
diff --git a/src/paimon/core/mergetree/compact/changelog_merge_tree_rewriter.h
b/src/paimon/core/mergetree/compact/changelog_merge_tree_rewriter.h
new file mode 100644
index 0000000..c5d8e89
--- /dev/null
+++ b/src/paimon/core/mergetree/compact/changelog_merge_tree_rewriter.h
@@ -0,0 +1,89 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 "arrow/api.h"
+#include "paimon/core/core_options.h"
+#include "paimon/core/io/data_file_meta.h"
+#include "paimon/core/key_value.h"
+#include "paimon/core/mergetree/compact/merge_tree_compact_rewriter.h"
+namespace paimon {
+/// A `MergeTreeCompactRewriter` which produces changelog files while
performing compaction.
+class ChangelogMergeTreeRewriter : public MergeTreeCompactRewriter {
+ public:
+ Result<CompactResult> Rewrite(int32_t output_level, bool drop_delete,
+ const std::vector<std::vector<SortedRun>>&
sections) override;
+
+ Result<CompactResult> Upgrade(int32_t output_level,
+ const std::shared_ptr<DataFileMeta>& file)
override;
+
+ protected:
+ ChangelogMergeTreeRewriter(
+ int32_t max_level, bool force_drop_delete, const BinaryRow& partition,
int32_t bucket,
+ int64_t schema_id, const std::vector<std::string>&
trimmed_primary_keys,
+ const CoreOptions& options, const std::shared_ptr<arrow::Schema>&
data_schema,
+ const std::shared_ptr<arrow::Schema>& write_schema,
DeletionVector::Factory dv_factory,
+ const std::shared_ptr<FileStorePathFactoryCache>& path_factory_cache,
+ std::unique_ptr<MergeFileSplitRead>&& merge_file_split_read,
+ MergeFunctionWrapperFactory merge_function_wrapper_factory,
+ const std::shared_ptr<CancellationController>& cancellation_controller,
+ const std::shared_ptr<MemoryPool>& pool);
+
+ struct UpgradeStrategy {
+ static UpgradeStrategy NoChangelogNoRewrite() {
+ static const UpgradeStrategy ret = {false, false};
+ return ret;
+ }
+ static UpgradeStrategy ChangelogNoRewrite() {
+ static const UpgradeStrategy ret = {true, false};
+ return ret;
+ }
+ static UpgradeStrategy ChangelogWithRewrite() {
+ static const UpgradeStrategy ret = {true, true};
+ return ret;
+ }
+
+ bool operator==(const UpgradeStrategy& other) const {
+ if (this == &other) {
+ return true;
+ }
+ return changelog == other.changelog && rewrite == other.rewrite;
+ }
+ bool changelog;
+ bool rewrite;
+ };
+
+ virtual UpgradeStrategy GenerateUpgradeStrategy(
+ int32_t output_level, const std::shared_ptr<DataFileMeta>& file) const
= 0;
+
+ virtual bool RewriteChangelog(int32_t output_level, bool drop_delete,
+ const std::vector<std::vector<SortedRun>>&
sections) const = 0;
+
+ bool RewriteLookupChangelog(int32_t output_level,
+ const std::vector<std::vector<SortedRun>>&
sections) const;
+
+ int32_t max_level_;
+ bool force_drop_delete_;
+
+ private:
+ Result<CompactResult> RewriteOrProduceChangelog(
+ int32_t output_level, const std::vector<std::vector<SortedRun>>&
sections, bool drop_delete,
+ bool rewrite_compact_file);
+};
+} // namespace paimon
diff --git a/src/paimon/core/mergetree/compact/early_full_compaction.cpp
b/src/paimon/core/mergetree/compact/early_full_compaction.cpp
new file mode 100644
index 0000000..4aa57a0
--- /dev/null
+++ b/src/paimon/core/mergetree/compact/early_full_compaction.cpp
@@ -0,0 +1,87 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.
+ */
+
+#include "paimon/core/mergetree/compact/early_full_compaction.h"
+
+#include "paimon/common/utils/date_time_utils.h"
+namespace paimon {
+std::shared_ptr<EarlyFullCompaction> EarlyFullCompaction::Create(const
CoreOptions& options) {
+ std::optional<int64_t> interval = options.GetOptimizedCompactionInterval();
+ std::optional<int64_t> total_size_threshold =
options.GetCompactionTotalSizeThreshold();
+ std::optional<int64_t> incremental_size_threshold =
+ options.GetCompactionIncrementalSizeThreshold();
+ if (!interval && !total_size_threshold && !incremental_size_threshold) {
+ return nullptr;
+ }
+ return std::shared_ptr<EarlyFullCompaction>(
+ new EarlyFullCompaction(interval, total_size_threshold,
incremental_size_threshold));
+}
+
+void EarlyFullCompaction::UpdateLastFullCompaction() {
+ last_full_compaction_ = CurrentTimeMillis();
+}
+
+int64_t EarlyFullCompaction::CurrentTimeMillis() const {
+ return DateTimeUtils::GetCurrentUTCTimeUs() /
+
DateTimeUtils::CONVERSION_FACTORS[DateTimeUtils::TimeType::MILLISECOND];
+}
+EarlyFullCompaction::EarlyFullCompaction(const std::optional<int64_t>&
full_compaction_interval,
+ const std::optional<int64_t>&
total_size_threshold,
+ const std::optional<int64_t>&
incremental_size_threshold)
+ : full_compaction_interval_(full_compaction_interval),
+ total_size_threshold_(total_size_threshold),
+ incremental_size_threshold_(incremental_size_threshold) {}
+
+std::optional<CompactUnit> EarlyFullCompaction::TryFullCompact(
+ int32_t num_levels, const std::vector<LevelSortedRun>& runs) {
+ if (runs.empty() || runs.size() == 1) {
+ return std::nullopt;
+ }
+ int32_t max_level = num_levels - 1;
+ if (full_compaction_interval_) {
+ if (!last_full_compaction_ || CurrentTimeMillis() -
last_full_compaction_.value() >
+ full_compaction_interval_.value()) {
+ UpdateLastFullCompaction();
+ return CompactUnit::FromLevelRuns(max_level, runs);
+ }
+ }
+ if (total_size_threshold_) {
+ int64_t total_size = 0;
+ for (const auto& run : runs) {
+ total_size += run.run.TotalSize();
+ }
+ if (total_size < total_size_threshold_.value()) {
+ UpdateLastFullCompaction();
+ return CompactUnit::FromLevelRuns(max_level, runs);
+ }
+ }
+ if (incremental_size_threshold_) {
+ int64_t incremental_size = 0;
+ for (const auto& run : runs) {
+ if (run.level != max_level) {
+ incremental_size += run.run.TotalSize();
+ }
+ }
+ if (incremental_size > incremental_size_threshold_.value()) {
+ UpdateLastFullCompaction();
+ return CompactUnit::FromLevelRuns(max_level, runs);
+ }
+ }
+ return std::nullopt;
+}
+} // namespace paimon
diff --git a/src/paimon/core/mergetree/compact/early_full_compaction.h
b/src/paimon/core/mergetree/compact/early_full_compaction.h
new file mode 100644
index 0000000..ed91e64
--- /dev/null
+++ b/src/paimon/core/mergetree/compact/early_full_compaction.h
@@ -0,0 +1,50 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 "paimon/core/compact/compact_unit.h"
+#include "paimon/core/core_options.h"
+namespace paimon {
+/// Early trigger full compaction.
+class EarlyFullCompaction {
+ public:
+ virtual ~EarlyFullCompaction() = default;
+ /// @return Pointer to `EarlyFullCompaction` if the options contain
EarlyFullCompaction
+ /// settings; otherwise, nullptr.
+ static std::shared_ptr<EarlyFullCompaction> Create(const CoreOptions&
options);
+
+ void UpdateLastFullCompaction();
+
+ std::optional<CompactUnit> TryFullCompact(int32_t num_levels,
+ const
std::vector<LevelSortedRun>& runs);
+
+ protected:
+ // virtual only for test
+ virtual int64_t CurrentTimeMillis() const;
+
+ EarlyFullCompaction(const std::optional<int64_t>& full_compaction_interval,
+ const std::optional<int64_t>& total_size_threshold,
+ const std::optional<int64_t>&
incremental_size_threshold);
+
+ private:
+ std::optional<int64_t> full_compaction_interval_;
+ std::optional<int64_t> total_size_threshold_;
+ std::optional<int64_t> incremental_size_threshold_;
+ std::optional<int64_t> last_full_compaction_;
+};
+} // namespace paimon
diff --git a/src/paimon/core/mergetree/compact/early_full_compaction_test.cpp
b/src/paimon/core/mergetree/compact/early_full_compaction_test.cpp
new file mode 100644
index 0000000..a4cc9c6
--- /dev/null
+++ b/src/paimon/core/mergetree/compact/early_full_compaction_test.cpp
@@ -0,0 +1,267 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.
+ */
+
+#include "paimon/core/mergetree/compact/early_full_compaction.h"
+
+#include "paimon/testing/utils/testharness.h"
+
+namespace paimon::test {
+class EarlyFullCompactionTest : public testing::Test {
+ public:
+ class TestableEarlyFullCompaction : public EarlyFullCompaction {
+ TestableEarlyFullCompaction(const std::optional<int64_t>&
full_compaction_interval,
+ const std::optional<int64_t>&
total_size_threshold,
+ const std::optional<int64_t>&
incremental_size_threshold,
+ const int64_t* current_time)
+ : EarlyFullCompaction(full_compaction_interval,
total_size_threshold,
+ incremental_size_threshold),
+ current_time_(current_time) {}
+
+ int64_t CurrentTimeMillis() const override {
+ return *current_time_;
+ }
+
+ private:
+ const int64_t* current_time_;
+ };
+
+ LevelSortedRun CreateLevelSortedRun(int32_t level, int64_t total_size)
const {
+ auto file_meta = std::make_shared<DataFileMeta>(
+ "fake.data", /*file_size=*/total_size, /*row_count=*/1,
+ /*min_key=*/BinaryRow::EmptyRow(), /*max_key=*/
+ BinaryRow::EmptyRow(),
+ /*key_stats=*/
+ SimpleStats::EmptyStats(),
+ /*value_stats=*/
+ SimpleStats::EmptyStats(),
+ /*min_sequence_number=*/0, /*max_sequence_number=*/6,
/*schema_id=*/0,
+ /*level=*/0,
/*extra_files=*/std::vector<std::optional<std::string>>(),
+ /*creation_time=*/Timestamp(0ll, 0),
+ /*delete_row_count=*/0, /*embedded_index=*/nullptr,
FileSource::Append(),
+ /*value_stats_cols=*/std::nullopt, /*external_path=*/std::nullopt,
+ /*first_row_id=*/std::nullopt,
+ /*write_cols=*/std::nullopt);
+ return {level, SortedRun::FromSingle(file_meta)};
+ }
+
+ std::vector<LevelSortedRun> CreateRuns(const std::vector<int64_t>& sizes)
const {
+ std::vector<LevelSortedRun> runs;
+ for (const auto& total_size : sizes) {
+ runs.push_back(CreateLevelSortedRun(/*level=*/0, total_size));
+ }
+ return runs;
+ }
+};
+
+TEST_F(EarlyFullCompactionTest, TestCreateNoOptions) {
+ std::map<std::string, std::string> options = {};
+ ASSERT_OK_AND_ASSIGN(CoreOptions core_options,
CoreOptions::FromMap(options));
+ ASSERT_FALSE(EarlyFullCompaction::Create(core_options));
+}
+
+TEST_F(EarlyFullCompactionTest, TestCreateWithInterval) {
+ std::map<std::string, std::string> options = {
+ {Options::COMPACTION_OPTIMIZATION_INTERVAL, "1h"},
+ };
+ ASSERT_OK_AND_ASSIGN(CoreOptions core_options,
CoreOptions::FromMap(options));
+ ASSERT_TRUE(EarlyFullCompaction::Create(core_options));
+}
+
+TEST_F(EarlyFullCompactionTest, TestCreateWithThreshold) {
+ std::map<std::string, std::string> options = {
+ {Options::COMPACTION_TOTAL_SIZE_THRESHOLD, "100MB"},
+ };
+ ASSERT_OK_AND_ASSIGN(CoreOptions core_options,
CoreOptions::FromMap(options));
+ ASSERT_TRUE(EarlyFullCompaction::Create(core_options));
+}
+
+TEST_F(EarlyFullCompactionTest, TestCreateWithBoth) {
+ std::map<std::string, std::string> options = {
+ {Options::COMPACTION_OPTIMIZATION_INTERVAL, "1h"},
+ {Options::COMPACTION_TOTAL_SIZE_THRESHOLD, "100MB"},
+ };
+ ASSERT_OK_AND_ASSIGN(CoreOptions core_options,
CoreOptions::FromMap(options));
+ ASSERT_TRUE(EarlyFullCompaction::Create(core_options));
+}
+
+TEST_F(EarlyFullCompactionTest, TestInterval) {
+ int64_t current_time = 10000l;
+ auto runs = CreateRuns({100l, 200l});
+ TestableEarlyFullCompaction
early_full_compaction(/*full_compaction_interval=*/1000l,
+
/*total_size_threshold=*/std::nullopt,
+
/*incremental_size_threshold=*/std::nullopt,
+ ¤t_time);
+ // First time, should trigger
+ auto compact_unit = early_full_compaction.TryFullCompact(/*num_levels=*/5,
runs);
+ ASSERT_TRUE(compact_unit);
+ ASSERT_EQ(compact_unit->output_level, 4);
+ ASSERT_EQ(compact_unit->files.size(), 2);
+
+ // Last compaction time is now 10000.
+ // Advance time, but not enough for interval to trigger.
+ current_time += 500;
+ ASSERT_FALSE(early_full_compaction.TryFullCompact(/*num_levels=*/5, runs));
+
+ // Advance time to be greater than interval.
+ current_time += 501;
+ compact_unit = early_full_compaction.TryFullCompact(/*num_levels=*/5,
runs);
+ ASSERT_TRUE(compact_unit);
+ ASSERT_EQ(compact_unit->output_level, 4);
+ ASSERT_EQ(compact_unit->files.size(), 2);
+}
+
+TEST_F(EarlyFullCompactionTest, TestTotalSizeThreshold) {
+ EarlyFullCompaction
early_full_compaction(/*full_compaction_interval=*/std::nullopt,
+ /*total_size_threshold=*/1000l,
+
/*incremental_size_threshold=*/std::nullopt);
+
+ // total size 300 < 1000, should trigger
+ auto runs = CreateRuns({100l, 200l});
+ auto compact_unit = early_full_compaction.TryFullCompact(/*num_levels=*/5,
runs);
+ ASSERT_TRUE(compact_unit);
+ ASSERT_EQ(compact_unit->output_level, 4);
+ ASSERT_EQ(compact_unit->files.size(), 2);
+ // total size 1000 == 1000, should not trigger
+ runs = CreateRuns({500l, 500l});
+ ASSERT_FALSE(early_full_compaction.TryFullCompact(/*num_levels=*/5, runs));
+ // total size 1500 > 1000, should not trigger
+ runs = CreateRuns({500l, 1000l});
+ ASSERT_FALSE(early_full_compaction.TryFullCompact(/*num_levels=*/5, runs));
+}
+
+TEST_F(EarlyFullCompactionTest, TestIncrementalSizeThreshold) {
+ EarlyFullCompaction
early_full_compaction(/*full_compaction_interval=*/std::nullopt,
+
/*total_size_threshold=*/std::nullopt,
+
/*incremental_size_threshold=*/500l);
+
+ // trigger, no max level
+ auto runs = CreateRuns({400l, 200l});
+ auto compact_unit = early_full_compaction.TryFullCompact(/*num_levels=*/5,
runs);
+ ASSERT_TRUE(compact_unit);
+ ASSERT_EQ(compact_unit->output_level, 4);
+ ASSERT_EQ(compact_unit->files.size(), 2);
+ // no trigger, no max level
+ runs = CreateRuns({100l, 200l});
+ ASSERT_FALSE(early_full_compaction.TryFullCompact(/*num_levels=*/5, runs));
+ // no trigger, with max level
+ runs = {CreateLevelSortedRun(0, 100), CreateLevelSortedRun(0, 300),
+ CreateLevelSortedRun(4, 500)};
+ ASSERT_FALSE(early_full_compaction.TryFullCompact(/*num_levels=*/5, runs));
+ // trigger, with max level
+ runs = {CreateLevelSortedRun(0, 100), CreateLevelSortedRun(0, 300),
+ CreateLevelSortedRun(0, 300), CreateLevelSortedRun(4, 500)};
+ compact_unit = early_full_compaction.TryFullCompact(/*num_levels=*/5,
runs);
+ ASSERT_TRUE(compact_unit);
+ ASSERT_EQ(compact_unit->output_level, 4);
+ ASSERT_EQ(compact_unit->files.size(), 4);
+}
+
+TEST_F(EarlyFullCompactionTest, TestIntervalTriggersFirst) {
+ int64_t current_time = 10000l;
+
+ // Interval will trigger, but size is > threshold
+ TestableEarlyFullCompaction
early_full_compaction(/*full_compaction_interval=*/1000l,
+
/*total_size_threshold=*/500l,
+
/*incremental_size_threshold=*/std::nullopt,
+ ¤t_time);
+ // First time, interval should trigger even if size (600) > threshold (500)
+ auto runs = CreateRuns({300l, 300l});
+ auto compact_unit = early_full_compaction.TryFullCompact(/*num_levels=*/5,
runs);
+ ASSERT_TRUE(compact_unit);
+ ASSERT_EQ(compact_unit->output_level, 4);
+}
+
+TEST_F(EarlyFullCompactionTest, TestThresholdTriggersWhenIntervalFails) {
+ int64_t current_time = 10000l;
+ TestableEarlyFullCompaction
early_full_compaction(/*full_compaction_interval=*/1000l,
+
/*total_size_threshold=*/500l,
+
/*incremental_size_threshold=*/std::nullopt,
+ ¤t_time);
+ // Trigger once to set last compaction time
+ auto runs = CreateRuns({10l, 20l});
+ auto compact_unit = early_full_compaction.TryFullCompact(/*num_levels=*/5,
runs);
+ ASSERT_TRUE(compact_unit);
+ // Advance time, but not enough for interval to trigger
+ current_time += 500;
+
+ // Size (60) < threshold (500), should trigger
+ runs = CreateRuns({30l, 30l});
+ compact_unit = early_full_compaction.TryFullCompact(/*num_levels=*/5,
runs);
+ ASSERT_TRUE(compact_unit);
+ ASSERT_EQ(compact_unit->output_level, 4);
+
+ // Size (600) > threshold (500), should not trigger
+ runs = CreateRuns({300l, 300l});
+ ASSERT_FALSE(early_full_compaction.TryFullCompact(/*num_levels=*/5, runs));
+}
+
+TEST_F(EarlyFullCompactionTest,
TestUpdateLastWhenFullCompactIsTriggeredByTotalSize) {
+ int64_t current_time = 10000l;
+
+ TestableEarlyFullCompaction
early_full_compaction(/*full_compaction_interval=*/1000l,
+
/*total_size_threshold=*/500l,
+
/*incremental_size_threshold=*/std::nullopt,
+ ¤t_time);
+ // First time, interval should trigger even if size (600) > threshold (500)
+ auto runs = CreateRuns({300l, 300l});
+ auto compact_unit = early_full_compaction.TryFullCompact(/*num_levels=*/5,
runs);
+ ASSERT_TRUE(compact_unit);
+ ASSERT_EQ(compact_unit->output_level, 4);
+
+ current_time = 10100l;
+ // Second time, compaction triggered by total_size_threshold
+ runs = CreateRuns({300l, 100l});
+ compact_unit = early_full_compaction.TryFullCompact(/*num_levels=*/5,
runs);
+ ASSERT_TRUE(compact_unit);
+ ASSERT_EQ(compact_unit->output_level, 4);
+
+ current_time = 11001l;
+ // Third time, compaction cannot be triggered as 11001 - 10100 < 1000
full_compaction_interval
+ runs = CreateRuns({300l, 300l});
+ compact_unit = early_full_compaction.TryFullCompact(/*num_levels=*/5,
runs);
+ ASSERT_FALSE(compact_unit);
+}
+
+TEST_F(EarlyFullCompactionTest,
TestUpdateLastWhenFullCompactIsTriggeredByIncSize) {
+ int64_t current_time = 10000l;
+
+ TestableEarlyFullCompaction
early_full_compaction(/*full_compaction_interval=*/1000l,
+
/*total_size_threshold=*/std::nullopt,
+
/*incremental_size_threshold=*/500,
+ ¤t_time);
+ // First time, interval should trigger even if size (400) < threshold (500)
+ std::vector<LevelSortedRun> runs = {CreateLevelSortedRun(0, 300),
CreateLevelSortedRun(0, 100)};
+ auto compact_unit = early_full_compaction.TryFullCompact(/*num_levels=*/5,
runs);
+ ASSERT_TRUE(compact_unit);
+ ASSERT_EQ(compact_unit->output_level, 4);
+
+ current_time = 10100l;
+ // Second time, compaction triggered by total_size_threshold
+ runs = {CreateLevelSortedRun(0, 300), CreateLevelSortedRun(0, 300)};
+ compact_unit = early_full_compaction.TryFullCompact(/*num_levels=*/5,
runs);
+ ASSERT_TRUE(compact_unit);
+ ASSERT_EQ(compact_unit->output_level, 4);
+
+ current_time = 11001l;
+ // Third time, compaction cannot be triggered as 11001 - 10100 < 1000
full_compaction_interval
+ runs = {CreateLevelSortedRun(0, 300), CreateLevelSortedRun(0, 100)};
+ compact_unit = early_full_compaction.TryFullCompact(/*num_levels=*/5,
runs);
+ ASSERT_FALSE(compact_unit);
+}
+
+} // namespace paimon::test
diff --git a/src/paimon/core/mergetree/compact/force_up_level0_compaction.h
b/src/paimon/core/mergetree/compact/force_up_level0_compaction.h
new file mode 100644
index 0000000..d0d4160
--- /dev/null
+++ b/src/paimon/core/mergetree/compact/force_up_level0_compaction.h
@@ -0,0 +1,72 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 <atomic>
+
+#include "paimon/core/mergetree/compact/compact_strategy.h"
+#include "paimon/core/mergetree/compact/universal_compaction.h"
+
+namespace paimon {
+/// A `CompactStrategy` to force compacting level 0 files.
+class ForceUpLevel0Compaction : public CompactStrategy {
+ public:
+ ForceUpLevel0Compaction(const std::shared_ptr<UniversalCompaction>&
universal,
+ const std::optional<int32_t>& max_compact_interval)
+ : universal_(universal), max_compact_interval_(max_compact_interval) {
+ assert(universal_);
+ if (max_compact_interval_) {
+ compact_trigger_count_ = std::make_unique<std::atomic<int32_t>>(0);
+ }
+ }
+
+ std::optional<int32_t> MaxCompactInterval() const {
+ return max_compact_interval_;
+ }
+
+ Result<std::optional<CompactUnit>> Pick(int32_t num_levels,
+ const std::vector<LevelSortedRun>&
runs) override {
+ PAIMON_ASSIGN_OR_RAISE(std::optional<CompactUnit> unit,
universal_->Pick(num_levels, runs));
+ if (unit) {
+ return unit;
+ }
+ if (!max_compact_interval_ || !compact_trigger_count_) {
+ return universal_->ForcePickL0(num_levels, runs);
+ }
+
+ compact_trigger_count_->fetch_add(1);
+ // We must copy max_compact_interval because
compare_exchange_strong(T& expected, T desired)
+ // modifies 'expected' to the current actual value of the atomic if
the comparison fails.
+ int32_t expected_compact_interval = max_compact_interval_.value();
+ if
(compact_trigger_count_->compare_exchange_strong(expected_compact_interval, 0))
{
+ // Universal compaction due to max lookup compaction interval
+ return universal_->ForcePickL0(num_levels, runs);
+ } else {
+ // Skip universal compaction due to lookup compaction trigger
count is less than the max
+ // interval
+ return std::optional<CompactUnit>();
+ }
+ }
+
+ private:
+ std::shared_ptr<UniversalCompaction> universal_;
+ std::optional<int32_t> max_compact_interval_;
+ std::unique_ptr<std::atomic<int32_t>> compact_trigger_count_;
+};
+} // namespace paimon
diff --git
a/src/paimon/core/mergetree/compact/force_up_level0_compaction_test.cpp
b/src/paimon/core/mergetree/compact/force_up_level0_compaction_test.cpp
new file mode 100644
index 0000000..18a1122
--- /dev/null
+++ b/src/paimon/core/mergetree/compact/force_up_level0_compaction_test.cpp
@@ -0,0 +1,117 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.
+ */
+
+#include "paimon/core/mergetree/compact/force_up_level0_compaction.h"
+
+#include "paimon/core/mergetree/compact/universal_compaction.h"
+#include "paimon/testing/utils/testharness.h"
+
+namespace paimon::test {
+class ForceUpLevel0CompactionTest : public testing::Test {
+ public:
+ LevelSortedRun CreateLevelSortedRun(int32_t level, int64_t size) const {
+ auto file_meta = std::make_shared<DataFileMeta>(
+ "fake.data", /*file_size=*/size, /*row_count=*/1,
+ /*min_key=*/BinaryRow::EmptyRow(), /*max_key=*/
+ BinaryRow::EmptyRow(),
+ /*key_stats=*/
+ SimpleStats::EmptyStats(),
+ /*value_stats=*/
+ SimpleStats::EmptyStats(),
+ /*min_sequence_number=*/0, /*max_sequence_number=*/6,
/*schema_id=*/0,
+ /*level=*/0,
/*extra_files=*/std::vector<std::optional<std::string>>(),
+ /*creation_time=*/Timestamp(0ll, 0),
+ /*delete_row_count=*/0, /*embedded_index=*/nullptr,
FileSource::Append(),
+ /*value_stats_cols=*/std::nullopt, /*external_path=*/std::nullopt,
+ /*first_row_id=*/std::nullopt,
+ /*write_cols=*/std::nullopt);
+ return {level, SortedRun::FromSingle(file_meta)};
+ }
+
+ std::vector<LevelSortedRun> CreateRunsWithLevelAndSize(
+ const std::vector<int32_t>& levels, const std::vector<int64_t>& sizes)
const {
+ EXPECT_EQ(levels.size(), sizes.size());
+ std::vector<LevelSortedRun> runs;
+ for (size_t i = 0; i < levels.size(); i++) {
+ runs.push_back(CreateLevelSortedRun(levels[i], sizes[i]));
+ }
+ return runs;
+ }
+};
+
+TEST_F(ForceUpLevel0CompactionTest, TestForceCompaction0) {
+ auto universal =
+ std::make_shared<UniversalCompaction>(/*max_size_amp=*/200,
/*size_ratio=*/1,
+
/*num_run_compaction_trigger=*/5, nullptr, nullptr);
+ ForceUpLevel0Compaction compaction(universal,
/*max_compact_interval=*/std::nullopt);
+
+ ASSERT_OK_AND_ASSIGN(
+ auto unit, compaction.Pick(/*num_levels=*/3,
CreateRunsWithLevelAndSize({0, 0}, {1, 1})));
+ ASSERT_TRUE(unit);
+ ASSERT_EQ(unit.value().output_level, 2);
+
+ ASSERT_OK_AND_ASSIGN(
+ unit, compaction.Pick(/*num_levels=*/3, CreateRunsWithLevelAndSize({0,
1}, {1, 10})));
+ ASSERT_TRUE(unit);
+ ASSERT_EQ(unit.value().output_level, 2);
+
+ ASSERT_OK_AND_ASSIGN(
+ unit, compaction.Pick(/*num_levels=*/3, CreateRunsWithLevelAndSize({0,
0, 2}, {1, 5, 10})));
+ ASSERT_TRUE(unit);
+ ASSERT_EQ(unit.value().output_level, 1);
+
+ ASSERT_OK_AND_ASSIGN(unit,
+ compaction.Pick(/*num_levels=*/3,
CreateRunsWithLevelAndSize({2}, {10})));
+ ASSERT_FALSE(unit);
+
+ ASSERT_OK_AND_ASSIGN(unit,
+ compaction.Pick(/*num_levels=*/3,
+ CreateRunsWithLevelAndSize({0, 0, 0,
0}, {1, 5, 10, 20})));
+ ASSERT_TRUE(unit);
+ ASSERT_EQ(unit.value().output_level, 2);
+}
+
+TEST_F(ForceUpLevel0CompactionTest, TestMaxCompactIntervalConfiguration) {
+ auto universal =
+ std::make_shared<UniversalCompaction>(/*max_size_amp=*/200,
/*size_ratio=*/1,
+
/*num_run_compaction_trigger=*/5, nullptr, nullptr);
+
+ ForceUpLevel0Compaction radical(universal,
/*max_compact_interval=*/std::nullopt);
+ ASSERT_EQ(radical.MaxCompactInterval(), std::nullopt);
+
+ ForceUpLevel0Compaction gentle(universal, /*max_compact_interval=*/10);
+ ASSERT_EQ(gentle.MaxCompactInterval(), 10);
+}
+
+TEST_F(ForceUpLevel0CompactionTest,
TestGentleIntervalShouldForcePickAfterThreshold) {
+ auto universal =
+ std::make_shared<UniversalCompaction>(/*max_size_amp=*/200,
/*size_ratio=*/1,
+
/*num_run_compaction_trigger=*/5, nullptr, nullptr);
+ ForceUpLevel0Compaction compaction(universal, /*max_compact_interval=*/2);
+ auto runs = CreateRunsWithLevelAndSize({0, 0}, {1, 1});
+
+ // First trigger only increases the counter in gentle mode.
+ ASSERT_OK_AND_ASSIGN(auto unit, compaction.Pick(/*num_levels=*/3, runs));
+ ASSERT_FALSE(unit);
+
+ // Second trigger reaches the threshold and forces a level-0 pick.
+ ASSERT_OK_AND_ASSIGN(unit, compaction.Pick(/*num_levels=*/3, runs));
+ ASSERT_TRUE(unit);
+ ASSERT_EQ(unit.value().output_level, 2);
+}
+} // namespace paimon::test
diff --git a/src/paimon/core/mergetree/compact/merge_tree_compact_rewriter.cpp
b/src/paimon/core/mergetree/compact/merge_tree_compact_rewriter.cpp
new file mode 100644
index 0000000..519c170
--- /dev/null
+++ b/src/paimon/core/mergetree/compact/merge_tree_compact_rewriter.cpp
@@ -0,0 +1,291 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.
+ */
+#include "paimon/core/mergetree/compact/merge_tree_compact_rewriter.h"
+
+#include <cassert>
+
+#include "arrow/c/bridge.h"
+#include "arrow/c/helpers.h"
+#include "paimon/common/table/special_fields.h"
+#include "paimon/common/utils/scope_guard.h"
+#include "paimon/core/io/key_value_data_file_writer.h"
+#include "paimon/core/io/key_value_meta_projection_consumer.h"
+#include "paimon/core/io/key_value_record_reader.h"
+#include "paimon/core/io/row_to_arrow_array_converter.h"
+#include "paimon/core/io/single_file_writer.h"
+#include "paimon/core/manifest/file_source.h"
+#include "paimon/core/operation/internal_read_context.h"
+#include "paimon/format/file_format.h"
+#include "paimon/format/writer_builder.h"
+#include "paimon/read_context.h"
+namespace paimon {
+MergeTreeCompactRewriter::MergeTreeCompactRewriter(
+ const BinaryRow& partition, int32_t bucket, int64_t schema_id,
+ const std::vector<std::string>& trimmed_primary_keys, const CoreOptions&
options,
+ const std::shared_ptr<arrow::Schema>& data_schema,
+ const std::shared_ptr<arrow::Schema>& write_schema,
DeletionVector::Factory dv_factory,
+ const std::shared_ptr<FileStorePathFactoryCache>& path_factory_cache,
+ std::unique_ptr<MergeFileSplitRead>&& merge_file_split_read,
+ MergeFunctionWrapperFactory merge_function_wrapper_factory,
+ const std::shared_ptr<CancellationController>& cancellation_controller,
+ const std::shared_ptr<MemoryPool>& pool)
+ : options_(options),
+ merge_file_split_read_(std::move(merge_file_split_read)),
+ pool_(pool),
+ partition_(partition),
+ bucket_(bucket),
+ schema_id_(schema_id),
+ trimmed_primary_keys_(trimmed_primary_keys),
+ data_schema_(data_schema),
+ write_schema_(write_schema),
+ dv_factory_(std::move(dv_factory)),
+ path_factory_cache_(path_factory_cache),
+
merge_function_wrapper_factory_(std::move(merge_function_wrapper_factory)),
+ cancellation_controller_(cancellation_controller) {
+ assert(cancellation_controller_ != nullptr);
+}
+
+Result<std::unique_ptr<MergeTreeCompactRewriter>>
MergeTreeCompactRewriter::Create(
+ int32_t bucket, const BinaryRow& partition, const
std::shared_ptr<TableSchema>& table_schema,
+ DeletionVector::Factory dv_factory,
+ const std::shared_ptr<FileStorePathFactoryCache>& path_factory_cache,
+ const CoreOptions& options,
+ const std::shared_ptr<CancellationController>& cancellation_controller,
+ const std::shared_ptr<MemoryPool>& pool) {
+ PAIMON_ASSIGN_OR_RAISE(std::vector<std::string> trimmed_primary_keys,
+ table_schema->TrimmedPrimaryKeys());
+ auto data_schema =
DataField::ConvertDataFieldsToArrowSchema(table_schema->Fields());
+ auto write_schema =
SpecialFields::CompleteSequenceAndValueKindField(data_schema);
+
+ // TODO(xinyu.lxy): set executor
+ ReadContextBuilder read_context_builder(path_factory_cache->RootPath());
+ read_context_builder.SetOptions(options.ToMap())
+ .WithFileSystem(options.GetFileSystem())
+ .EnablePrefetch(true)
+ .SetPrefetchMaxParallelNum(1)
+ .SetPrefetchBatchCount(3)
+ .WithMemoryPool(pool);
+ PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<ReadContext> read_context,
+ read_context_builder.Finish());
+ // TODO(xinyu.lxy): temporarily disabled pre-buffer for parquet, which may
cause high memory
+ // usage during compaction. Will fix via parquet format refactor.
+ auto new_options = options.ToMap();
+ if (new_options.find("parquet.read.enable-pre-buffer") ==
new_options.end()) {
+ new_options["parquet.read.enable-pre-buffer"] = "false";
+ }
+ PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<InternalReadContext>
internal_context,
+ InternalReadContext::Create(read_context,
table_schema, new_options));
+ PAIMON_ASSIGN_OR_RAISE(
+ std::shared_ptr<FileStorePathFactory> path_factory,
+
path_factory_cache->GetOrCreatePathFactory(options.GetFileFormat()->Identifier()));
+ PAIMON_ASSIGN_OR_RAISE(
+ std::unique_ptr<MergeFileSplitRead> merge_file_split_read,
+ MergeFileSplitRead::Create(path_factory, internal_context, pool,
CreateDefaultExecutor()));
+ auto merge_function_wrapper_factory =
+ [](int32_t output_level) ->
Result<std::shared_ptr<MergeFunctionWrapper<KeyValue>>> {
+ return std::shared_ptr<MergeFunctionWrapper<KeyValue>>();
+ };
+ return std::unique_ptr<MergeTreeCompactRewriter>(new
MergeTreeCompactRewriter(
+ partition, bucket, table_schema->Id(), trimmed_primary_keys, options,
data_schema,
+ write_schema, std::move(dv_factory), path_factory_cache,
std::move(merge_file_split_read),
+ merge_function_wrapper_factory, cancellation_controller, pool));
+}
+
+Result<CompactResult> MergeTreeCompactRewriter::Upgrade(int32_t output_level,
+ const
std::shared_ptr<DataFileMeta>& file) {
+ PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<DataFileMeta> upgraded_file,
+ file->Upgrade(output_level));
+ return CompactResult({file}, {upgraded_file});
+}
+
+Result<CompactResult> MergeTreeCompactRewriter::Rewrite(
+ int32_t output_level, bool drop_delete, const
std::vector<std::vector<SortedRun>>& sections) {
+ return RewriteCompaction(output_level, drop_delete, sections);
+}
+
+std::vector<std::shared_ptr<DataFileMeta>>
MergeTreeCompactRewriter::ExtractFilesFromSections(
+ const std::vector<std::vector<SortedRun>>& sections) {
+ std::vector<std::shared_ptr<DataFileMeta>> files;
+ for (const auto& section : sections) {
+ for (const auto& sorted_run : section) {
+ auto files_in_run = sorted_run.Files();
+ files.insert(files.end(), files_in_run.begin(),
files_in_run.end());
+ }
+ }
+ return files;
+}
+
+std::unique_ptr<MergeTreeCompactRewriter::KeyValueRollingFileWriter>
+MergeTreeCompactRewriter::CreateRollingRowWriter(int32_t level) {
+ auto create_file_writer = [this, level]()
+ -> Result<std::unique_ptr<SingleFileWriter<KeyValueBatch,
std::shared_ptr<DataFileMeta>>>> {
+ ::ArrowSchema arrow_schema{};
+ ScopeGuard guard([&arrow_schema]() {
ArrowSchemaRelease(&arrow_schema); });
+ PAIMON_RETURN_NOT_OK_FROM_ARROW(arrow::ExportSchema(*write_schema_,
&arrow_schema));
+ auto format = options_.GetWriteFileFormat(level);
+ PAIMON_ASSIGN_OR_RAISE(
+ std::shared_ptr<WriterBuilder> writer_builder,
+ format->CreateWriterBuilder(&arrow_schema,
options_.GetWriteBatchSize()));
+ writer_builder->WithMemoryPool(pool_);
+ PAIMON_RETURN_NOT_OK_FROM_ARROW(arrow::ExportSchema(*write_schema_,
&arrow_schema));
+ PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<FormatStatsExtractor>
stats_extractor,
+ format->CreateStatsExtractor(&arrow_schema));
+ auto converter = [](KeyValueBatch key_value_batch, ArrowArray* array)
-> Status {
+ ArrowArrayMove(key_value_batch.batch.get(), array);
+ return Status::OK();
+ };
+ PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<DataFilePathFactory>
data_file_path_factory,
+
CreateDataFilePathFactory(format->Identifier()));
+
+ auto writer = std::make_unique<KeyValueDataFileWriter>(
+ options_.GetWriteFileCompression(level), converter, schema_id_,
level,
+ FileSource::Compact(), trimmed_primary_keys_, stats_extractor,
write_schema_,
+ data_file_path_factory->IsExternalPath(), pool_);
+ PAIMON_RETURN_NOT_OK(writer->Init(options_.GetFileSystem(),
+ data_file_path_factory->NewPath(),
writer_builder));
+ return writer;
+ };
+ return
std::make_unique<MergeTreeCompactRewriter::KeyValueRollingFileWriter>(
+ options_.GetTargetFileSize(/*has_primary_key=*/true),
create_file_writer);
+}
+
+Result<MergeTreeCompactRewriter::KeyValueConsumerCreator>
+MergeTreeCompactRewriter::GenerateKeyValueConsumer() const {
+ if (!merge_file_split_read_) {
+ return Status::Invalid(
+ "merge_file_split_read in MergeTreeCompactRewriter cannot be
nullptr");
+ }
+ PAIMON_ASSIGN_OR_RAISE(std::vector<int32_t> target_to_src_mapping,
+ ArrowUtils::CreateProjection(
+
/*src_schema=*/merge_file_split_read_->GetValueSchema(),
+ /*target_fields=*/data_schema_->fields()));
+ return MergeTreeCompactRewriter::KeyValueConsumerCreator(
+ [target_schema = write_schema_, pool = pool_,
+ target_to_src_mapping = std::move(target_to_src_mapping)]()
+ -> Result<std::unique_ptr<RowToArrowArrayConverter<KeyValue,
KeyValueBatch>>> {
+ return KeyValueMetaProjectionConsumer::Create(target_schema,
target_to_src_mapping,
+ pool);
+ });
+}
+
+Result<std::shared_ptr<DataFilePathFactory>>
MergeTreeCompactRewriter::CreateDataFilePathFactory(
+ const std::string& format) {
+ PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<FileStorePathFactory> path_factory,
+
path_factory_cache_->GetOrCreatePathFactory(format));
+ return path_factory->CreateDataFilePathFactory(partition_, bucket_);
+}
+
+Status MergeTreeCompactRewriter::MergeReadAndWrite(
+ int32_t output_level, bool drop_delete, const std::vector<SortedRun>&
section,
+ const MergeTreeCompactRewriter::KeyValueConsumerCreator& create_consumer,
+ MergeTreeCompactRewriter::KeyValueRollingFileWriter* rolling_writer,
+
std::vector<std::shared_ptr<MergeTreeCompactRewriter::KeyValueMergeReader>>*
+ reader_holders_ptr) {
+ if (!merge_file_split_read_) {
+ return Status::Invalid(
+ "merge_file_split_read in MergeTreeCompactRewriter cannot be
nullptr");
+ }
+ auto& reader_holders = *reader_holders_ptr;
+ // prepare sort merge reader
+ PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<DataFilePathFactory>
data_file_path_factory,
+
CreateDataFilePathFactory(options_.GetFileFormat()->Identifier()));
+ PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<MergeFunctionWrapper<KeyValue>>
wrapper,
+ merge_function_wrapper_factory_(output_level));
+ merge_file_split_read_->SetMergeFunctionWrapper(wrapper);
+ PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<SortMergeReader> sort_merge_reader,
+
merge_file_split_read_->CreateSortMergeReaderForSection(
+ section, partition_, dv_factory_,
+ /*predicate=*/nullptr, data_file_path_factory,
drop_delete));
+ if (!rolling_writer) {
+ // Short-circuit logic: for no rolling writers, simply iterating
through the KeyValue
+ // iterator is sufficient to ensure lookup merge function take effect.
+ while (true) {
+ if (cancellation_controller_->IsCancelled()) {
+ return Status::Cancelled("Compaction is cancelled");
+ }
+ PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<SortMergeReader::Iterator>
key_value_iter,
+ sort_merge_reader->NextBatch());
+ if (key_value_iter == nullptr) {
+ break;
+ }
+ while (true) {
+ PAIMON_ASSIGN_OR_RAISE(bool has_next,
key_value_iter->HasNext());
+ if (!has_next) {
+ break;
+ }
+ [[maybe_unused]] KeyValue kv = key_value_iter->Next();
+ }
+ }
+ return Status::OK();
+ }
+
+ // consumer batch size is WriteBatchSize
+ auto async_key_value_producer_consumer =
+ std::make_shared<AsyncKeyValueProducerAndConsumer<KeyValue,
KeyValueBatch>>(
+ std::move(sort_merge_reader), create_consumer,
options_.GetWriteBatchSize(),
+ /*projection_thread_num=*/1, pool_);
+ reader_holders.push_back(async_key_value_producer_consumer);
+ // read KeyValueBatch from SortMergeReader and write to RollingWriter
+ while (true) {
+ if (cancellation_controller_->IsCancelled()) {
+ return Status::Cancelled("Compaction is cancelled");
+ }
+ 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)));
+ }
+ return Status::OK();
+}
+
+Result<CompactResult> MergeTreeCompactRewriter::RewriteCompaction(
+ int32_t output_level, bool drop_delete, const
std::vector<std::vector<SortedRun>>& sections) {
+ PAIMON_ASSIGN_OR_RAISE(MergeTreeCompactRewriter::KeyValueConsumerCreator
create_consumer,
+ GenerateKeyValueConsumer());
+
+
std::vector<std::shared_ptr<MergeTreeCompactRewriter::KeyValueMergeReader>>
reader_holders;
+ auto rolling_writer = CreateRollingRowWriter(output_level);
+
+ ScopeGuard write_guard([&]() -> void {
+ rolling_writer->Abort();
+ for (const auto& reader : reader_holders) {
+ reader->Close();
+ }
+ merge_file_split_read_.reset();
+ });
+
+ for (const auto& section : sections) {
+ PAIMON_RETURN_NOT_OK(MergeReadAndWrite(output_level, drop_delete,
section, create_consumer,
+ rolling_writer.get(),
&reader_holders));
+ }
+
+ PAIMON_RETURN_NOT_OK(rolling_writer->Close());
+
+ auto before = ExtractFilesFromSections(sections);
+ NotifyRewriteCompactBefore(before);
+ PAIMON_ASSIGN_OR_RAISE(std::vector<std::shared_ptr<DataFileMeta>> after,
+ rolling_writer->GetResult());
+ PAIMON_ASSIGN_OR_RAISE(after, NotifyRewriteCompactAfter(after));
+ write_guard.Release();
+
+ return CompactResult(before, after);
+}
+
+} // namespace paimon
diff --git a/src/paimon/core/mergetree/compact/merge_tree_compact_rewriter.h
b/src/paimon/core/mergetree/compact/merge_tree_compact_rewriter.h
new file mode 100644
index 0000000..3c7ab00
--- /dev/null
+++ b/src/paimon/core/mergetree/compact/merge_tree_compact_rewriter.h
@@ -0,0 +1,126 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 "arrow/api.h"
+#include "paimon/core/compact/cancellation_controller.h"
+#include "paimon/core/core_options.h"
+#include "paimon/core/io/async_key_value_producer_and_consumer.h"
+#include "paimon/core/io/data_file_meta.h"
+#include "paimon/core/io/rolling_file_writer.h"
+#include "paimon/core/key_value.h"
+#include "paimon/core/mergetree/compact/compact_rewriter.h"
+#include "paimon/core/mergetree/merge_tree_writer.h"
+#include "paimon/core/operation/merge_file_split_read.h"
+#include "paimon/core/schema/table_schema.h"
+#include "paimon/core/utils/file_store_path_factory.h"
+#include "paimon/core/utils/file_store_path_factory_cache.h"
+namespace paimon {
+/// Default `CompactRewriter` for merge trees.
+class MergeTreeCompactRewriter : public CompactRewriter {
+ public:
+ using MergeFunctionWrapperFactory =
+
std::function<Result<std::shared_ptr<MergeFunctionWrapper<KeyValue>>>(int32_t)>;
+
+ static Result<std::unique_ptr<MergeTreeCompactRewriter>> Create(
+ int32_t bucket, const BinaryRow& partition,
+ const std::shared_ptr<TableSchema>& table_schema,
DeletionVector::Factory dv_factory,
+ const std::shared_ptr<FileStorePathFactoryCache>& path_factory_cache,
+ const CoreOptions& options,
+ const std::shared_ptr<CancellationController>& cancellation_controller,
+ const std::shared_ptr<MemoryPool>& memory_pool);
+
+ Result<CompactResult> Rewrite(int32_t output_level, bool drop_delete,
+ const std::vector<std::vector<SortedRun>>&
sections) override;
+
+ Result<CompactResult> Upgrade(int32_t output_level,
+ const std::shared_ptr<DataFileMeta>& file)
override;
+
+ Status Close() override {
+ return Status::OK();
+ }
+
+ protected:
+ Result<CompactResult> RewriteCompaction(int32_t output_level, bool
drop_delete,
+ const
std::vector<std::vector<SortedRun>>& sections);
+
+ virtual void NotifyRewriteCompactBefore(
+ const std::vector<std::shared_ptr<DataFileMeta>>& files) {}
+
+ virtual Result<std::vector<std::shared_ptr<DataFileMeta>>>
NotifyRewriteCompactAfter(
+ const std::vector<std::shared_ptr<DataFileMeta>>& files) {
+ return files;
+ }
+
+ static std::vector<std::shared_ptr<DataFileMeta>> ExtractFilesFromSections(
+ const std::vector<std::vector<SortedRun>>& sections);
+
+ MergeTreeCompactRewriter(const BinaryRow& partition, int32_t bucket,
int64_t schema_id,
+ const std::vector<std::string>&
trimmed_primary_keys,
+ const CoreOptions& options,
+ const std::shared_ptr<arrow::Schema>& data_schema,
+ const std::shared_ptr<arrow::Schema>&
write_schema,
+ DeletionVector::Factory dv_factory,
+ const std::shared_ptr<FileStorePathFactoryCache>&
path_factory_cache,
+ std::unique_ptr<MergeFileSplitRead>&&
merge_file_split_read,
+ MergeFunctionWrapperFactory
merge_function_wrapper_factory,
+ const std::shared_ptr<CancellationController>&
cancellation_controller,
+ const std::shared_ptr<MemoryPool>& pool);
+
+ using KeyValueRollingFileWriter =
+ RollingFileWriter<KeyValueBatch, std::shared_ptr<DataFileMeta>>;
+ using KeyValueMergeReader = AsyncKeyValueProducerAndConsumer<KeyValue,
KeyValueBatch>;
+ using KeyValueConsumerCreator =
+ AsyncKeyValueProducerAndConsumer<KeyValue,
KeyValueBatch>::ConsumerCreator;
+
+ std::unique_ptr<KeyValueRollingFileWriter> CreateRollingRowWriter(int32_t
level);
+
+ Result<KeyValueConsumerCreator> GenerateKeyValueConsumer() const;
+
+ Status MergeReadAndWrite(int32_t output_level, bool drop_delete,
+ const std::vector<SortedRun>& section,
+ const KeyValueConsumerCreator& create_consumer,
+ KeyValueRollingFileWriter* rolling_writer,
+
std::vector<std::shared_ptr<KeyValueMergeReader>>* reader_holders_ptr);
+
+ protected:
+ CoreOptions options_;
+ std::unique_ptr<MergeFileSplitRead> merge_file_split_read_;
+
+ private:
+ Result<std::shared_ptr<DataFilePathFactory>> CreateDataFilePathFactory(
+ const std::string& format);
+
+ private:
+ std::shared_ptr<MemoryPool> pool_;
+ BinaryRow partition_;
+ int32_t bucket_;
+ int64_t schema_id_;
+ std::vector<std::string> trimmed_primary_keys_;
+ // all data fields in table schema
+ std::shared_ptr<arrow::Schema> data_schema_;
+ // SequenceNumber + ValueKind + data_schema_
+ std::shared_ptr<arrow::Schema> write_schema_;
+ DeletionVector::Factory dv_factory_;
+ std::shared_ptr<FileStorePathFactoryCache> path_factory_cache_;
+ MergeFunctionWrapperFactory merge_function_wrapper_factory_;
+ std::shared_ptr<CancellationController> cancellation_controller_;
+};
+
+} // namespace paimon
diff --git
a/src/paimon/core/mergetree/compact/merge_tree_compact_rewriter_test.cpp
b/src/paimon/core/mergetree/compact/merge_tree_compact_rewriter_test.cpp
new file mode 100644
index 0000000..d546035
--- /dev/null
+++ b/src/paimon/core/mergetree/compact/merge_tree_compact_rewriter_test.cpp
@@ -0,0 +1,331 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.
+ */
+
+#include "paimon/core/mergetree/compact/merge_tree_compact_rewriter.h"
+
+#include "arrow/api.h"
+#include "arrow/ipc/json_simple.h"
+#include "gtest/gtest.h"
+#include "paimon/common/factories/io_hook.h"
+#include "paimon/common/table/special_fields.h"
+#include "paimon/core/mergetree/compact/interval_partition.h"
+#include "paimon/core/schema/schema_manager.h"
+#include "paimon/core/table/source/data_split_impl.h"
+#include "paimon/format/file_format_factory.h"
+#include "paimon/scan_context.h"
+#include "paimon/table/source/table_scan.h"
+#include "paimon/testing/utils/binary_row_generator.h"
+#include "paimon/testing/utils/io_exception_helper.h"
+#include "paimon/testing/utils/read_result_collector.h"
+#include "paimon/testing/utils/testharness.h"
+
+namespace paimon::test {
+class MergeTreeCompactRewriterTest : public testing::Test {
+ public:
+ Result<std::unique_ptr<MergeTreeCompactRewriter>> CreateCompactRewriter(
+ const std::string& table_path, const std::shared_ptr<TableSchema>&
table_schema,
+ int32_t bucket, const BinaryRow& partition) const {
+ PAIMON_ASSIGN_OR_RAISE(auto options,
CoreOptions::FromMap(table_schema->Options()));
+ auto arrow_schema =
DataField::ConvertDataFieldsToArrowSchema(table_schema->Fields());
+ auto dv_factory = [](const std::string&) ->
Result<std::shared_ptr<DeletionVector>> {
+ return std::shared_ptr<DeletionVector>();
+ };
+
+ auto cancellation_controller =
std::make_shared<CancellationController>();
+ auto path_factory_cache =
+ std::make_shared<FileStorePathFactoryCache>(table_path,
table_schema, options, pool_);
+ return MergeTreeCompactRewriter::Create(bucket, partition,
table_schema, dv_factory,
+ path_factory_cache, options,
+ cancellation_controller,
pool_);
+ }
+
+ Result<std::vector<std::vector<SortedRun>>> GenerateSortedRuns(
+ const std::string& table_path, const std::shared_ptr<TableSchema>&
table_schema,
+ int32_t bucket, const std::map<std::string, std::string>& partition)
const {
+ ScanContextBuilder scan_context_builder(table_path);
+
scan_context_builder.SetBucketFilter(bucket).SetPartitionFilter({partition});
+ PAIMON_ASSIGN_OR_RAISE(auto scan_context,
scan_context_builder.Finish());
+ PAIMON_ASSIGN_OR_RAISE(auto table_scan,
TableScan::Create(std::move(scan_context)));
+ PAIMON_ASSIGN_OR_RAISE(auto result_plan, table_scan->CreatePlan());
+ auto splits = result_plan->Splits();
+ EXPECT_EQ(1, splits.size());
+
+ auto data_split_impl =
std::dynamic_pointer_cast<DataSplitImpl>(splits[0]);
+ EXPECT_TRUE(data_split_impl);
+ auto metas = data_split_impl->DataFiles();
+
+ PAIMON_ASSIGN_OR_RAISE(auto pk_fields,
+
table_schema->GetFields(table_schema->TrimmedPrimaryKeys().value()));
+ PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<FieldsComparator>
key_comparator,
+ FieldsComparator::Create(pk_fields,
/*is_ascending_order=*/true));
+ IntervalPartition interval_partition(metas, key_comparator);
+ return interval_partition.Partition();
+ }
+
+ void CheckResult(const std::string& compact_file_name, const
std::shared_ptr<FileSystem>& fs,
+ const std::shared_ptr<TableSchema>& table_schema,
+ const std::shared_ptr<arrow::ChunkedArray>&
expected_array) const {
+ ASSERT_OK_AND_ASSIGN(auto file_format,
+ FileFormatFactory::Get("orc",
table_schema->Options()));
+ ASSERT_OK_AND_ASSIGN(auto reader_builder,
+
file_format->CreateReaderBuilder(/*batch_size=*/10));
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<InputStream> input_stream,
+ fs->Open(compact_file_name));
+ ASSERT_OK_AND_ASSIGN(auto file_batch_reader,
reader_builder->Build(input_stream));
+ ASSERT_OK_AND_ASSIGN(auto result_array,
+
ReadResultCollector::CollectResult(file_batch_reader.get()));
+ // handle type nullable, as result_array does not have not null flag
+ result_array = result_array->View(expected_array->type()).ValueOrDie();
+
+ ASSERT_TRUE(expected_array->type()->Equals(result_array->type()))
+ << "result=" << result_array->type()->ToString()
+ << ", expected=" << expected_array->type()->ToString() <<
std::endl;
+ ASSERT_TRUE(expected_array->Equals(*result_array)) <<
result_array->ToString();
+ }
+
+ private:
+ std::shared_ptr<MemoryPool> pool_ = GetDefaultPool();
+};
+
+TEST_F(MergeTreeCompactRewriterTest, TestSimple) {
+ std::string origin_table_path = GetDataDir() +
"/orc/pk_table_scan_and_read_mor.db/";
+ auto table_dir = UniqueTestDirectory::Create("local");
+ ASSERT_TRUE(TestUtil::CopyDirectory(origin_table_path, table_dir->Str()));
+ std::string table_path = table_dir->Str() + "/pk_table_scan_and_read_mor";
+ auto fs = table_dir->GetFileSystem();
+
+ // load table schema
+ SchemaManager schema_manager(fs, table_path);
+ ASSERT_OK_AND_ASSIGN(auto table_schema, schema_manager.ReadSchema(0));
+ ASSERT_OK_AND_ASSIGN(
+ auto rewriter,
+ CreateCompactRewriter(table_path, table_schema, /*bucket=*/1,
+
/*partition=*/BinaryRowGenerator::GenerateRow({10}, pool_.get())));
+
+ // generate sorted runs and rewrite
+ ASSERT_OK_AND_ASSIGN(auto runs, GenerateSortedRuns(table_path,
table_schema, /*bucket=*/1,
+ /*partition=*/{{"f1",
"10"}}))
+ ASSERT_OK_AND_ASSIGN(auto compact_result, rewriter->Rewrite(
+ /*output_level=*/5,
/*drop_delete=*/true, runs));
+ // check compact result
+ ASSERT_EQ(4, compact_result.Before().size());
+ ASSERT_EQ(1, compact_result.After().size());
+ const auto& compact_file_meta = compact_result.After()[0];
+ auto expected_file_meta = std::make_shared<DataFileMeta>(
+ "file.orc", 100l, /*row_count=*/7,
+ /*min_key=*/BinaryRowGenerator::GenerateRow({std::string("Bob"), 0},
pool_.get()),
+ /*max_key=*/BinaryRowGenerator::GenerateRow({std::string("Skye2"), 0},
pool_.get()),
+ /*key_stats=*/
+ BinaryRowGenerator::GenerateStats({std::string("Bob"), 0},
{std::string("Skye2"), 0},
+ {0, 0}, pool_.get()),
+ /*value_stats=*/
+ BinaryRowGenerator::GenerateStats({std::string("Bob"), 10, 0, 12.1},
+ {std::string("Skye2"), 10, 0, 31.1},
{0, 0, 0, 0},
+ pool_.get()),
+ /*min_sequence_number=*/0l, /*max_sequence_number=*/10l,
/*schema_id=*/0, /*level=*/5,
+ std::vector<std::optional<std::string>>(), Timestamp(0l, 0),
/*delete_row_count=*/0,
+ nullptr, FileSource::Compact(), std::nullopt, std::nullopt,
std::nullopt, std::nullopt);
+ ASSERT_TRUE(expected_file_meta->TEST_Equal(*compact_file_meta));
+ // check compact file exist
+ std::string compact_file_name =
+ table_path + "/f1=10/bucket-1/" + compact_result.After()[0]->file_name;
+ ASSERT_OK_AND_ASSIGN(bool exist, fs->Exists(compact_file_name));
+ ASSERT_TRUE(exist);
+
+ // check file content
+ auto arrow_schema =
DataField::ConvertDataFieldsToArrowSchema(table_schema->Fields());
+ auto type_with_special_fields =
+
arrow::struct_(SpecialFields::CompleteSequenceAndValueKindField(arrow_schema)->fields());
+ std::shared_ptr<arrow::ChunkedArray> expected_array;
+ auto array_status =
+
arrow::ipc::internal::json::ChunkedArrayFromJSON(type_with_special_fields, {R"([
+[0, 0, "Bob", 10, 0, 12.1],
+[4, 0, "David", 10, 0, 17.1],
+[1, 0, "Emily", 10, 0, 13.1],
+[7, 0, "Marco", 10, 0, 21.1],
+[10, 0, "Marco2", 10, 0, 31.1],
+[6, 0, "Skye", 10, 0, 21],
+[9, 0, "Skye2", 10, 0, 31]
+])"},
+ &expected_array);
+ ASSERT_TRUE(array_status.ok());
+ CheckResult(compact_file_name, fs, table_schema, expected_array);
+}
+
+TEST_F(MergeTreeCompactRewriterTest, TestCancel) {
+ std::string origin_table_path = GetDataDir() +
"/orc/pk_table_scan_and_read_mor.db/";
+ auto table_dir = UniqueTestDirectory::Create("local");
+ ASSERT_TRUE(TestUtil::CopyDirectory(origin_table_path, table_dir->Str()));
+ std::string table_path = table_dir->Str() + "/pk_table_scan_and_read_mor";
+ auto fs = table_dir->GetFileSystem();
+
+ SchemaManager schema_manager(fs, table_path);
+ ASSERT_OK_AND_ASSIGN(auto table_schema, schema_manager.ReadSchema(0));
+ ASSERT_OK_AND_ASSIGN(auto options,
CoreOptions::FromMap(table_schema->Options()));
+ auto arrow_schema =
DataField::ConvertDataFieldsToArrowSchema(table_schema->Fields());
+ auto dv_factory = [](const std::string&) ->
Result<std::shared_ptr<DeletionVector>> {
+ return std::shared_ptr<DeletionVector>();
+ };
+ auto cancellation_controller = std::make_shared<CancellationController>();
+ auto path_factory_cache =
+ std::make_shared<FileStorePathFactoryCache>(table_path, table_schema,
options, pool_);
+ ASSERT_OK_AND_ASSIGN(
+ auto rewriter,
+ MergeTreeCompactRewriter::Create(
+ /*bucket=*/1, /*partition=*/BinaryRowGenerator::GenerateRow({10},
pool_.get()),
+ table_schema, dv_factory, path_factory_cache, options,
cancellation_controller, pool_));
+
+ ASSERT_OK_AND_ASSIGN(auto runs, GenerateSortedRuns(table_path,
table_schema, /*bucket=*/1,
+ /*partition=*/{{"f1",
"10"}}));
+
+ // cancel compaction here
+ cancellation_controller->Cancel();
+ ASSERT_NOK_WITH_MSG(rewriter->Rewrite(/*output_level=*/5,
/*drop_delete=*/true, runs),
+ "Compaction is cancelled");
+}
+
+TEST_F(MergeTreeCompactRewriterTest, TestNotDropDelete) {
+ std::string origin_table_path = GetDataDir() +
"/orc/pk_table_scan_and_read_mor.db/";
+ auto table_dir = UniqueTestDirectory::Create("local");
+ ASSERT_TRUE(TestUtil::CopyDirectory(origin_table_path, table_dir->Str()));
+ std::string table_path = table_dir->Str() + "/pk_table_scan_and_read_mor";
+ auto fs = table_dir->GetFileSystem();
+
+ // load table schema
+ SchemaManager schema_manager(fs, table_path);
+ ASSERT_OK_AND_ASSIGN(auto table_schema, schema_manager.ReadSchema(0));
+ ASSERT_OK_AND_ASSIGN(
+ auto rewriter,
+ CreateCompactRewriter(table_path, table_schema, /*bucket=*/1,
+
/*partition=*/BinaryRowGenerator::GenerateRow({10}, pool_.get())));
+
+ // generate sorted runs and rewrite
+ ASSERT_OK_AND_ASSIGN(auto runs, GenerateSortedRuns(table_path,
table_schema, /*bucket=*/1,
+ /*partition=*/{{"f1",
"10"}}))
+ ASSERT_OK_AND_ASSIGN(auto compact_result, rewriter->Rewrite(
+ /*output_level=*/5,
/*drop_delete=*/false, runs));
+ // check compact result
+ ASSERT_EQ(4, compact_result.Before().size());
+ ASSERT_EQ(1, compact_result.After().size());
+ const auto& compact_file_meta = compact_result.After()[0];
+ auto expected_file_meta = std::make_shared<DataFileMeta>(
+ "file.orc", 100l, /*row_count=*/9,
+ /*min_key=*/BinaryRowGenerator::GenerateRow({std::string("Alex"), 0},
pool_.get()),
+ /*max_key=*/BinaryRowGenerator::GenerateRow({std::string("Tony"), 0},
pool_.get()),
+ /*key_stats=*/
+ BinaryRowGenerator::GenerateStats({std::string("Alex"), 0},
{std::string("Tony"), 0},
+ {0, 0}, pool_.get()),
+ /*value_stats=*/
+ BinaryRowGenerator::GenerateStats({std::string("Alex"), 10, 0, 12.1},
+ {std::string("Tony"), 10, 0, 31.2},
{0, 0, 0, 0},
+ pool_.get()),
+ /*min_sequence_number=*/0l, /*max_sequence_number=*/11l,
/*schema_id=*/0, /*level=*/5,
+ std::vector<std::optional<std::string>>(), Timestamp(0l, 0),
/*delete_row_count=*/2,
+ nullptr, FileSource::Compact(), std::nullopt, std::nullopt,
std::nullopt, std::nullopt);
+ ASSERT_TRUE(expected_file_meta->TEST_Equal(*compact_file_meta));
+
+ std::string compact_file_name =
+ table_path + "/f1=10/bucket-1/" + compact_result.After()[0]->file_name;
+ ASSERT_OK_AND_ASSIGN(bool exist, fs->Exists(compact_file_name));
+ ASSERT_TRUE(exist);
+
+ // check file content
+ auto arrow_schema =
DataField::ConvertDataFieldsToArrowSchema(table_schema->Fields());
+ auto type_with_special_fields =
+
arrow::struct_(SpecialFields::CompleteSequenceAndValueKindField(arrow_schema)->fields());
+ std::shared_ptr<arrow::ChunkedArray> expected_array;
+ auto array_status =
+
arrow::ipc::internal::json::ChunkedArrayFromJSON(type_with_special_fields, {R"([
+[11, 3, "Alex", 10, 0, 31.2],
+[0, 0, "Bob", 10, 0, 12.1],
+[4, 0, "David", 10, 0, 17.1],
+[1, 0, "Emily", 10, 0, 13.1],
+[7, 0, "Marco", 10, 0, 21.1],
+[10, 0, "Marco2", 10, 0, 31.1],
+[6, 0, "Skye", 10, 0, 21],
+[9, 0, "Skye2", 10, 0, 31],
+[5, 3, "Tony", 10, 0, 14.1]
+])"},
+ &expected_array);
+ ASSERT_TRUE(array_status.ok());
+ CheckResult(compact_file_name, fs, table_schema, expected_array);
+}
+
+TEST_F(MergeTreeCompactRewriterTest, TestIOException) {
+ std::string origin_table_path = GetDataDir() +
"/orc/pk_table_scan_and_read_mor.db/";
+
+ bool run_complete = false;
+ auto io_hook = IOHook::GetInstance();
+ for (size_t i = 0; i < 500; i += RandomNumber(1, 17)) {
+ auto table_dir = UniqueTestDirectory::Create("local");
+ ASSERT_TRUE(TestUtil::CopyDirectory(origin_table_path,
table_dir->Str()));
+ std::string table_path = table_dir->Str() +
"/pk_table_scan_and_read_mor";
+ auto fs = table_dir->GetFileSystem();
+
+ // load table schema
+ SchemaManager schema_manager(fs, table_path);
+ ASSERT_OK_AND_ASSIGN(auto table_schema, schema_manager.ReadSchema(0));
+ ASSERT_OK_AND_ASSIGN(auto rewriter,
+ CreateCompactRewriter(
+ table_path, table_schema, /*bucket=*/1,
+
/*partition=*/BinaryRowGenerator::GenerateRow({10}, pool_.get())));
+
+ // generate sorted runs and rewrite
+ ASSERT_OK_AND_ASSIGN(auto runs, GenerateSortedRuns(table_path,
table_schema, /*bucket=*/1,
+
/*partition=*/{{"f1", "10"}}))
+ // rewrite may trigger I/O exception
+ ScopeGuard guard([&io_hook]() { io_hook->Clear(); });
+ io_hook->Reset(i, IOHook::Mode::RETURN_ERROR);
+ auto compact_result = rewriter->Rewrite(
+ /*output_level=*/5, /*drop_delete=*/true, runs);
+ CHECK_HOOK_STATUS(compact_result.status(), i);
+ io_hook->Clear();
+
+ // check compact result
+ ASSERT_EQ(4, compact_result.value().Before().size());
+ ASSERT_EQ(1, compact_result.value().After().size());
+ std::string compact_file_name =
+ table_path + "/f1=10/bucket-1/" +
compact_result.value().After()[0]->file_name;
+ ASSERT_OK_AND_ASSIGN(bool exist, fs->Exists(compact_file_name));
+ ASSERT_TRUE(exist);
+
+ // check file content
+ auto arrow_schema =
DataField::ConvertDataFieldsToArrowSchema(table_schema->Fields());
+ auto type_with_special_fields = arrow::struct_(
+
SpecialFields::CompleteSequenceAndValueKindField(arrow_schema)->fields());
+ std::shared_ptr<arrow::ChunkedArray> expected_array;
+ auto array_status =
+
arrow::ipc::internal::json::ChunkedArrayFromJSON(type_with_special_fields, {R"([
+[0, 0, "Bob", 10, 0, 12.1],
+[4, 0, "David", 10, 0, 17.1],
+[1, 0, "Emily", 10, 0, 13.1],
+[7, 0, "Marco", 10, 0, 21.1],
+[10, 0, "Marco2", 10, 0, 31.1],
+[6, 0, "Skye", 10, 0, 21],
+[9, 0, "Skye2", 10, 0, 31]
+])"},
+ &expected_array);
+ ASSERT_TRUE(array_status.ok());
+ CheckResult(compact_file_name, fs, table_schema, expected_array);
+ run_complete = true;
+ break;
+ }
+ ASSERT_TRUE(run_complete);
+}
+
+} // namespace paimon::test
diff --git a/src/paimon/core/mergetree/compact/universal_compaction.cpp
b/src/paimon/core/mergetree/compact/universal_compaction.cpp
new file mode 100644
index 0000000..3271513
--- /dev/null
+++ b/src/paimon/core/mergetree/compact/universal_compaction.cpp
@@ -0,0 +1,185 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.
+ */
+
+#include "paimon/core/mergetree/compact/universal_compaction.h"
+
+#include "paimon/common/utils/date_time_utils.h"
+namespace paimon {
+UniversalCompaction::UniversalCompaction(
+ int32_t max_size_amp, int32_t size_ratio, int32_t
num_run_compaction_trigger,
+ const std::shared_ptr<EarlyFullCompaction>& early_full_compaction,
+ const std::shared_ptr<OffPeakHours>& off_peak_hours)
+ : max_size_amp_(max_size_amp),
+ size_ratio_(size_ratio),
+ num_run_compaction_trigger_(num_run_compaction_trigger),
+ early_full_compaction_(early_full_compaction),
+ off_peak_hours_(off_peak_hours) {
+ assert(num_run_compaction_trigger_ >= 1);
+}
+
+Result<std::optional<CompactUnit>> UniversalCompaction::Pick(
+ int32_t num_levels, const std::vector<LevelSortedRun>& runs) {
+ int32_t max_level = num_levels - 1;
+ // 0 try full compaction by trigger
+ if (early_full_compaction_) {
+ std::optional<CompactUnit> compact_unit =
+ early_full_compaction_->TryFullCompact(num_levels, runs);
+ if (compact_unit) {
+ return compact_unit;
+ }
+ }
+ // 1 checking for reducing size amplification
+ std::optional<CompactUnit> compact_unit = PickForSizeAmp(max_level, runs);
+ if (compact_unit) {
+ return compact_unit;
+ }
+ // 2 checking for size ratio
+ PAIMON_ASSIGN_OR_RAISE(compact_unit, PickForSizeRatio(max_level, runs));
+ if (compact_unit) {
+ return compact_unit;
+ }
+ // 3 checking for file num
+ if (runs.size() > static_cast<size_t>(num_run_compaction_trigger_)) {
+ // compacting for file num
+ int32_t candidate_count = runs.size() - num_run_compaction_trigger_ +
1;
+ PAIMON_ASSIGN_OR_RAISE(std::optional<CompactUnit> compact_unit,
+ PickForSizeRatio(max_level, runs,
candidate_count));
+ return compact_unit;
+ }
+ return std::optional<CompactUnit>();
+}
+
+Result<std::optional<CompactUnit>> UniversalCompaction::ForcePickL0(
+ int32_t num_levels, const std::vector<LevelSortedRun>& runs) {
+ // collect all level 0 files
+ int32_t candidate_count = 0;
+ for (; static_cast<size_t>(candidate_count) < runs.size();
++candidate_count) {
+ if (runs[candidate_count].level > 0) {
+ break;
+ }
+ }
+ if (candidate_count == 0) {
+ return std::optional<CompactUnit>();
+ }
+ return PickForSizeRatio(num_levels - 1, runs, candidate_count,
/*force_pick=*/true);
+}
+
+std::optional<CompactUnit> UniversalCompaction::PickForSizeAmp(
+ int32_t max_level, const std::vector<LevelSortedRun>& runs) {
+ if (runs.size() < static_cast<size_t>(num_run_compaction_trigger_)) {
+ return std::nullopt;
+ }
+ int64_t candidate_size = 0;
+ for (size_t i = 0; i < runs.size() - 1; ++i) {
+ candidate_size += runs[i].run.TotalSize();
+ }
+ int64_t earliest_size = runs[runs.size() - 1].run.TotalSize();
+
+ // size amplification = percentage of additional size
+ if (candidate_size * 100 > max_size_amp_ * earliest_size) {
+ if (early_full_compaction_) {
+ early_full_compaction_->UpdateLastFullCompaction();
+ }
+ return CompactUnit::FromLevelRuns(max_level, runs);
+ }
+ return std::nullopt;
+}
+
+Result<std::optional<CompactUnit>> UniversalCompaction::PickForSizeRatio(
+ int32_t max_level, const std::vector<LevelSortedRun>& runs) {
+ if (runs.size() < static_cast<size_t>(num_run_compaction_trigger_)) {
+ return std::optional<CompactUnit>();
+ }
+ return PickForSizeRatio(max_level, runs, /*candidate_count=*/1);
+}
+
+Result<std::optional<CompactUnit>> UniversalCompaction::PickForSizeRatio(
+ int32_t max_level, const std::vector<LevelSortedRun>& runs, int32_t
candidate_count) {
+ return PickForSizeRatio(max_level, runs, candidate_count,
/*force_pick=*/false);
+}
+
+Result<std::optional<CompactUnit>> UniversalCompaction::PickForSizeRatio(
+ int32_t max_level, const std::vector<LevelSortedRun>& runs, int32_t
candidate_count,
+ bool force_pick) {
+ int64_t candidate_size = CandidateSize(runs, candidate_count);
+ for (size_t i = candidate_count; i < runs.size(); ++i) {
+ LevelSortedRun next = runs[i];
+ PAIMON_ASSIGN_OR_RAISE(int32_t current_hour_ratio, RatioForOffPeak());
+ if (static_cast<double>(candidate_size) * (100.0 + size_ratio_ +
current_hour_ratio) /
+ 100.0 <
+ next.run.TotalSize()) {
+ break;
+ }
+ candidate_size += next.run.TotalSize();
+ candidate_count++;
+ }
+ if (force_pick || candidate_count > 1) {
+ return std::optional<CompactUnit>(CreateUnit(runs, max_level,
candidate_count));
+ }
+ return std::optional<CompactUnit>();
+}
+
+int64_t UniversalCompaction::CandidateSize(const std::vector<LevelSortedRun>&
runs,
+ int32_t candidate_count) {
+ int64_t size = 0;
+ for (int32_t i = 0; i < candidate_count; ++i) {
+ size += runs[i].run.TotalSize();
+ }
+ return size;
+}
+
+Result<int32_t> UniversalCompaction::RatioForOffPeak() const {
+ PAIMON_ASSIGN_OR_RAISE(int32_t local_hour,
DateTimeUtils::GetCurrentLocalHour());
+ return !off_peak_hours_ ? 0 : off_peak_hours_->CurrentRatio(local_hour);
+}
+
+CompactUnit UniversalCompaction::CreateUnit(const std::vector<LevelSortedRun>&
runs,
+ int32_t max_level, int32_t
run_count) {
+ int32_t output_level;
+ if (static_cast<size_t>(run_count) == runs.size()) {
+ output_level = max_level;
+ } else {
+ // level of next run - 1
+ output_level = std::max(0, runs[run_count].level - 1);
+ }
+
+ if (output_level == 0) {
+ // do not output level 0
+ for (size_t i = run_count; i < runs.size(); ++i) {
+ LevelSortedRun next = runs[i];
+ run_count++;
+ if (next.level != 0) {
+ output_level = next.level;
+ break;
+ }
+ }
+ }
+ if (static_cast<size_t>(run_count) == runs.size()) {
+ if (early_full_compaction_) {
+ early_full_compaction_->UpdateLastFullCompaction();
+ }
+ output_level = max_level;
+ }
+ std::vector<LevelSortedRun> result_runs;
+ result_runs.reserve(run_count);
+ for (int32_t i = 0; i < run_count; ++i) {
+ result_runs.push_back(runs[i]);
+ }
+ return CompactUnit::FromLevelRuns(output_level, result_runs);
+}
+} // namespace paimon
diff --git a/src/paimon/core/mergetree/compact/universal_compaction.h
b/src/paimon/core/mergetree/compact/universal_compaction.h
new file mode 100644
index 0000000..735d450
--- /dev/null
+++ b/src/paimon/core/mergetree/compact/universal_compaction.h
@@ -0,0 +1,66 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 "paimon/core/mergetree/compact/compact_strategy.h"
+#include "paimon/core/mergetree/compact/early_full_compaction.h"
+#include "paimon/core/mergetree/compact/off_peak_hours.h"
+
+namespace paimon {
+/// Universal Compaction Style is a compaction style, targeting the use cases
requiring lower write
+/// amplification, trading off read amplification and space amplification.
+///
+/// See RocksDb Universal-Compaction:
+/// https://github.com/facebook/rocksdb/wiki/Universal-Compaction.
+class UniversalCompaction : public CompactStrategy {
+ public:
+ UniversalCompaction(int32_t max_size_amp, int32_t size_ratio,
+ int32_t num_run_compaction_trigger,
+ const std::shared_ptr<EarlyFullCompaction>&
early_full_compaction,
+ const std::shared_ptr<OffPeakHours>& off_peak_hours);
+ Result<std::optional<CompactUnit>> Pick(int32_t num_levels,
+ const std::vector<LevelSortedRun>&
runs) override;
+
+ Result<std::optional<CompactUnit>> ForcePickL0(int32_t num_levels,
+ const
std::vector<LevelSortedRun>& runs);
+
+ private:
+ std::optional<CompactUnit> PickForSizeAmp(int32_t max_level,
+ const
std::vector<LevelSortedRun>& runs);
+ Result<std::optional<CompactUnit>> PickForSizeRatio(int32_t max_level,
+ const
std::vector<LevelSortedRun>& runs);
+ Result<std::optional<CompactUnit>> PickForSizeRatio(int32_t max_level,
+ const
std::vector<LevelSortedRun>& runs,
+ int32_t
candidate_count);
+ Result<std::optional<CompactUnit>> PickForSizeRatio(int32_t max_level,
+ const
std::vector<LevelSortedRun>& runs,
+ int32_t
candidate_count, bool force_pick);
+ Result<int32_t> RatioForOffPeak() const;
+ CompactUnit CreateUnit(const std::vector<LevelSortedRun>& runs, int32_t
max_level,
+ int32_t run_count);
+ static int64_t CandidateSize(const std::vector<LevelSortedRun>& runs,
int32_t candidate_count);
+
+ private:
+ int32_t max_size_amp_;
+ int32_t size_ratio_;
+ int32_t num_run_compaction_trigger_;
+ std::shared_ptr<EarlyFullCompaction> early_full_compaction_;
+ std::shared_ptr<OffPeakHours> off_peak_hours_;
+};
+} // namespace paimon
diff --git a/src/paimon/core/mergetree/compact/universal_compaction_test.cpp
b/src/paimon/core/mergetree/compact/universal_compaction_test.cpp
new file mode 100644
index 0000000..74abaca
--- /dev/null
+++ b/src/paimon/core/mergetree/compact/universal_compaction_test.cpp
@@ -0,0 +1,464 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.
+ */
+
+#include "paimon/core/mergetree/compact/universal_compaction.h"
+
+#include "paimon/core/mergetree/compact/force_up_level0_compaction.h"
+#include "paimon/testing/utils/testharness.h"
+
+namespace paimon::test {
+class UniversalCompactionTest : public testing::Test {
+ public:
+ class TestableEarlyFullCompaction : public EarlyFullCompaction {
+ TestableEarlyFullCompaction(const std::optional<int64_t>&
full_compaction_interval,
+ const std::optional<int64_t>&
total_size_threshold,
+ const std::optional<int64_t>&
incremental_size_threshold,
+ const int64_t* current_time)
+ : EarlyFullCompaction(full_compaction_interval,
total_size_threshold,
+ incremental_size_threshold),
+ current_time_(current_time) {}
+
+ int64_t CurrentTimeMillis() const override {
+ return *current_time_;
+ }
+
+ private:
+ const int64_t* current_time_;
+ };
+
+ LevelSortedRun CreateLevelSortedRun(int32_t level, int64_t size) const {
+ auto file_meta = std::make_shared<DataFileMeta>(
+ "fake.data", /*file_size=*/size, /*row_count=*/1,
+ /*min_key=*/BinaryRow::EmptyRow(), /*max_key=*/
+ BinaryRow::EmptyRow(),
+ /*key_stats=*/
+ SimpleStats::EmptyStats(),
+ /*value_stats=*/
+ SimpleStats::EmptyStats(),
+ /*min_sequence_number=*/0, /*max_sequence_number=*/6,
/*schema_id=*/0,
+ /*level=*/0,
/*extra_files=*/std::vector<std::optional<std::string>>(),
+ /*creation_time=*/Timestamp(0ll, 0),
+ /*delete_row_count=*/0, /*embedded_index=*/nullptr,
FileSource::Append(),
+ /*value_stats_cols=*/std::nullopt, /*external_path=*/std::nullopt,
+ /*first_row_id=*/std::nullopt,
+ /*write_cols=*/std::nullopt);
+ return {level, SortedRun::FromSingle(file_meta)};
+ }
+
+ std::vector<LevelSortedRun> CreateRunsWithLevel(const
std::vector<int32_t>& levels) const {
+ std::vector<LevelSortedRun> runs;
+ for (const auto& level : levels) {
+ runs.push_back(CreateLevelSortedRun(level, /*size=*/1));
+ }
+ return runs;
+ }
+
+ std::vector<LevelSortedRun> CreateRunsWithSize(const std::vector<int64_t>&
sizes) const {
+ std::vector<LevelSortedRun> runs;
+ for (const auto& size : sizes) {
+ runs.push_back(CreateLevelSortedRun(/*level=*/0, size));
+ }
+ return runs;
+ }
+
+ std::vector<LevelSortedRun> CreateRunsWithLevelAndSize(
+ const std::vector<int32_t>& levels, const std::vector<int64_t>& sizes)
const {
+ EXPECT_EQ(levels.size(), sizes.size());
+ std::vector<LevelSortedRun> runs;
+ for (size_t i = 0; i < levels.size(); i++) {
+ runs.push_back(CreateLevelSortedRun(levels[i], sizes[i]));
+ }
+ return runs;
+ }
+
+ std::vector<int64_t> GetFileSizeVecFromCompactUnit(const CompactUnit&
unit) const {
+ std::vector<int64_t> sizes;
+ for (const auto& file : unit.files) {
+ sizes.push_back(file->file_size);
+ }
+ return sizes;
+ }
+};
+
+TEST_F(UniversalCompactionTest, TestOutputLevel) {
+ UniversalCompaction compaction(/*max_size_amp=*/25, /*size_ratio=*/1,
+ /*num_run_compaction_trigger=*/3, nullptr,
nullptr);
+ ASSERT_EQ(
+ 1, compaction
+ .CreateUnit(CreateRunsWithLevel({0, 0, 1, 3, 4}),
/*max_level=*/5, /*run_count=*/1)
+ .output_level);
+ ASSERT_EQ(
+ 1, compaction
+ .CreateUnit(CreateRunsWithLevel({0, 0, 1, 3, 4}),
/*max_level=*/5, /*run_count=*/2)
+ .output_level);
+ ASSERT_EQ(
+ 2, compaction
+ .CreateUnit(CreateRunsWithLevel({0, 0, 1, 3, 4}),
/*max_level=*/5, /*run_count=*/3)
+ .output_level);
+ ASSERT_EQ(
+ 3, compaction
+ .CreateUnit(CreateRunsWithLevel({0, 0, 1, 3, 4}),
/*max_level=*/5, /*run_count=*/4)
+ .output_level);
+ ASSERT_EQ(
+ 5, compaction
+ .CreateUnit(CreateRunsWithLevel({0, 0, 1, 3, 4}),
/*max_level=*/5, /*run_count=*/5)
+ .output_level);
+}
+
+TEST_F(UniversalCompactionTest, TestPick) {
+ UniversalCompaction compaction(/*max_size_amp=*/25, /*size_ratio=*/1,
+ /*num_run_compaction_trigger=*/3, nullptr,
nullptr);
+ // by size amplification
+ ASSERT_OK_AND_ASSIGN(auto pick,
+ compaction.Pick(/*num_levels=*/3,
CreateRunsWithSize({1, 2, 3, 3})));
+ ASSERT_TRUE(pick);
+ ASSERT_EQ(GetFileSizeVecFromCompactUnit(pick.value()),
std::vector<int64_t>({1, 2, 3, 3}));
+
+ // by size ratio
+ ASSERT_OK_AND_ASSIGN(pick, compaction.Pick(/*num_levels=*/4,
CreateRunsWithLevelAndSize(
+
/*levels=*/{0, 1, 2, 3},
+
/*sizes=*/{1, 1, 1, 50})));
+ ASSERT_TRUE(pick);
+ ASSERT_EQ(GetFileSizeVecFromCompactUnit(pick.value()),
std::vector<int64_t>({1, 1, 1}));
+
+ // by file num
+ ASSERT_OK_AND_ASSIGN(pick, compaction.Pick(/*num_levels=*/4,
CreateRunsWithLevelAndSize(
+
/*levels=*/{0, 1, 2, 3},
+
/*sizes=*/{1, 50, 3, 500})));
+ ASSERT_TRUE(pick);
+ // 3 should be in the candidate, by size ratio after picking by file num
+ ASSERT_EQ(GetFileSizeVecFromCompactUnit(pick.value()),
std::vector<int64_t>({1, 50, 3}));
+}
+
+TEST_F(UniversalCompactionTest, TestAllLevelRunsInvolved) {
+ int64_t current_time = 0;
+ auto full_compact_trigger = std::make_shared<TestableEarlyFullCompaction>(
+ /*full_compaction_interval=*/std::nullopt,
+ /*total_size_threshold=*/std::nullopt,
+ /*incremental_size_threshold=*/1000l, ¤t_time);
+ UniversalCompaction compaction(/*max_size_amp=*/100, /*size_ratio=*/1,
+ /*num_run_compaction_trigger=*/3,
full_compact_trigger, nullptr);
+ ASSERT_OK_AND_ASSIGN(auto pick, compaction.Pick(/*num_levels=*/3,
CreateRunsWithLevelAndSize(
+
/*levels=*/{0, 0, 0},
+
/*sizes=*/{1, 1, 3})));
+ ASSERT_TRUE(pick);
+ ASSERT_EQ(GetFileSizeVecFromCompactUnit(pick.value()),
std::vector<int64_t>({1, 1, 3}));
+}
+
+TEST_F(UniversalCompactionTest, TestOptimizedCompactionInterval) {
+ int64_t current_time = 0;
+ auto full_compact_trigger = std::make_shared<TestableEarlyFullCompaction>(
+ /*full_compaction_interval=*/1000L,
+ /*total_size_threshold=*/std::nullopt,
+ /*incremental_size_threshold=*/std::nullopt, ¤t_time);
+ UniversalCompaction compaction(/*max_size_amp=*/100, /*size_ratio=*/1,
+ /*num_run_compaction_trigger=*/3,
full_compact_trigger, nullptr);
+ // first time, force optimized compaction
+ ASSERT_OK_AND_ASSIGN(auto pick, compaction.Pick(/*num_levels=*/3,
CreateRunsWithLevelAndSize(
+
/*levels=*/{0, 1, 2},
+
/*sizes=*/{1, 3, 5})));
+ ASSERT_TRUE(pick);
+ ASSERT_EQ(GetFileSizeVecFromCompactUnit(pick.value()),
std::vector<int64_t>({1, 3, 5}));
+
+ // modify time, optimized compaction
+ current_time = 1001L;
+ ASSERT_OK_AND_ASSIGN(pick, compaction.Pick(/*num_levels=*/3,
CreateRunsWithLevelAndSize(
+
/*levels=*/{0, 1, 2},
+
/*sizes=*/{1, 3, 5})));
+ ASSERT_TRUE(pick);
+ ASSERT_EQ(GetFileSizeVecFromCompactUnit(pick.value()),
std::vector<int64_t>({1, 3, 5}));
+
+ // third time, no compaction
+ ASSERT_OK_AND_ASSIGN(pick, compaction.Pick(/*num_levels=*/3,
CreateRunsWithLevelAndSize(
+
/*levels=*/{0, 1, 2},
+
/*sizes=*/{1, 3, 5})));
+ ASSERT_FALSE(pick);
+
+ // 4 time, pickForSizeAmp
+ current_time = 1500L;
+ ASSERT_OK_AND_ASSIGN(pick, compaction.Pick(/*num_levels=*/3,
CreateRunsWithLevelAndSize(
+
/*levels=*/{0, 1, 2},
+
/*sizes=*/{3, 3, 5})));
+ ASSERT_TRUE(pick);
+ ASSERT_EQ(GetFileSizeVecFromCompactUnit(pick.value()),
std::vector<int64_t>({3, 3, 5}));
+
+ // 5 time, no compaction because pickForSizeAmp already done
+ current_time = 2001L;
+ ASSERT_OK_AND_ASSIGN(pick, compaction.Pick(/*num_levels=*/3,
CreateRunsWithLevelAndSize(
+
/*levels=*/{0, 1, 2},
+
/*sizes=*/{1, 3, 5})));
+ ASSERT_FALSE(pick);
+}
+
+TEST_F(UniversalCompactionTest, TestTotalSizeThreshold) {
+ auto full_compact_trigger = std::make_shared<EarlyFullCompaction>(
+ /*full_compaction_interval=*/std::nullopt,
+ /*total_size_threshold=*/10L,
+ /*incremental_size_threshold=*/std::nullopt);
+
+ UniversalCompaction compaction(/*max_size_amp=*/100, /*size_ratio=*/1,
+ /*num_run_compaction_trigger=*/3,
full_compact_trigger, nullptr);
+ // total size less than threshold
+ ASSERT_OK_AND_ASSIGN(auto pick, compaction.Pick(/*num_levels=*/3,
CreateRunsWithLevelAndSize(
+
/*levels=*/{0, 1, 2},
+
/*sizes=*/{1, 3, 5})));
+ ASSERT_TRUE(pick);
+ ASSERT_EQ(GetFileSizeVecFromCompactUnit(pick.value()),
std::vector<int64_t>({1, 3, 5}));
+
+ // total size bigger than threshold
+ ASSERT_OK_AND_ASSIGN(pick, compaction.Pick(/*num_levels=*/3,
CreateRunsWithLevelAndSize(
+
/*levels=*/{0, 1, 2},
+
/*sizes=*/{2, 6, 10})));
+ ASSERT_FALSE(pick);
+ // one sort run, not trigger
+ ASSERT_OK_AND_ASSIGN(pick, compaction.Pick(/*num_levels=*/3,
CreateRunsWithLevelAndSize(
+
/*levels=*/{3},
+
/*sizes=*/{5})));
+ ASSERT_FALSE(pick);
+}
+
+TEST_F(UniversalCompactionTest, TestNoOutputLevel0) {
+ UniversalCompaction compaction(/*max_size_amp=*/25, /*size_ratio=*/1,
+ /*num_run_compaction_trigger=*/3, nullptr,
nullptr);
+ ASSERT_OK_AND_ASSIGN(auto pick,
+ compaction.Pick(/*num_levels=*/3,
CreateRunsWithLevelAndSize(
+ /*levels=*/{0,
0, 1, 2},
+ /*sizes=*/{1,
1, 1, 50})));
+ ASSERT_TRUE(pick);
+ ASSERT_EQ(GetFileSizeVecFromCompactUnit(pick.value()),
std::vector<int64_t>({1, 1, 1}));
+
+ ASSERT_OK_AND_ASSIGN(pick, compaction.Pick(/*num_levels=*/3,
CreateRunsWithLevelAndSize(
+
/*levels=*/{0, 0, 1, 2},
+
/*sizes=*/{1, 2, 3, 50})));
+ ASSERT_TRUE(pick);
+ // 3 should be in the candidate, by size ratio after picking by file num
+ ASSERT_EQ(GetFileSizeVecFromCompactUnit(pick.value()),
std::vector<int64_t>({1, 2, 3}));
+}
+
+TEST_F(UniversalCompactionTest, TestExtremeCaseNoOutputLevel0) {
+ UniversalCompaction compaction(/*max_size_amp=*/200, /*size_ratio=*/1,
+ /*num_run_compaction_trigger=*/5, nullptr,
nullptr);
+ ASSERT_OK_AND_ASSIGN(
+ auto pick, compaction.Pick(/*num_levels=*/6,
CreateRunsWithLevelAndSize(
+ /*levels=*/{0, 0, 0,
0, 0},
+ /*sizes=*/{1, 1, 1,
1024, 1024 * 1024})));
+ ASSERT_TRUE(pick);
+ ASSERT_EQ(GetFileSizeVecFromCompactUnit(pick.value()),
+ std::vector<int64_t>({1, 1, 1, 1024, 1024 * 1024}));
+}
+
+TEST_F(UniversalCompactionTest, TestSizeAmplification) {
+ UniversalCompaction compaction(/*max_size_amp=*/25, /*size_ratio=*/0,
+ /*num_run_compaction_trigger=*/1, nullptr,
nullptr);
+
+ std::vector<int64_t> sizes = {1};
+ auto append_and_pick = [&](const std::vector<int64_t>& expected_sizes) {
+ sizes.insert(sizes.begin(), 1);
+ auto unit = compaction.PickForSizeAmp(3, CreateRunsWithSize(sizes));
+ if (unit) {
+ auto files = GetFileSizeVecFromCompactUnit(unit.value());
+ int64_t total_size = std::accumulate(files.begin(), files.end(),
0l);
+ sizes = {total_size};
+ }
+ ASSERT_EQ(sizes, expected_sizes);
+ };
+
+ append_and_pick({2});
+ append_and_pick({3});
+ append_and_pick({4});
+ append_and_pick({1, 4});
+ append_and_pick({6});
+ append_and_pick({1, 6});
+ append_and_pick({8});
+ append_and_pick({1, 8});
+ append_and_pick({1, 1, 8});
+ append_and_pick({11});
+ append_and_pick({1, 11});
+ append_and_pick({1, 1, 11});
+ append_and_pick({14});
+ append_and_pick({1, 14});
+ append_and_pick({1, 1, 14});
+ append_and_pick({1, 1, 1, 14});
+ append_and_pick({18});
+}
+
+TEST_F(UniversalCompactionTest, TestSizeRatio) {
+ UniversalCompaction compaction(/*max_size_amp=*/25, /*size_ratio=*/1,
+ /*num_run_compaction_trigger=*/5, nullptr,
nullptr);
+
+ std::vector<int64_t> sizes = {1, 1, 1, 1};
+ auto append_and_pick = [&](const std::vector<int64_t>& expected_sizes) {
+ sizes.insert(sizes.begin(), 1);
+ std::vector<int32_t> levels;
+ for (size_t i = 0; i < sizes.size(); i++) {
+ levels.push_back(static_cast<int64_t>(i));
+ }
+ ASSERT_OK_AND_ASSIGN(
+ auto unit, compaction.PickForSizeRatio(/*max_level=*/sizes.size(),
+
CreateRunsWithLevelAndSize(levels, sizes)));
+ if (unit) {
+ std::vector<int64_t> compact;
+ compact.reserve(unit->files.size());
+ for (const auto& file : unit->files) {
+ compact.push_back(file->file_size);
+ }
+ std::vector<int64_t> result = sizes;
+ for (int64_t size_val : compact) {
+ auto it = std::find(result.begin(), result.end(), size_val);
+ if (it != result.end()) {
+ result.erase(it);
+ }
+ }
+ int64_t sum = std::accumulate(compact.begin(), compact.end(), 0l);
+ result.insert(result.begin(), sum);
+ sizes = result;
+ }
+ ASSERT_EQ(sizes, expected_sizes);
+ };
+
+ append_and_pick({5});
+ append_and_pick({1, 5});
+ append_and_pick({1, 1, 5});
+ append_and_pick({1, 1, 1, 5});
+ append_and_pick({4, 5});
+ append_and_pick({1, 4, 5});
+ append_and_pick({1, 1, 4, 5});
+ append_and_pick({3, 4, 5});
+ append_and_pick({1, 3, 4, 5});
+ append_and_pick({2, 3, 4, 5});
+ append_and_pick({1, 2, 3, 4, 5});
+ append_and_pick({16});
+ append_and_pick({1, 16});
+ append_and_pick({1, 1, 16});
+ append_and_pick({1, 1, 1, 16});
+ append_and_pick({4, 16});
+ append_and_pick({1, 4, 16});
+ append_and_pick({1, 1, 4, 16});
+ append_and_pick({3, 4, 16});
+ append_and_pick({1, 3, 4, 16});
+ append_and_pick({2, 3, 4, 16});
+ append_and_pick({1, 2, 3, 4, 16});
+ append_and_pick({11, 16});
+}
+TEST_F(UniversalCompactionTest, TestSizeRatioThreshold) {
+ {
+ UniversalCompaction compaction(/*max_size_amp=*/25, /*size_ratio=*/10,
+ /*num_run_compaction_trigger=*/2,
nullptr, nullptr);
+ ASSERT_OK_AND_ASSIGN(
+ auto unit, compaction.PickForSizeRatio(
+ /*max_level=*/3,
+ CreateRunsWithLevelAndSize(/*levels=*/{0, 1, 2},
/*sizes=*/{8, 9, 10})));
+ ASSERT_FALSE(unit);
+ }
+ {
+ UniversalCompaction compaction(/*max_size_amp=*/25, /*size_ratio=*/20,
+ /*num_run_compaction_trigger=*/2,
nullptr, nullptr);
+ ASSERT_OK_AND_ASSIGN(
+ auto unit, compaction.PickForSizeRatio(
+ /*max_level=*/3,
+ CreateRunsWithLevelAndSize(/*levels=*/{0, 1, 2},
/*sizes=*/{8, 9, 10})));
+ ASSERT_TRUE(unit);
+ ASSERT_EQ(GetFileSizeVecFromCompactUnit(unit.value()),
std::vector<int64_t>({8, 9, 10}));
+ }
+}
+
+TEST_F(UniversalCompactionTest, TestLookup) {
+ auto universal =
+ std::make_shared<UniversalCompaction>(/*max_size_amp=*/25,
/*size_ratio=*/1,
+
/*num_run_compaction_trigger=*/3, nullptr, nullptr);
+ ForceUpLevel0Compaction compaction(universal,
/*max_compact_interval=*/std::nullopt);
+
+ // level 0 to max level
+ ASSERT_OK_AND_ASSIGN(auto unit,
+ compaction.Pick(/*num_levels=*/3,
CreateRunsWithSize({1, 2, 2, 2})));
+ ASSERT_TRUE(unit);
+ ASSERT_EQ(GetFileSizeVecFromCompactUnit(unit.value()),
std::vector<int64_t>({1, 2, 2, 2}));
+ ASSERT_EQ(unit.value().output_level, 2);
+
+ // level 0 force pick
+ ASSERT_OK_AND_ASSIGN(
+ unit, compaction.Pick(/*num_levels=*/3,
CreateRunsWithLevelAndSize(/*levels=*/{0, 1, 2},
+
/*sizes=*/{1, 2, 2})));
+ ASSERT_TRUE(unit);
+ ASSERT_EQ(GetFileSizeVecFromCompactUnit(unit.value()),
std::vector<int64_t>({1, 2, 2}));
+ ASSERT_EQ(unit.value().output_level, 2);
+
+ // level 0 to empty level
+ ASSERT_OK_AND_ASSIGN(
+ unit, compaction.Pick(/*num_levels=*/3,
+ CreateRunsWithLevelAndSize(/*levels=*/{0, 2},
/*sizes=*/{1, 2})));
+ ASSERT_TRUE(unit);
+ ASSERT_EQ(GetFileSizeVecFromCompactUnit(unit.value()),
std::vector<int64_t>({1}));
+ ASSERT_EQ(unit.value().output_level, 1);
+}
+
+TEST_F(UniversalCompactionTest, TestForcePickL0) {
+ int32_t max_compact_interval = 5;
+ auto universal =
+ std::make_shared<UniversalCompaction>(/*max_size_amp=*/25,
/*size_ratio=*/1,
+
/*num_run_compaction_trigger=*/5, nullptr, nullptr);
+ ForceUpLevel0Compaction compaction(universal, max_compact_interval);
+
+ // level 0 to max level
+ auto level0_to_max = CreateRunsWithSize({1, 2, 2, 2});
+ std::optional<CompactUnit> unit;
+ for (int32_t i = 1; i <= max_compact_interval; i++) {
+ // level 0 to max level triggered
+ ASSERT_OK_AND_ASSIGN(unit, compaction.Pick(/*num_levels=*/3,
level0_to_max));
+ if (i == max_compact_interval) {
+ ASSERT_TRUE(unit);
+ ASSERT_EQ(GetFileSizeVecFromCompactUnit(unit.value()),
+ std::vector<int64_t>({1, 2, 2, 2}));
+ ASSERT_EQ(unit.value().output_level, 2);
+ } else {
+ // compact skipped
+ ASSERT_FALSE(unit);
+ }
+ }
+
+ // level 0 force pick
+ auto level0_force_pick = CreateRunsWithLevelAndSize(/*levels=*/{0, 1, 2},
/*sizes=*/{2, 2, 2});
+ for (int32_t i = 1; i <= max_compact_interval; i++) {
+ ASSERT_OK_AND_ASSIGN(unit, compaction.Pick(/*num_levels=*/3,
level0_force_pick));
+ if (i == max_compact_interval) {
+ // level 0 force pick triggered
+ ASSERT_TRUE(unit);
+ ASSERT_EQ(GetFileSizeVecFromCompactUnit(unit.value()),
std::vector<int64_t>({2, 2, 2}));
+ ASSERT_EQ(unit.value().output_level, 2);
+ } else {
+ // compact skipped
+ ASSERT_FALSE(unit);
+ }
+ }
+
+ // level 0 to empty level
+ auto level0_to_empty = CreateRunsWithLevelAndSize(/*levels=*/{0, 2},
/*sizes=*/{1, 2});
+ for (int32_t i = 1; i <= max_compact_interval; i++) {
+ ASSERT_OK_AND_ASSIGN(unit, compaction.Pick(/*num_levels=*/3,
level0_to_empty));
+ if (i == max_compact_interval) {
+ // level 0 force pick triggered
+ ASSERT_TRUE(unit);
+ ASSERT_EQ(GetFileSizeVecFromCompactUnit(unit.value()),
std::vector<int64_t>({1}));
+ ASSERT_EQ(unit.value().output_level, 1);
+ } else {
+ // compact skipped
+ ASSERT_FALSE(unit);
+ }
+ }
+}
+} // namespace paimon::test
