This is an automated email from the ASF dual-hosted git repository.
luwei16 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 04aa5a11584 [fix](binlog) Persist row binlog config updates in cloud
mode (#68094)
04aa5a11584 is described below
commit 04aa5a115846bd7a5fb82bd1c64f7ab5816e8a3d
Author: Luwei <[email protected]>
AuthorDate: Fri Sep 18 01:09:08 2026 +0800
[fix](binlog) Persist row binlog config updates in cloud mode (#68094)
### What problem does this PR solve?
Issue Number: None
Related PR: None
Problem Summary: Updating ROW binlog retention properties in cloud mode
used the shared-nothing UPDATE_TABLET_META_INFO agent task. Cloud
backends do not register that task because tablet metadata is owned by
MetaService, so every backend rejected it and the FE timed out. Route
cloud binlog config changes through the existing update-tablet
MetaService API, persist the config, and refresh cached CloudTablet
metadata on backends.
### Release note
Fix updating ROW binlog retention properties in cloud mode.
### Check List (For Author)
- Test: Unit Test
- CloudSchemaChangeHandlerTest
- MetaServiceTest.UpdateTablet
- CloudTabletSyncMetaTest.TestSyncMetaMultipleProperties
- Behavior changed: Yes (ROW binlog property updates in cloud mode now
use the cloud metadata path.)
- Does this need documentation: No
---
be/src/cloud/cloud_tablet.cpp | 10 +++---
be/src/storage/binlog_config.h | 2 ++
be/src/storage/tablet/base_tablet.cpp | 4 +--
be/src/storage/tablet/base_tablet.h | 5 +++
be/test/cloud/cloud_tablet_test.cpp | 30 +++++++++++++++++
cloud/src/meta-service/meta_service.cpp | 2 ++
cloud/test/meta_service_test.cpp | 34 +++++++++++++++++++
.../cloud/alter/CloudSchemaChangeHandler.java | 18 ++++++++++
.../cloud/alter/CloudSchemaChangeHandlerTest.java | 39 ++++++++++++++++++++++
gensrc/proto/cloud.proto | 1 +
10 files changed, 139 insertions(+), 6 deletions(-)
diff --git a/be/src/cloud/cloud_tablet.cpp b/be/src/cloud/cloud_tablet.cpp
index aeed9dac426..e6a8afec2ff 100644
--- a/be/src/cloud/cloud_tablet.cpp
+++ b/be/src/cloud/cloud_tablet.cpp
@@ -846,8 +846,7 @@ Result<std::unique_ptr<RowsetWriter>>
CloudTablet::create_rowset_writer(
context.enable_unique_key_merge_on_write =
enable_unique_key_merge_on_write();
context.encrypt_algorithm = tablet_meta()->encryption_algorithm();
if (context.write_binlog_opt().enable) {
- context.write_binlog_opt().set_need_before(
- tablet_meta()->binlog_config().need_historical_value());
+
context.write_binlog_opt().set_need_before(binlog_config().need_historical_value());
}
context.inverted_index_storage_format =
tablet_meta()->inverted_index_storage_format();
context.persist_inverted_index_storage_format =
@@ -891,8 +890,7 @@ Result<std::unique_ptr<RowsetWriter>>
CloudTablet::create_transient_rowset_write
context.is_transient_rowset_writer = true;
if (rowset.rowset_meta() != nullptr &&
rowset.rowset_meta()->is_row_binlog()) {
context.write_binlog_opt().enable = true;
- context.write_binlog_opt().set_need_before(
- tablet_meta()->binlog_config().need_historical_value());
+
context.write_binlog_opt().set_need_before(binlog_config().need_historical_value());
}
context.rowset_id = rowset.rowset_id();
context.tablet_id = tablet_id();
@@ -1577,6 +1575,7 @@ Status CloudTablet::sync_meta() {
auto new_disable_auto_compaction =
tablet_meta->tablet_schema()->disable_auto_compaction();
auto new_vertical_compaction_num_columns_per_group =
tablet_meta->vertical_compaction_num_columns_per_group();
+ auto new_binlog_config = tablet_meta->binlog_config();
{
std::unique_lock wlock(_meta_lock);
@@ -1621,6 +1620,9 @@ Status CloudTablet::sync_meta() {
_tablet_meta->set_vertical_compaction_num_columns_per_group(
new_vertical_compaction_num_columns_per_group);
}
+ if (_tablet_meta->binlog_config() != new_binlog_config) {
+ _tablet_meta->set_binlog_config(new_binlog_config);
+ }
}
last_sync_tablet_meta_time_s = ::time(nullptr);
diff --git a/be/src/storage/binlog_config.h b/be/src/storage/binlog_config.h
index 696eb696322..76751389a43 100644
--- a/be/src/storage/binlog_config.h
+++ b/be/src/storage/binlog_config.h
@@ -46,6 +46,8 @@ public:
BinlogConfig& operator=(BinlogConfig&&) = default;
~BinlogConfig() = default;
+ bool operator==(const BinlogConfig&) const = default;
+
bool is_enable() const { return _enable; }
void set_enable(bool enable) { _enable = enable; }
diff --git a/be/src/storage/tablet/base_tablet.cpp
b/be/src/storage/tablet/base_tablet.cpp
index fcdd19416c1..c7ad7a6c639 100644
--- a/be/src/storage/tablet/base_tablet.cpp
+++ b/be/src/storage/tablet/base_tablet.cpp
@@ -1581,9 +1581,9 @@ Status BaseTablet::update_delete_bitmap(const
BaseTabletSPtr& self, TabletTxnInf
binlog_rs->rowset_meta()->is_row_binlog()) {
DCHECK(txn_info->attach_row_binlog.tablet != nullptr);
row_binlog_rowset = binlog_rs;
- const auto& binlog_tablet_meta =
txn_info->attach_row_binlog.tablet->tablet_meta();
build_row_binlog =
- is_partial_update ||
binlog_tablet_meta->binlog_config().need_historical_value();
+ is_partial_update ||
+
txn_info->attach_row_binlog.tablet->binlog_config().need_historical_value();
}
// rewrite conflict only when partial update or need before
diff --git a/be/src/storage/tablet/base_tablet.h
b/be/src/storage/tablet/base_tablet.h
index 5248be6cbac..1c156d56351 100644
--- a/be/src/storage/tablet/base_tablet.h
+++ b/be/src/storage/tablet/base_tablet.h
@@ -103,6 +103,11 @@ public:
// Property encapsulated in TabletMeta
const TabletMetaSharedPtr& tablet_meta() const { return _tablet_meta; }
+ BinlogConfig binlog_config() const {
+ std::shared_lock rlock(_meta_lock);
+ return _tablet_meta->binlog_config();
+ }
+
int32_t max_version_config();
// FIXME(plat1ko): It is not appropriate to expose this lock
diff --git a/be/test/cloud/cloud_tablet_test.cpp
b/be/test/cloud/cloud_tablet_test.cpp
index a9184ea69cb..0480c0a2510 100644
--- a/be/test/cloud/cloud_tablet_test.cpp
+++ b/be/test/cloud/cloud_tablet_test.cpp
@@ -24,6 +24,8 @@
#include <chrono>
#include <cstdint>
+#include <future>
+#include <mutex>
#include "cloud/cloud_meta_mgr.h"
#include "cloud/cloud_storage_engine.h"
@@ -1004,6 +1006,8 @@ TEST_F(CloudTabletSyncMetaTest,
TestSyncMetaMultipleProperties) {
mock_tablet_meta->set_time_series_compaction_empty_rowsets_threshold(9);
mock_tablet_meta->set_time_series_compaction_level_threshold(7);
mock_tablet_meta->set_vertical_compaction_num_columns_per_group(13);
+ mock_tablet_meta->set_binlog_config(
+ BinlogConfig(true, 3600, 4096, 7, BinlogFormatPB::ROW, true));
// Mock get_tablet_meta to return tablet_meta with updated properties
sp->set_call_back("CloudMetaMgr::get_tablet_meta",
[mock_tablet_meta](auto&& args) {
@@ -1027,11 +1031,37 @@ TEST_F(CloudTabletSyncMetaTest,
TestSyncMetaMultipleProperties) {
EXPECT_EQ(_tablet->tablet_meta()->time_series_compaction_empty_rowsets_threshold(),
9);
EXPECT_EQ(_tablet->tablet_meta()->time_series_compaction_level_threshold(), 7);
EXPECT_EQ(_tablet->tablet_meta()->vertical_compaction_num_columns_per_group(),
13);
+ auto binlog_config = _tablet->binlog_config();
+ EXPECT_TRUE(binlog_config.is_enable());
+ EXPECT_EQ(binlog_config.ttl_seconds(), 3600);
+ EXPECT_EQ(binlog_config.max_bytes(), 4096);
+ EXPECT_EQ(binlog_config.max_history_nums(), 7);
+ EXPECT_EQ(binlog_config.binlog_format(), BinlogFormatPB::ROW);
+ EXPECT_TRUE(binlog_config.need_historical_value());
sp->disable_processing();
sp->clear_all_call_backs();
}
+TEST_F(CloudTabletSyncMetaTest, TestBinlogConfigReadUsesMetaLock) {
+ std::unique_lock meta_lock(_tablet->get_header_lock());
+ std::promise<void> reader_started;
+ auto reader_started_future = reader_started.get_future();
+ auto read_future = std::async(std::launch::async, [&]() {
+ reader_started.set_value();
+ return _tablet->binlog_config();
+ });
+
+ auto reader_started_status = reader_started_future.wait_for(seconds(5));
+ auto read_blocked_status = read_future.wait_for(milliseconds(100));
+ meta_lock.unlock();
+
+ EXPECT_EQ(reader_started_status, std::future_status::ready);
+ EXPECT_EQ(read_blocked_status, std::future_status::timeout);
+ ASSERT_EQ(read_future.wait_for(seconds(5)), std::future_status::ready);
+ EXPECT_EQ(read_future.get(), _tablet->binlog_config());
+}
+
TEST_F(CloudTabletSyncMetaTest, TestSyncMetaSyncsTtlWithoutChangingInMemory) {
EXPECT_EQ(0, _tablet->tablet_meta()->ttl_seconds());
EXPECT_FALSE(_tablet->tablet_meta()->tablet_schema()->is_in_memory());
diff --git a/cloud/src/meta-service/meta_service.cpp
b/cloud/src/meta-service/meta_service.cpp
index 6ad68ed39c4..58f1a746e37 100644
--- a/cloud/src/meta-service/meta_service.cpp
+++ b/cloud/src/meta-service/meta_service.cpp
@@ -1204,6 +1204,8 @@ void
MetaServiceImpl::update_tablet(::google::protobuf::RpcController* controlle
tablet_meta.set_is_persistent(tablet_meta_info.is_persistent());
} else if (tablet_meta_info.has_ttl_seconds()) {
tablet_meta.set_ttl_seconds(tablet_meta_info.ttl_seconds());
+ } else if (tablet_meta_info.has_binlog_config()) {
+
tablet_meta.mutable_binlog_config()->CopyFrom(tablet_meta_info.binlog_config());
} else if (tablet_meta_info.has_compaction_policy()) {
tablet_meta.set_compaction_policy(tablet_meta_info.compaction_policy());
} else if
(tablet_meta_info.has_time_series_compaction_goal_size_mbytes()) {
diff --git a/cloud/test/meta_service_test.cpp b/cloud/test/meta_service_test.cpp
index ab703c9a529..57c604e8299 100644
--- a/cloud/test/meta_service_test.cpp
+++ b/cloud/test/meta_service_test.cpp
@@ -5360,6 +5360,40 @@ TEST(MetaServiceTest, UpdateTablet) {
ASSERT_EQ(resp.status().code(), MetaServiceCode::OK);
}
get_and_check_tablet_meta(tablet_id1, 300, true, true);
+ {
+ brpc::Controller cntl;
+ UpdateTabletRequest req;
+ UpdateTabletResponse resp;
+ req.set_cloud_unique_id(cloud_unique_id);
+ TabletMetaInfoPB* tablet_meta_info = req.add_tablet_meta_infos();
+ tablet_meta_info->set_tablet_id(tablet_id1);
+ auto* binlog_config = tablet_meta_info->mutable_binlog_config();
+ binlog_config->set_enable(true);
+ binlog_config->set_ttl_seconds(3600);
+ binlog_config->set_max_bytes(4096);
+ binlog_config->set_max_history_nums(7);
+ binlog_config->set_binlog_format(BinlogFormatPB::ROW);
+ binlog_config->set_need_historical_value(true);
+ meta_service->update_tablet(&cntl, &req, &resp, nullptr);
+ ASSERT_EQ(resp.status().code(), MetaServiceCode::OK);
+ }
+ {
+ brpc::Controller cntl;
+ GetTabletRequest req;
+ req.set_cloud_unique_id(cloud_unique_id);
+ req.set_tablet_id(tablet_id1);
+ GetTabletResponse resp;
+ meta_service->get_tablet(&cntl, &req, &resp, nullptr);
+ ASSERT_EQ(resp.status().code(), MetaServiceCode::OK);
+ ASSERT_TRUE(resp.tablet_meta().has_binlog_config());
+ const auto& binlog_config = resp.tablet_meta().binlog_config();
+ EXPECT_TRUE(binlog_config.enable());
+ EXPECT_EQ(binlog_config.ttl_seconds(), 3600);
+ EXPECT_EQ(binlog_config.max_bytes(), 4096);
+ EXPECT_EQ(binlog_config.max_history_nums(), 7);
+ EXPECT_EQ(binlog_config.binlog_format(), BinlogFormatPB::ROW);
+ EXPECT_TRUE(binlog_config.need_historical_value());
+ }
}
TEST(MetaServiceTest, GetTabletStatsTest) {
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/cloud/alter/CloudSchemaChangeHandler.java
b/fe/fe-core/src/main/java/org/apache/doris/cloud/alter/CloudSchemaChangeHandler.java
index fa861cdcec6..ed6008c8391 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/cloud/alter/CloudSchemaChangeHandler.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/cloud/alter/CloudSchemaChangeHandler.java
@@ -18,6 +18,7 @@
package org.apache.doris.cloud.alter;
import org.apache.doris.alter.SchemaChangeHandler;
+import org.apache.doris.catalog.BinlogConfig;
import org.apache.doris.catalog.Database;
import org.apache.doris.catalog.Env;
import org.apache.doris.catalog.KeysType;
@@ -426,6 +427,18 @@ public class CloudSchemaChangeHandler extends
SchemaChangeHandler {
DynamicPartitionUtil.registerOrRemoveDynamicPartitionTable(db.getId(),
olapTable, false);
}
+ @Override
+ public void updatePartitionProperties(Database db, String tableName,
String partitionName,
+ long storagePolicyId, int isInMemory, BinlogConfig binlogConfig,
String compactionPolicy,
+ Map<String, Long> timeSeriesCompactionConfig, int
skipWriteIndexOnLoad,
+ int disableAutoCompaction, int
verticalCompactionNumColumnsPerGroup) throws UserException {
+ Preconditions.checkNotNull(binlogConfig);
+ UpdatePartitionMetaParam param = new UpdatePartitionMetaParam();
+ param.binlogConfig = binlogConfig;
+ param.type = UpdatePartitionMetaParam.TabletMetaType.BINLOG_CONFIG;
+ updateCloudPartitionMeta(db, tableName, partitionName, param);
+ }
+
private static class UpdatePartitionMetaParam {
public enum TabletMetaType {
INMEMORY,
@@ -443,6 +456,7 @@ public class CloudSchemaChangeHandler extends
SchemaChangeHandler {
DISABLE_AUTO_COMPACTION,
ENABLE_MOW_LIGHT_DELETE,
VERTICAL_COMPACTION_NUM_COLUMNS_PER_GROUP,
+ BINLOG_CONFIG,
}
TabletMetaType type;
@@ -461,6 +475,7 @@ public class CloudSchemaChangeHandler extends
SchemaChangeHandler {
boolean disableAutoCompaction = false;
boolean enableMowLightDelete = false;
int verticalCompactionNumColumnsPerGroup = 5;
+ BinlogConfig binlogConfig;
}
public void updateCloudPartitionMeta(Database db,
@@ -550,6 +565,9 @@ public class CloudSchemaChangeHandler extends
SchemaChangeHandler {
infoBuilder.setVerticalCompactionNumColumnsPerGroup(
param.verticalCompactionNumColumnsPerGroup);
break;
+ case BINLOG_CONFIG:
+
infoBuilder.setBinlogConfig(param.binlogConfig.toProtobuf());
+ break;
default:
throw new UserException("Unknown TabletMetaType");
}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/cloud/alter/CloudSchemaChangeHandlerTest.java
b/fe/fe-core/src/test/java/org/apache/doris/cloud/alter/CloudSchemaChangeHandlerTest.java
index 5d1fcca0bfe..c5037746aec 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/cloud/alter/CloudSchemaChangeHandlerTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/cloud/alter/CloudSchemaChangeHandlerTest.java
@@ -17,6 +17,7 @@
package org.apache.doris.cloud.alter;
+import org.apache.doris.catalog.BinlogConfig;
import org.apache.doris.catalog.Database;
import org.apache.doris.catalog.Env;
import org.apache.doris.catalog.KeysType;
@@ -155,6 +156,44 @@ public class CloudSchemaChangeHandlerTest {
Assertions.assertEquals(Arrays.asList(103L),
syncRequests.get(3).getTabletIdsList());
}
+ @Test
+ public void testUpdateBinlogConfigUsesCloudTabletMeta() throws Exception {
+ CloudSchemaChangeHandler handler = new CloudSchemaChangeHandler();
+ Database db = createMockDatabaseWithThreeTablets();
+ BinlogConfig binlogConfig = new BinlogConfig(true, 3600L, 4096L, 7L,
+ BinlogConfig.BinlogFormat.ROW, true);
+
+ MetaServiceProxy metaServiceProxy =
Mockito.mock(MetaServiceProxy.class);
+
Mockito.when(metaServiceProxy.updateTablet(Mockito.any())).thenReturn(okUpdateTabletResponse());
+ Config.enable_debug_points = true;
+
DebugPointUtil.addDebugPoint("CloudSchemaChangeHandler.notifyBackendsToSyncTabletMeta.skip");
+
+ try (MockedStatic<MetaServiceProxy> metaProxyMock =
Mockito.mockStatic(MetaServiceProxy.class)) {
+
metaProxyMock.when(MetaServiceProxy::getInstance).thenReturn(metaServiceProxy);
+ handler.updatePartitionProperties(db, "tbl", "p1", -1, -1,
binlogConfig,
+ null, null, -1, -1, -1);
+ }
+
+ ArgumentCaptor<Cloud.UpdateTabletRequest> updateCaptor =
+ ArgumentCaptor.forClass(Cloud.UpdateTabletRequest.class);
+ Mockito.verify(metaServiceProxy,
Mockito.times(2)).updateTablet(updateCaptor.capture());
+ List<Cloud.TabletMetaInfoPB> tabletMetaInfos =
updateCaptor.getAllValues().stream()
+ .flatMap(request -> request.getTabletMetaInfosList().stream())
+ .collect(Collectors.toList());
+ Assertions.assertEquals(Arrays.asList(101L, 102L, 103L),
tabletMetaInfos.stream()
+
.map(Cloud.TabletMetaInfoPB::getTabletId).collect(Collectors.toList()));
+ for (Cloud.TabletMetaInfoPB tabletMetaInfo : tabletMetaInfos) {
+ Assertions.assertTrue(tabletMetaInfo.hasBinlogConfig());
+
Assertions.assertTrue(tabletMetaInfo.getBinlogConfig().getEnable());
+ Assertions.assertEquals(3600L,
tabletMetaInfo.getBinlogConfig().getTtlSeconds());
+ Assertions.assertEquals(4096L,
tabletMetaInfo.getBinlogConfig().getMaxBytes());
+ Assertions.assertEquals(7L,
tabletMetaInfo.getBinlogConfig().getMaxHistoryNums());
+
Assertions.assertEquals(org.apache.doris.proto.OlapFile.BinlogFormatPB.ROW,
+ tabletMetaInfo.getBinlogConfig().getBinlogFormat());
+
Assertions.assertTrue(tabletMetaInfo.getBinlogConfig().getNeedHistoricalValue());
+ }
+ }
+
@Test
public void testUpdateCompactionPolicyExcludesRowBinlogTablets() throws
Exception {
CloudSchemaChangeHandler handler = new CloudSchemaChangeHandler();
diff --git a/gensrc/proto/cloud.proto b/gensrc/proto/cloud.proto
index c1657361a95..7d45d722226 100644
--- a/gensrc/proto/cloud.proto
+++ b/gensrc/proto/cloud.proto
@@ -736,6 +736,7 @@ message TabletMetaInfoPB { // For update tablet meta
optional bool enable_mow_light_delete = 14;
optional int32 vertical_compaction_num_columns_per_group = 15;
optional string group_commit_mode = 16;
+ optional doris.BinlogConfigPB binlog_config = 17;
}
message TabletCompactionJobPB {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
