This is an automated email from the ASF dual-hosted git repository.
gavinchou pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 21160d78b42 [fix](binlog) Decouple row binlog compaction from CCR
binlog config (#68057)
21160d78b42 is described below
commit 21160d78b42df3e88e5a8ba1bb5700201a0356c3
Author: Luwei <[email protected]>
AuthorDate: Thu Sep 17 18:42:36 2026 +0800
[fix](binlog) Decouple row binlog compaction from CCR binlog config (#68057)
Problem Summary: Row Binlog compaction scheduling and execution were
incorrectly gated by the CCR binlog feature switch. This prevented Cloud
and non-Cloud Row Binlog compaction when CCR was disabled. Decouple Row
Binlog compaction from the CCR switch while preserving
automatic-compaction, memory, and tablet-role checks. CCR download,
ingest, and enablement paths remain unchanged.
### Release note
Fix Row Binlog compaction scheduling and execution when CCR binlog is
disabled.
---
be/src/cloud/cloud_storage_engine.cpp | 2 +-
be/src/storage/olap_server.cpp | 2 +-
be/src/storage/tablet/tablet.cpp | 3 +-
be/test/cloud/cloud_compaction_test.cpp | 50 ++++++++++++++++++++++
.../storage/compaction/compaction_task_test.cpp | 40 +++++++++++++++++
be/test/storage/tablet/tablet_test.cpp | 26 +++++++++++
6 files changed, 119 insertions(+), 4 deletions(-)
diff --git a/be/src/cloud/cloud_storage_engine.cpp
b/be/src/cloud/cloud_storage_engine.cpp
index 6a2d46469d8..b9d88260ee4 100644
--- a/be/src/cloud/cloud_storage_engine.cpp
+++ b/be/src/cloud/cloud_storage_engine.cpp
@@ -681,7 +681,7 @@ void
CloudStorageEngine::_binlog_compaction_tasks_producer_callback() {
int64_t interval = config::generate_compaction_tasks_interval_ms;
do {
int64_t cur_time = UnixMillis();
- if (config::enable_feature_binlog && !config::disable_auto_compaction)
{
+ if (!config::disable_auto_compaction) {
Status st = _adjust_compaction_thread_num();
if (!st.ok()) {
break;
diff --git a/be/src/storage/olap_server.cpp b/be/src/storage/olap_server.cpp
index 2dc715cd327..7bcf615cf04 100644
--- a/be/src/storage/olap_server.cpp
+++ b/be/src/storage/olap_server.cpp
@@ -776,7 +776,7 @@ void
StorageEngine::_binlog_compaction_tasks_producer_callback() {
int64_t interval = config::generate_compaction_tasks_interval_ms;
do {
int64_t cur_time = UnixMillis();
- if (config::enable_feature_binlog && !config::disable_auto_compaction
&&
+ if (!config::disable_auto_compaction &&
(!config::enable_compaction_pause_on_high_memory ||
!GlobalMemoryArbitrator::is_exceed_soft_mem_limit(GB_EXCHANGE_BYTE))) {
_adjust_compaction_thread_num();
diff --git a/be/src/storage/tablet/tablet.cpp b/be/src/storage/tablet/tablet.cpp
index 9da3e99b512..9a549b84ac1 100644
--- a/be/src/storage/tablet/tablet.cpp
+++ b/be/src/storage/tablet/tablet.cpp
@@ -1019,8 +1019,7 @@ bool Tablet::can_do_compaction(size_t path_hash,
CompactionType compaction_type)
return false;
}
- if (compaction_type == CompactionType::CUMU_BINLOG_COMPACTION &&
- (!config::enable_feature_binlog || !is_row_binlog_tablet())) {
+ if (compaction_type == CompactionType::CUMU_BINLOG_COMPACTION &&
!is_row_binlog_tablet()) {
return false;
}
diff --git a/be/test/cloud/cloud_compaction_test.cpp
b/be/test/cloud/cloud_compaction_test.cpp
index 683e72b3064..fdca1efa4ae 100644
--- a/be/test/cloud/cloud_compaction_test.cpp
+++ b/be/test/cloud/cloud_compaction_test.cpp
@@ -389,6 +389,56 @@ TEST_F(CloudCompactionTest,
generate_cloud_binlog_compaction_tasks_updates_only_
ASSERT_EQ(metrics->tablet_time_series_max_compaction_score->value(), 103);
}
+TEST_F(CloudCompactionTest, binlog_compaction_producer_ignores_ccr_switch) {
+ const bool old_enable_feature_binlog = config::enable_feature_binlog;
+ const bool old_disable_auto_compaction = config::disable_auto_compaction;
+ const int32_t old_binlog_compaction_task_num_per_disk =
+ config::binlog_compaction_task_num_per_disk;
+ Defer restore_config {[&]() {
+ config::enable_feature_binlog = old_enable_feature_binlog;
+ config::disable_auto_compaction = old_disable_auto_compaction;
+ config::binlog_compaction_task_num_per_disk =
old_binlog_compaction_task_num_per_disk;
+ }};
+
+ ASSERT_TRUE(ThreadPoolBuilder("BaseCompactionTaskThreadPoolTest")
+ .set_min_threads(1)
+ .set_max_threads(1)
+ .build(&_engine._base_compaction_thread_pool)
+ .ok());
+ ASSERT_TRUE(ThreadPoolBuilder("CumuCompactionTaskThreadPoolTest")
+ .set_min_threads(1)
+ .set_max_threads(1)
+ .build(&_engine._cumu_compaction_thread_pool)
+ .ok());
+ ASSERT_TRUE(ThreadPoolBuilder("BinlogCompactionTaskThreadPoolTest")
+ .set_min_threads(1)
+ .set_max_threads(1)
+ .build(&_engine._binlog_compaction_thread_pool)
+ .ok());
+
+ auto binlog_meta = std::make_shared<TabletMeta>(*_tablet_meta);
+ binlog_meta->_tablet_id = 11004;
+ binlog_meta->set_tablet_role(TabletRolePB::TABLET_ROLE_ROW_BINLOG);
+ auto binlog_tablet = std::make_shared<CloudTablet>(_engine, binlog_meta);
+
binlog_tablet->tablet_meta()->tablet_schema()->set_disable_auto_compaction(false);
+ binlog_tablet->_approximate_cumu_num_deltas = 7;
+ _engine.tablet_mgr().put_tablet_for_UT(binlog_tablet);
+
+ config::enable_feature_binlog = false;
+ config::disable_auto_compaction = false;
+ config::binlog_compaction_task_num_per_disk = 0;
+ auto* metric =
DorisMetrics::instance()->tablet_binlog_max_compaction_score;
+ metric->set_value(0);
+ _engine._stop_background_threads_latch.count_down();
+ _engine._binlog_compaction_tasks_producer_callback();
+ EXPECT_EQ(metric->value(), 7);
+
+ config::disable_auto_compaction = true;
+ metric->set_value(11);
+ _engine._binlog_compaction_tasks_producer_callback();
+ EXPECT_EQ(metric->value(), 11);
+}
+
TEST_F(CloudCompactionTest,
generate_cloud_compaction_tasks_clears_metrics_without_tablets) {
auto* metrics = DorisMetrics::instance();
metrics->tablet_cumulative_max_compaction_score->set_value(101);
diff --git a/be/test/storage/compaction/compaction_task_test.cpp
b/be/test/storage/compaction/compaction_task_test.cpp
index 85ae15a7357..3f2c1d29c99 100644
--- a/be/test/storage/compaction/compaction_task_test.cpp
+++ b/be/test/storage/compaction/compaction_task_test.cpp
@@ -130,6 +130,46 @@ TEST_F(CompactionTaskTest, TestSubmitCompactionTask) {
EXPECT_EQ(executing_task_num, 2);
}
+TEST_F(CompactionTaskTest, BinlogCompactionProducerIgnoresCcrSwitch) {
+ const bool old_enable_feature_binlog = config::enable_feature_binlog;
+ const bool old_disable_auto_compaction = config::disable_auto_compaction;
+ const bool old_enable_compaction_pause_on_high_memory =
+ config::enable_compaction_pause_on_high_memory;
+ Defer restore_config {[&]() {
+ config::enable_feature_binlog = old_enable_feature_binlog;
+ config::disable_auto_compaction = old_disable_auto_compaction;
+ config::enable_compaction_pause_on_high_memory =
old_enable_compaction_pause_on_high_memory;
+ }};
+
+ auto* sp = SyncPoint::get_instance();
+ sp->enable_processing();
+ int generation_count = 0;
+
sp->set_call_back("StorageEngine::_adjust_compaction_thread_num.return_void",
+ [](auto&& args) { *try_any_cast<bool*>(args.back()) =
true; });
+ sp->set_call_back("olap_server::_generate_compaction_tasks.return_empty",
+ [&generation_count](auto&& values) {
+ ++generation_count;
+ auto* ret =
+
try_any_cast_ret<std::vector<TabletCompactionContext>>(values);
+ ret->second = true;
+ });
+ Defer clear_sync_points {[&]() {
+ sp->clear_all_call_backs();
+ sp->disable_processing();
+ }};
+
+ config::enable_feature_binlog = false;
+ config::disable_auto_compaction = false;
+ config::enable_compaction_pause_on_high_memory = false;
+ _storage_engine->_stop_background_threads_latch.count_down();
+ _storage_engine->_binlog_compaction_tasks_producer_callback();
+ EXPECT_EQ(generation_count, 1);
+
+ config::disable_auto_compaction = true;
+ _storage_engine->_binlog_compaction_tasks_producer_callback();
+ EXPECT_EQ(generation_count, 1);
+}
+
TEST_F(CompactionTaskTest, TestAutoSetCompactionIncreaseTaskNum) {
auto st = ThreadPoolBuilder("BaseCompactionTaskThreadPool")
.set_min_threads(2)
diff --git a/be/test/storage/tablet/tablet_test.cpp
b/be/test/storage/tablet/tablet_test.cpp
index ea4719e31c9..21c6d91afba 100644
--- a/be/test/storage/tablet/tablet_test.cpp
+++ b/be/test/storage/tablet/tablet_test.cpp
@@ -26,6 +26,7 @@
#include <memory>
+#include "common/config.h"
#include "gtest/gtest_pred_impl.h"
#include "io/fs/local_file_system.h"
#include "json2pb/json_to_pb.h"
@@ -39,6 +40,7 @@
#include "storage/tablet/tablet_meta.h"
#include "storage/utils.h"
#include "testutil/mock_rowset.h"
+#include "util/defer_op.h"
#include "util/time.h"
#include "util/uid_util.h"
@@ -269,6 +271,30 @@ TEST_F(TestTablet, delete_expired_stale_rowset) {
_tablet.reset();
}
+TEST_F(TestTablet, RowBinlogCompactionDoesNotDependOnCcrSwitch) {
+ const bool old_enable_feature_binlog = config::enable_feature_binlog;
+ Defer restore_config {[&]() { config::enable_feature_binlog =
old_enable_feature_binlog; }};
+ config::enable_feature_binlog = false;
+
+ ASSERT_TRUE(_data_dir->init().ok());
+ _tablet_meta->set_tablet_role(TabletRolePB::TABLET_ROLE_ROW_BINLOG);
+ TabletSharedPtr row_binlog_tablet(
+ new Tablet(*k_engine, _tablet_meta, _data_dir.get(),
CUMULATIVE_SIZE_BASED_POLICY));
+ ASSERT_TRUE(row_binlog_tablet->init().ok());
+ EXPECT_TRUE(row_binlog_tablet->can_do_compaction(_data_dir->path_hash(),
+
CompactionType::CUMU_BINLOG_COMPACTION));
+ EXPECT_FALSE(row_binlog_tablet->can_do_compaction(_data_dir->path_hash(),
+
CompactionType::CUMULATIVE_COMPACTION));
+
+ auto data_tablet_meta = new_tablet_meta(TTabletSchema());
+ data_tablet_meta->set_tablet_role(TabletRolePB::TABLET_ROLE_DATA);
+ TabletSharedPtr data_tablet(
+ new Tablet(*k_engine, data_tablet_meta, _data_dir.get(),
CUMULATIVE_SIZE_BASED_POLICY));
+ ASSERT_TRUE(data_tablet->init().ok());
+ EXPECT_FALSE(data_tablet->can_do_compaction(_data_dir->path_hash(),
+
CompactionType::CUMU_BINLOG_COMPACTION));
+}
+
TEST_F(TestTablet, pad_rowset) {
std::vector<RowsetMetaSharedPtr> rs_metas;
auto ptr1 = std::make_shared<RowsetMeta>();
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
