This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-4.1
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.1 by this push:
new 22d7dd9e493 branch-4.1: [enhance](cloud) proactively sync tablet meta
after alter #61585 (#66798)
22d7dd9e493 is described below
commit 22d7dd9e4934d8b7a9a5abb3b5f18cae87d503c0
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Thu Nov 5 09:20:42 2026 +0800
branch-4.1: [enhance](cloud) proactively sync tablet meta after alter
#61585 (#66798)
Cherry-picked from #61585
---------
Co-authored-by: Luwei <[email protected]>
---
be/src/cloud/cloud_internal_service.cpp | 77 ++++
be/src/cloud/cloud_internal_service.h | 8 +
be/src/cloud/cloud_tablet.cpp | 95 ++--
be/src/cloud/cloud_tablet_mgr.cpp | 4 +
be/src/cloud/cloud_tablet_mgr.h | 2 +
be/src/service/internal_service.cpp | 9 +
be/src/service/internal_service.h | 4 +
be/test/cloud/cloud_internal_service_test.cpp | 180 ++++++++
be/test/cloud/cloud_tablet_mgr_test.cpp | 32 ++
be/test/cloud/cloud_tablet_test.cpp | 63 ++-
.../cloud/alter/CloudSchemaChangeHandler.java | 83 ++++
.../org/apache/doris/rpc/BackendServiceClient.java | 5 +
.../org/apache/doris/rpc/BackendServiceProxy.java | 12 +
.../cloud/alter/CloudSchemaChangeHandlerTest.java | 478 +++++++++++++++++++++
.../apache/doris/rpc/BackendServiceClientTest.java | 30 ++
.../apache/doris/rpc/BackendServiceProxyTest.java | 54 +++
gensrc/proto/internal_service.proto | 12 +
.../test_cloud_sync_tablet_meta_after_alter.groovy | 284 ++++++++++++
18 files changed, 1384 insertions(+), 48 deletions(-)
diff --git a/be/src/cloud/cloud_internal_service.cpp
b/be/src/cloud/cloud_internal_service.cpp
index 6ad86dabfaf..ab5c51d84d6 100644
--- a/be/src/cloud/cloud_internal_service.cpp
+++ b/be/src/cloud/cloud_internal_service.cpp
@@ -103,6 +103,67 @@ bvar::Adder<int64_t>
g_peer_server_fill_rejected("peer_server_fill_rejected");
bvar::LatencyRecorder g_peer_server_fill_latency("peer_server_fill_latency");
bvar::LatencyRecorder
g_cloud_internal_service_get_file_cache_meta_by_tablet_id_latency(
"cloud_internal_service_get_file_cache_meta_by_tablet_id_latency");
+bvar::Adder<int64_t> g_cloud_sync_tablet_meta_requests_total(
+ "cloud_sync_tablet_meta_requests_total");
+bvar::Adder<int64_t>
g_cloud_sync_tablet_meta_synced_total("cloud_sync_tablet_meta_synced_total");
+bvar::Adder<int64_t>
g_cloud_sync_tablet_meta_skipped_total("cloud_sync_tablet_meta_skipped_total");
+bvar::Adder<int64_t>
g_cloud_sync_tablet_meta_failed_total("cloud_sync_tablet_meta_failed_total");
+
+namespace {
+
+void submit_sync_tablet_meta(CloudStorageEngine& engine, FifoThreadPool&
work_pool,
+ const PSyncTabletMetaRequest* request,
+ PSyncTabletMetaResponse* response,
google::protobuf::Closure* done) {
+ auto start_time = std::chrono::steady_clock::now();
+ bool ret = work_pool.try_offer([engine = &engine, request, response, done,
start_time]() {
+ brpc::ClosureGuard closure_guard(done);
+ LOG(INFO) << "begin to sync tablet meta, request=" <<
request->ShortDebugString();
+ int64_t synced = 0;
+ int64_t skipped = 0;
+ int64_t failed = 0;
+ g_cloud_sync_tablet_meta_requests_total << 1;
+ for (const auto tablet_id : request->tablet_ids()) {
+ auto tablet = engine->tablet_mgr().get_tablet_if_cached(tablet_id);
+ if (!tablet) {
+ ++skipped;
+ continue;
+ }
+ auto st = tablet->sync_meta();
+ if (!st.ok()) {
+ ++failed;
+ LOG(WARNING) << "failed to sync tablet meta from cloud meta
service, tablet="
+ << tablet_id << ", err=" << st;
+ continue;
+ }
+ ++synced;
+ }
+ g_cloud_sync_tablet_meta_synced_total << synced;
+ g_cloud_sync_tablet_meta_skipped_total << skipped;
+ g_cloud_sync_tablet_meta_failed_total << failed;
+ response->set_synced_tablets(synced);
+ response->set_skipped_tablets(skipped);
+ response->set_failed_tablets(failed);
+ Status::OK().to_protobuf(response->mutable_status());
+ auto cost_ms = std::chrono::duration_cast<std::chrono::milliseconds>(
+ std::chrono::steady_clock::now() - start_time)
+ .count();
+ LOG(INFO) << "finish to sync tablet meta, request=" <<
request->ShortDebugString()
+ << ", response=" << response->ShortDebugString() << ",
cost_ms=" << cost_ms;
+ });
+ if (!ret) {
+ brpc::ClosureGuard closure_guard(done);
+ Status::InternalError("failed to offer sync_tablet_meta request to
work pool")
+ .to_protobuf(response->mutable_status());
+ auto cost_ms = std::chrono::duration_cast<std::chrono::milliseconds>(
+ std::chrono::steady_clock::now() - start_time)
+ .count();
+ LOG(WARNING) << "failed to offer sync_tablet_meta request to work
pool, request="
+ << request->ShortDebugString() << ", response=" <<
response->ShortDebugString()
+ << ", cost_ms=" << cost_ms;
+ }
+}
+
+} // namespace
// Concurrency guard for server-side S3 pull-through fills.
static std::atomic<int32_t> g_active_server_fills {0};
@@ -115,6 +176,22 @@
CloudInternalServiceImpl::CloudInternalServiceImpl(CloudStorageEngine& engine, E
CloudInternalServiceImpl::~CloudInternalServiceImpl() = default;
+void
CloudInternalServiceImpl::sync_tablet_meta(google::protobuf::RpcController*
controller,
+ const PSyncTabletMetaRequest*
request,
+ PSyncTabletMetaResponse*
response,
+ google::protobuf::Closure*
done) {
+ submit_sync_tablet_meta(_engine, _light_work_pool, request, response,
done);
+}
+
+#ifdef BE_TEST
+void test_submit_sync_tablet_meta(CloudStorageEngine& engine, FifoThreadPool&
work_pool,
+ const PSyncTabletMetaRequest* request,
+ PSyncTabletMetaResponse* response,
+ google::protobuf::Closure* done) {
+ submit_sync_tablet_meta(engine, work_pool, request, response, done);
+}
+#endif
+
void
CloudInternalServiceImpl::alter_vault_sync(google::protobuf::RpcController*
controller,
const
doris::PAlterVaultSyncRequest* request,
PAlterVaultSyncResponse*
response,
diff --git a/be/src/cloud/cloud_internal_service.h
b/be/src/cloud/cloud_internal_service.h
index 75fad0ba0ce..524ce14d275 100644
--- a/be/src/cloud/cloud_internal_service.h
+++ b/be/src/cloud/cloud_internal_service.h
@@ -35,6 +35,10 @@ Status test_handle_peer_file_cache_block_request(const
PFetchPeerDataRequest* re
brpc::Controller* cntl);
bool test_try_reject_if_queue_timed_out(std::chrono::steady_clock::time_point
enqueue_ts,
PFetchPeerDataResponse* response);
+void test_submit_sync_tablet_meta(CloudStorageEngine& engine, FifoThreadPool&
work_pool,
+ const PSyncTabletMetaRequest* request,
+ PSyncTabletMetaResponse* response,
+ google::protobuf::Closure* done);
#endif
class CloudInternalServiceImpl final : public PInternalService {
@@ -48,6 +52,10 @@ public:
PAlterVaultSyncResponse* response,
google::protobuf::Closure* done) override;
+ void sync_tablet_meta(google::protobuf::RpcController* controller,
+ const PSyncTabletMetaRequest* request,
PSyncTabletMetaResponse* response,
+ google::protobuf::Closure* done) override;
+
// Get messages (filename, offset, size) about the tablet data in cache
void get_file_cache_meta_by_tablet_id(google::protobuf::RpcController*
controller,
const PGetFileCacheMetaRequest*
request,
diff --git a/be/src/cloud/cloud_tablet.cpp b/be/src/cloud/cloud_tablet.cpp
index ec67ac5bc18..52010e5d92c 100644
--- a/be/src/cloud/cloud_tablet.cpp
+++ b/be/src/cloud/cloud_tablet.cpp
@@ -959,6 +959,12 @@ void CloudTablet::get_compaction_status(std::string*
json_result) {
// get snapshot version path json_doc
_timestamped_version_tracker.get_stale_version_path_json_doc(path_arr);
+ rapidjson::Value compaction_policy_value;
+ auto compaction_policy = _tablet_meta->compaction_policy();
+ compaction_policy_value.SetString(compaction_policy.c_str(),
+
cast_set<uint32_t>(compaction_policy.length()),
+ root.GetAllocator());
+ root.AddMember("compaction policy", compaction_policy_value,
root.GetAllocator());
root.AddMember("cumulative point", _cumulative_point.load(),
root.GetAllocator());
rapidjson::Value cumu_value;
std::string format_str =
ToStringFromUnixMillis(_last_cumu_compaction_failure_millis.load());
@@ -1490,9 +1496,9 @@ void CloudTablet::agg_delete_bitmap_for_compaction(
}
Status CloudTablet::sync_meta() {
- if (!config::enable_file_cache) {
- return Status::OK();
- }
+ // Keep tablet metadata refresh serialized with rowset sync and apply
local metadata updates
+ // under _meta_lock. The lock order must stay _sync_meta_lock ->
_meta_lock.
+ std::unique_lock lock(_sync_meta_lock);
TabletMetaSharedPtr tablet_meta;
auto st = _engine.meta_mgr().get_tablet_meta(tablet_id(), &tablet_meta);
@@ -1504,57 +1510,64 @@ Status CloudTablet::sync_meta() {
}
auto new_compaction_policy = tablet_meta->compaction_policy();
- if (_tablet_meta->compaction_policy() != new_compaction_policy) {
- _tablet_meta->set_compaction_policy(new_compaction_policy);
- }
+ auto new_ttl_seconds = tablet_meta->ttl_seconds();
auto new_time_series_compaction_goal_size_mbytes =
tablet_meta->time_series_compaction_goal_size_mbytes();
- if (_tablet_meta->time_series_compaction_goal_size_mbytes() !=
- new_time_series_compaction_goal_size_mbytes) {
- _tablet_meta->set_time_series_compaction_goal_size_mbytes(
- new_time_series_compaction_goal_size_mbytes);
- }
auto new_time_series_compaction_file_count_threshold =
tablet_meta->time_series_compaction_file_count_threshold();
- if (_tablet_meta->time_series_compaction_file_count_threshold() !=
- new_time_series_compaction_file_count_threshold) {
- _tablet_meta->set_time_series_compaction_file_count_threshold(
- new_time_series_compaction_file_count_threshold);
- }
auto new_time_series_compaction_time_threshold_seconds =
tablet_meta->time_series_compaction_time_threshold_seconds();
- if (_tablet_meta->time_series_compaction_time_threshold_seconds() !=
- new_time_series_compaction_time_threshold_seconds) {
- _tablet_meta->set_time_series_compaction_time_threshold_seconds(
- new_time_series_compaction_time_threshold_seconds);
- }
auto new_time_series_compaction_empty_rowsets_threshold =
tablet_meta->time_series_compaction_empty_rowsets_threshold();
- if (_tablet_meta->time_series_compaction_empty_rowsets_threshold() !=
- new_time_series_compaction_empty_rowsets_threshold) {
- _tablet_meta->set_time_series_compaction_empty_rowsets_threshold(
- new_time_series_compaction_empty_rowsets_threshold);
- }
auto new_time_series_compaction_level_threshold =
tablet_meta->time_series_compaction_level_threshold();
- if (_tablet_meta->time_series_compaction_level_threshold() !=
- new_time_series_compaction_level_threshold) {
- _tablet_meta->set_time_series_compaction_level_threshold(
- new_time_series_compaction_level_threshold);
- }
- // Sync disable_auto_compaction (stored in tablet_schema)
auto new_disable_auto_compaction =
tablet_meta->tablet_schema()->disable_auto_compaction();
- if (_tablet_meta->tablet_schema()->disable_auto_compaction() !=
new_disable_auto_compaction) {
- _tablet_meta->mutable_tablet_schema()->set_disable_auto_compaction(
- new_disable_auto_compaction);
- }
- // Sync vertical_compaction_num_columns_per_group
auto new_vertical_compaction_num_columns_per_group =
tablet_meta->vertical_compaction_num_columns_per_group();
- if (_tablet_meta->vertical_compaction_num_columns_per_group() !=
- new_vertical_compaction_num_columns_per_group) {
- _tablet_meta->set_vertical_compaction_num_columns_per_group(
- new_vertical_compaction_num_columns_per_group);
+
+ {
+ std::unique_lock wlock(_meta_lock);
+ if (_tablet_meta->compaction_policy() != new_compaction_policy) {
+ _tablet_meta->set_compaction_policy(new_compaction_policy);
+ }
+ if (_tablet_meta->ttl_seconds() != new_ttl_seconds) {
+ _tablet_meta->set_ttl_seconds(new_ttl_seconds);
+ }
+ if (_tablet_meta->time_series_compaction_goal_size_mbytes() !=
+ new_time_series_compaction_goal_size_mbytes) {
+ _tablet_meta->set_time_series_compaction_goal_size_mbytes(
+ new_time_series_compaction_goal_size_mbytes);
+ }
+ if (_tablet_meta->time_series_compaction_file_count_threshold() !=
+ new_time_series_compaction_file_count_threshold) {
+ _tablet_meta->set_time_series_compaction_file_count_threshold(
+ new_time_series_compaction_file_count_threshold);
+ }
+ if (_tablet_meta->time_series_compaction_time_threshold_seconds() !=
+ new_time_series_compaction_time_threshold_seconds) {
+ _tablet_meta->set_time_series_compaction_time_threshold_seconds(
+ new_time_series_compaction_time_threshold_seconds);
+ }
+ if (_tablet_meta->time_series_compaction_empty_rowsets_threshold() !=
+ new_time_series_compaction_empty_rowsets_threshold) {
+ _tablet_meta->set_time_series_compaction_empty_rowsets_threshold(
+ new_time_series_compaction_empty_rowsets_threshold);
+ }
+ if (_tablet_meta->time_series_compaction_level_threshold() !=
+ new_time_series_compaction_level_threshold) {
+ _tablet_meta->set_time_series_compaction_level_threshold(
+ new_time_series_compaction_level_threshold);
+ }
+ if (_tablet_meta->tablet_schema()->disable_auto_compaction() !=
+ new_disable_auto_compaction) {
+ _tablet_meta->mutable_tablet_schema()->set_disable_auto_compaction(
+ new_disable_auto_compaction);
+ }
+ if (_tablet_meta->vertical_compaction_num_columns_per_group() !=
+ new_vertical_compaction_num_columns_per_group) {
+ _tablet_meta->set_vertical_compaction_num_columns_per_group(
+ new_vertical_compaction_num_columns_per_group);
+ }
}
last_sync_tablet_meta_time_s = ::time(nullptr);
diff --git a/be/src/cloud/cloud_tablet_mgr.cpp
b/be/src/cloud/cloud_tablet_mgr.cpp
index 1e10cb243fd..3e5c7d0b764 100644
--- a/be/src/cloud/cloud_tablet_mgr.cpp
+++ b/be/src/cloud/cloud_tablet_mgr.cpp
@@ -293,6 +293,10 @@ bool CloudTabletMgr::peek_tablet_meta(int64_t tablet_id,
TabletMetaSharedPtr* ta
return true;
}
+std::shared_ptr<CloudTablet> CloudTabletMgr::get_tablet_if_cached(int64_t
tablet_id) {
+ return _tablet_map->get(tablet_id);
+}
+
void CloudTabletMgr::erase_tablet(int64_t tablet_id) {
auto tablet_id_str = std::to_string(tablet_id);
CacheKey key(tablet_id_str.data(), tablet_id_str.size());
diff --git a/be/src/cloud/cloud_tablet_mgr.h b/be/src/cloud/cloud_tablet_mgr.h
index c44f4f36f8c..5513468809b 100644
--- a/be/src/cloud/cloud_tablet_mgr.h
+++ b/be/src/cloud/cloud_tablet_mgr.h
@@ -62,6 +62,8 @@ public:
// Return true if cached tablet meta is found (without triggering RPC) and
filled.
bool peek_tablet_meta(int64_t tablet_id, TabletMetaSharedPtr* tablet_meta);
+ std::shared_ptr<CloudTablet> get_tablet_if_cached(int64_t tablet_id);
+
void erase_tablet(int64_t tablet_id);
void vacuum_stale_rowsets(const CountDownLatch& stop_latch);
diff --git a/be/src/service/internal_service.cpp
b/be/src/service/internal_service.cpp
index 887140a97aa..033a1694c6e 100644
--- a/be/src/service/internal_service.cpp
+++ b/be/src/service/internal_service.cpp
@@ -2565,5 +2565,14 @@ void
PInternalService::request_cdc_client(google::protobuf::RpcController* contr
}
}
+void PInternalService::sync_tablet_meta(google::protobuf::RpcController*
controller,
+ const PSyncTabletMetaRequest* request,
+ PSyncTabletMetaResponse* response,
+ google::protobuf::Closure* done) {
+ brpc::ClosureGuard closure_guard(done);
+ Status::NotSupported("sync_tablet_meta only supports cloud mode")
+ .to_protobuf(response->mutable_status());
+}
+
#include "common/compile_check_avoid_end.h"
} // namespace doris
diff --git a/be/src/service/internal_service.h
b/be/src/service/internal_service.h
index 4098c82120a..5fdc9fa3db0 100644
--- a/be/src/service/internal_service.h
+++ b/be/src/service/internal_service.h
@@ -223,6 +223,10 @@ public:
const PGetBeResourceRequest* request,
PGetBeResourceResponse* response,
google::protobuf::Closure* done) override;
+ void sync_tablet_meta(google::protobuf::RpcController* controller,
+ const PSyncTabletMetaRequest* request,
PSyncTabletMetaResponse* response,
+ google::protobuf::Closure* done) override;
+
void delete_dictionary(google::protobuf::RpcController* controller,
const PDeleteDictionaryRequest* request,
PDeleteDictionaryResponse* response,
diff --git a/be/test/cloud/cloud_internal_service_test.cpp
b/be/test/cloud/cloud_internal_service_test.cpp
new file mode 100644
index 00000000000..7bcb0f90a42
--- /dev/null
+++ b/be/test/cloud/cloud_internal_service_test.cpp
@@ -0,0 +1,180 @@
+// 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 "cloud/cloud_internal_service.h"
+
+#include <chrono>
+#include <condition_variable>
+#include <functional>
+#include <mutex>
+#include <unordered_map>
+
+#include "cloud/cloud_storage_engine.h"
+#include "cloud/cloud_tablet_mgr.h"
+#include "cpp/sync_point.h"
+#include "gen_cpp/Status_types.h"
+#include "storage/tablet/tablet_meta.h"
+#include "util/uid_util.h"
+#include "util/work_thread_pool.hpp"
+
+namespace doris {
+
+namespace {
+
+class TestClosure final : public google::protobuf::Closure {
+public:
+ explicit TestClosure(std::function<void()> callback) :
_callback(std::move(callback)) {}
+
+ void Run() override {
+ if (_callback) {
+ _callback();
+ }
+ }
+
+private:
+ std::function<void()> _callback;
+};
+
+} // namespace
+
+class CloudInternalServiceTest : public testing::Test {
+public:
+ CloudInternalServiceTest() : _engine(CloudStorageEngine(EngineOptions()))
{}
+
+protected:
+ static constexpr int64_t kSyncedTabletId = 90001;
+ static constexpr int64_t kFailedTabletId = 90002;
+ static constexpr int64_t kUncachedTabletId = 90003;
+
+ void SetUp() override {
+ auto sp = SyncPoint::get_instance();
+ sp->clear_all_call_backs();
+ sp->enable_processing();
+
+ sp->set_call_back("CloudMetaMgr::sync_tablet_rowsets",
+ [](auto&& args) {
try_any_cast_ret<Status>(args)->second = true; });
+ sp->set_call_back("CloudMetaMgr::get_tablet_meta", [this](auto&& args)
{
+ const auto tablet_id = try_any_cast<int64_t>(args[0]);
+ auto* tablet_meta_ptr =
try_any_cast<TabletMetaSharedPtr*>(args[1]);
+ auto* ret = try_any_cast_ret<Status>(args);
+
+ auto& call_count = _tablet_meta_call_count[tablet_id];
+ ++call_count;
+ if (tablet_id == kFailedTabletId && call_count >= 2) {
+ ret->first = Status::InternalError<false>("injected sync_meta
failure");
+ ret->second = true;
+ return;
+ }
+
+ *tablet_meta_ptr =
+ create_tablet_meta(tablet_id, call_count >= 2 ?
"time_series" : "size_based");
+ ret->second = true;
+ });
+ }
+
+ void TearDown() override {
+ auto sp = SyncPoint::get_instance();
+ sp->disable_processing();
+ sp->clear_all_call_backs();
+ }
+
+ TabletMetaSharedPtr create_tablet_meta(int64_t tablet_id, std::string
compaction_policy) {
+ TTabletSchema tablet_schema;
+ TColumn column;
+ column.__set_column_name("c1");
+ column.__set_column_type(TColumnType());
+ column.column_type.__set_type(TPrimitiveType::INT);
+ column.__set_is_key(true);
+ column.__set_aggregation_type(TAggregationType::NONE);
+ column.__set_col_unique_id(0);
+ tablet_schema.__set_columns({column});
+ tablet_schema.__set_keys_type(TKeysType::DUP_KEYS);
+
+ std::unordered_map<uint32_t, uint32_t> col_ordinal_to_unique_id = {{0,
0}};
+ auto tablet_meta = std::make_shared<TabletMeta>(
+ 1, 2, tablet_id, 15674, 4, 5, tablet_schema, 1,
col_ordinal_to_unique_id,
+ UniqueId(9, 10), TTabletType::TABLET_TYPE_DISK,
TCompressionType::LZ4F);
+ tablet_meta->set_compaction_policy(std::move(compaction_policy));
+ return tablet_meta;
+ }
+
+ CloudStorageEngine _engine;
+ std::unordered_map<int64_t, int> _tablet_meta_call_count;
+};
+
+TEST_F(CloudInternalServiceTest,
TestSyncTabletMetaCountsSyncedSkippedAndFailedTablets) {
+ auto synced_res = _engine.tablet_mgr().get_tablet(kSyncedTabletId);
+ ASSERT_TRUE(synced_res.has_value()) << synced_res.error();
+ auto failed_res = _engine.tablet_mgr().get_tablet(kFailedTabletId);
+ ASSERT_TRUE(failed_res.has_value()) << failed_res.error();
+
+ auto cached_synced_tablet = synced_res.value();
+ auto cached_failed_tablet = failed_res.value();
+ ASSERT_EQ("size_based",
cached_synced_tablet->tablet_meta()->compaction_policy());
+ ASSERT_EQ("size_based",
cached_failed_tablet->tablet_meta()->compaction_policy());
+
+ FifoThreadPool work_pool(1, 1, "sync_tablet_meta_test");
+ PSyncTabletMetaRequest request;
+ request.add_tablet_ids(kSyncedTabletId);
+ request.add_tablet_ids(kUncachedTabletId);
+ request.add_tablet_ids(kFailedTabletId);
+ PSyncTabletMetaResponse response;
+
+ std::mutex mutex;
+ std::condition_variable cv;
+ bool done = false;
+ TestClosure closure([&]() {
+ std::lock_guard lock(mutex);
+ done = true;
+ cv.notify_all();
+ });
+
+ test_submit_sync_tablet_meta(_engine, work_pool, &request, &response,
&closure);
+
+ std::unique_lock lock(mutex);
+ ASSERT_TRUE(cv.wait_for(lock, std::chrono::seconds(5), [&] { return done;
}));
+
+ ASSERT_TRUE(response.has_status());
+ EXPECT_EQ(TStatusCode::OK, response.status().status_code()) <<
response.status().DebugString();
+ EXPECT_EQ(1, response.synced_tablets());
+ EXPECT_EQ(1, response.skipped_tablets());
+ EXPECT_EQ(1, response.failed_tablets());
+ EXPECT_EQ("time_series",
cached_synced_tablet->tablet_meta()->compaction_policy());
+ EXPECT_EQ("size_based",
cached_failed_tablet->tablet_meta()->compaction_policy());
+ EXPECT_EQ(nullptr,
_engine.tablet_mgr().get_tablet_if_cached(kUncachedTabletId));
+}
+
+TEST_F(CloudInternalServiceTest, TestSyncTabletMetaOfferFailure) {
+ FifoThreadPool work_pool(1, 0, "sync_tablet_meta_offer_failure_test");
+ PSyncTabletMetaRequest request;
+ request.add_tablet_ids(kSyncedTabletId);
+ PSyncTabletMetaResponse response;
+
+ bool done = false;
+ TestClosure closure([&]() { done = true; });
+
+ test_submit_sync_tablet_meta(_engine, work_pool, &request, &response,
&closure);
+
+ EXPECT_TRUE(done);
+ ASSERT_TRUE(response.has_status());
+ EXPECT_EQ(TStatusCode::INTERNAL_ERROR, response.status().status_code());
+ EXPECT_TRUE(response.status().error_msgs(0).find(
+ "failed to offer sync_tablet_meta request to work
pool") !=
+ std::string::npos);
+}
+
+} // namespace doris
diff --git a/be/test/cloud/cloud_tablet_mgr_test.cpp
b/be/test/cloud/cloud_tablet_mgr_test.cpp
index 1838cc9c544..2521cbb9e74 100644
--- a/be/test/cloud/cloud_tablet_mgr_test.cpp
+++ b/be/test/cloud/cloud_tablet_mgr_test.cpp
@@ -363,4 +363,36 @@ TEST_F(CloudTabletMgrTest,
SyncTabletsRefreshesMetaOfContinuouslyIngestedTablet)
sp->clear_all_call_backs();
}
+TEST_F(CloudTabletMgrTest, TestGetTabletIfCachedOnlyReturnsCachedTablet) {
+ auto sp = SyncPoint::get_instance();
+ sp->clear_all_call_backs();
+ sp->enable_processing();
+
+ sp->set_call_back("CloudMetaMgr::get_tablet_meta", [this](auto&& args) {
+ auto* tablet_meta_ptr = try_any_cast<TabletMetaSharedPtr*>(args[1]);
+ *tablet_meta_ptr = _tablet_meta;
+ try_any_cast_ret<Status>(args)->second = true;
+ });
+ sp->set_call_back("CloudMetaMgr::sync_tablet_rowsets",
+ [](auto&& args) { try_any_cast_ret<Status>(args)->second
= true; });
+
+ CloudTabletMgr mgr(_engine);
+ const int64_t cached_tablet_id = 99999;
+ const int64_t uncached_tablet_id = 100000;
+
+ EXPECT_EQ(nullptr, mgr.get_tablet_if_cached(cached_tablet_id));
+
+ auto res = mgr.get_tablet(cached_tablet_id);
+ ASSERT_TRUE(res.has_value()) << res.error();
+
+ auto cached_tablet = mgr.get_tablet_if_cached(cached_tablet_id);
+ ASSERT_NE(nullptr, cached_tablet);
+ EXPECT_EQ(cached_tablet_id, cached_tablet->tablet_id());
+
+ EXPECT_EQ(nullptr, mgr.get_tablet_if_cached(uncached_tablet_id));
+
+ sp->disable_processing();
+ sp->clear_all_call_backs();
+}
+
} // namespace doris
diff --git a/be/test/cloud/cloud_tablet_test.cpp
b/be/test/cloud/cloud_tablet_test.cpp
index 533e39ca1b4..7d92cc35dc0 100644
--- a/be/test/cloud/cloud_tablet_test.cpp
+++ b/be/test/cloud/cloud_tablet_test.cpp
@@ -1094,6 +1094,7 @@ protected:
const std::string&
compaction_policy = "size_based") {
TTabletSchema new_schema;
new_schema.__set_disable_auto_compaction(disable_auto_compaction);
+ new_schema.__set_is_in_memory(false);
TColumn col;
col.__set_column_name("test_col_" +
std::to_string(_current_tablet_id));
col.__set_column_type(TColumnType());
@@ -1251,8 +1252,8 @@ TEST_F(CloudTabletSyncMetaTest,
TestSyncMetaDisableAutoCompactionUnchanged) {
sp->clear_all_call_backs();
}
-// Test sync_meta is skipped when enable_file_cache is false
-TEST_F(CloudTabletSyncMetaTest, TestSyncMetaSkippedWhenFileCacheDisabled) {
+// sync_meta should keep refreshing tablet meta even when file cache is
disabled.
+TEST_F(CloudTabletSyncMetaTest, TestSyncMetaWhenFileCacheDisabled) {
// Disable file cache
config::enable_file_cache = false;
@@ -1263,17 +1264,22 @@ TEST_F(CloudTabletSyncMetaTest,
TestSyncMetaSkippedWhenFileCacheDisabled) {
sp->clear_all_call_backs();
sp->enable_processing();
+ auto mock_tablet_meta = createMockTabletMeta(true, "time_series");
bool callback_called = false;
sp->set_call_back("CloudMetaMgr::get_tablet_meta",
- [&callback_called](auto&& args) { callback_called =
true; });
+ [mock_tablet_meta, &callback_called](auto&& args) {
+ callback_called = true;
+ auto* tablet_meta_ptr =
try_any_cast<TabletMetaSharedPtr*>(args[1]);
+ *tablet_meta_ptr = mock_tablet_meta;
+ try_any_cast_ret<Status>(args)->second = true;
+ });
- // Call sync_meta - should return early without calling get_tablet_meta
Status st = _tablet->sync_meta();
EXPECT_TRUE(st.ok());
- EXPECT_FALSE(callback_called);
+ EXPECT_TRUE(callback_called);
- // Verify disable_auto_compaction is not changed
-
EXPECT_FALSE(_tablet->tablet_meta()->tablet_schema()->disable_auto_compaction());
+
EXPECT_TRUE(_tablet->tablet_meta()->tablet_schema()->disable_auto_compaction());
+ EXPECT_EQ(_tablet->tablet_meta()->compaction_policy(), "time_series");
sp->disable_processing();
sp->clear_all_call_backs();
@@ -1292,6 +1298,12 @@ TEST_F(CloudTabletSyncMetaTest,
TestSyncMetaMultipleProperties) {
// Create mock tablet meta with updated properties
auto mock_tablet_meta = createMockTabletMeta(true, "time_series");
+ mock_tablet_meta->set_time_series_compaction_goal_size_mbytes(1234);
+ mock_tablet_meta->set_time_series_compaction_file_count_threshold(17);
+ mock_tablet_meta->set_time_series_compaction_time_threshold_seconds(3601);
+ 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 get_tablet_meta to return tablet_meta with updated properties
sp->set_call_back("CloudMetaMgr::get_tablet_meta",
[mock_tablet_meta](auto&& args) {
@@ -1309,10 +1321,47 @@ TEST_F(CloudTabletSyncMetaTest,
TestSyncMetaMultipleProperties) {
// Verify both properties are synced
EXPECT_TRUE(_tablet->tablet_meta()->tablet_schema()->disable_auto_compaction());
EXPECT_EQ(_tablet->tablet_meta()->compaction_policy(), "time_series");
+
EXPECT_EQ(_tablet->tablet_meta()->time_series_compaction_goal_size_mbytes(),
1234);
+
EXPECT_EQ(_tablet->tablet_meta()->time_series_compaction_file_count_threshold(),
17);
+
EXPECT_EQ(_tablet->tablet_meta()->time_series_compaction_time_threshold_seconds(),
3601);
+
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);
+
+ sp->disable_processing();
+ sp->clear_all_call_backs();
+}
+
+TEST_F(CloudTabletSyncMetaTest, TestSyncMetaSyncsTtlWithoutChangingInMemory) {
+ EXPECT_EQ(0, _tablet->tablet_meta()->ttl_seconds());
+ EXPECT_FALSE(_tablet->tablet_meta()->tablet_schema()->is_in_memory());
+
+ auto sp = SyncPoint::get_instance();
+ sp->clear_all_call_backs();
+ sp->enable_processing();
+
+ auto mock_tablet_meta = createMockTabletMeta(true, "time_series");
+ mock_tablet_meta->set_ttl_seconds(3600);
+ mock_tablet_meta->mutable_tablet_schema()->set_is_in_memory(true);
+
+ sp->set_call_back("CloudMetaMgr::get_tablet_meta",
[mock_tablet_meta](auto&& args) {
+ auto* tablet_meta_ptr = try_any_cast<TabletMetaSharedPtr*>(args[1]);
+ *tablet_meta_ptr = mock_tablet_meta;
+ try_any_cast_ret<Status>(args)->second = true;
+ });
+
+ Status st = _tablet->sync_meta();
+ EXPECT_TRUE(st.ok());
+
+ EXPECT_EQ(3600, _tablet->tablet_meta()->ttl_seconds());
+ EXPECT_FALSE(_tablet->tablet_meta()->tablet_schema()->is_in_memory());
+
EXPECT_TRUE(_tablet->tablet_meta()->tablet_schema()->disable_auto_compaction());
+ EXPECT_EQ(_tablet->tablet_meta()->compaction_policy(), "time_series");
sp->disable_processing();
sp->clear_all_call_backs();
}
+
class CloudTabletApplyVisiblePendingTest : public testing::Test {
public:
CloudTabletApplyVisiblePendingTest() :
_engine(CloudStorageEngine(EngineOptions {})) {}
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 87a5f3f5f4a..afbd99f5512 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
@@ -34,12 +34,21 @@ import org.apache.doris.common.Config;
import org.apache.doris.common.DdlException;
import org.apache.doris.common.MetaNotFoundException;
import org.apache.doris.common.UserException;
+import org.apache.doris.common.util.DebugPointUtil;
import org.apache.doris.common.util.DynamicPartitionUtil;
import org.apache.doris.common.util.PropertyAnalyzer;
+import org.apache.doris.proto.InternalService;
+import org.apache.doris.rpc.BackendServiceProxy;
import org.apache.doris.service.FrontendOptions;
+import org.apache.doris.system.Backend;
+import org.apache.doris.thrift.TStatusCode;
import com.google.common.base.Preconditions;
import com.google.common.collect.Lists;
+import com.google.common.util.concurrent.FutureCallback;
+import com.google.common.util.concurrent.Futures;
+import com.google.common.util.concurrent.ListenableFuture;
+import com.google.common.util.concurrent.MoreExecutors;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
@@ -538,6 +547,80 @@ public class CloudSchemaChangeHandler extends
SchemaChangeHandler {
if (response.getStatus().getCode() != Cloud.MetaServiceCode.OK) {
throw new UserException(response.getStatus().getMsg());
}
+
+ notifyBackendsToSyncTabletMeta(tableName,
updateTabletReq.getTabletMetaInfosList().stream()
+ .map(Cloud.TabletMetaInfoPB::getTabletId)
+ .collect(Collectors.toList()));
+ }
+ }
+
+ void notifyBackendsToSyncTabletMeta(String tableName, List<Long>
tabletIds) {
+ if (tabletIds.isEmpty()) {
+ return;
+ }
+ if
(DebugPointUtil.isEnable("CloudSchemaChangeHandler.notifyBackendsToSyncTabletMeta.skip"))
{
+ LOG.info("skip sync tablet meta rpc dispatch by debug point,
tableName={}, tabletIds={}",
+ tableName, tabletIds);
+ return;
+ }
+ List<Backend> backends;
+ try {
+ backends =
Env.getCurrentSystemInfo().getAllBackendsByAllCluster().values().asList();
+ } catch (UserException e) {
+ LOG.warn("failed to get alive backends for sync tablet meta,
tableName={}, tabletIds={}",
+ tableName, tabletIds, e);
+ return;
+ }
+
+ InternalService.PSyncTabletMetaRequest request =
InternalService.PSyncTabletMetaRequest.newBuilder()
+ .addAllTabletIds(tabletIds)
+ .build();
+ long dispatchStartTimeMs = System.currentTimeMillis();
+ int sentBackends = 0;
+ int skippedBackends = 0;
+ LOG.info("start to dispatch sync tablet meta rpc, tableName={},
tabletIds={}, request={}",
+ tableName, tabletIds, request);
+ for (Backend backend : backends) {
+ if (!backend.isAlive() || backend.getBrpcPort() <= 0) {
+ skippedBackends++;
+ continue;
+ }
+ try {
+ long rpcStartTimeMs = System.currentTimeMillis();
+ ListenableFuture<InternalService.PSyncTabletMetaResponse>
future =
+
BackendServiceProxy.getInstance().syncTabletMeta(backend.getBrpcAddress(),
request);
+ sentBackends++;
+ Futures.addCallback(future, new
FutureCallback<InternalService.PSyncTabletMetaResponse>() {
+ @Override
+ public void
onSuccess(InternalService.PSyncTabletMetaResponse response) {
+ long costMs = System.currentTimeMillis() -
rpcStartTimeMs;
+ if (response == null || !response.hasStatus()
+ || response.getStatus().getStatusCode() !=
TStatusCode.OK.getValue()) {
+ LOG.warn("sync tablet meta rpc returned non-ok
response, backendId={}, tabletIds={},"
+ + " tableName={}, response={},
costMs={}",
+ backend.getId(), tabletIds, tableName,
response, costMs);
+ return;
+ }
+ LOG.info("sync tablet meta rpc finished, backendId={},
tableName={}, tabletIds={},"
+ + " response={}, costMs={}",
+ backend.getId(), tableName, tabletIds,
response, costMs);
+ }
+
+ @Override
+ public void onFailure(Throwable t) {
+ LOG.warn("sync tablet meta rpc failed, backendId={},
tableName={}, tabletIds={}, costMs={}",
+ backend.getId(), tableName, tabletIds,
+ System.currentTimeMillis() - rpcStartTimeMs,
t);
+ }
+ }, MoreExecutors.directExecutor());
+ } catch (Exception e) {
+ LOG.warn("failed to dispatch sync tablet meta rpc,
backendId={}, tableName={}, tabletIds={}",
+ backend.getId(), tableName, tabletIds, e);
+ }
}
+ LOG.info("finish dispatching sync tablet meta rpc, tableName={},
tabletIds={}, backendNum={},"
+ + " sentBackends={}, skippedBackends={}, costMs={}",
+ tableName, tabletIds, backends.size(), sentBackends,
skippedBackends,
+ System.currentTimeMillis() - dispatchStartTimeMs);
}
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/rpc/BackendServiceClient.java
b/fe/fe-core/src/main/java/org/apache/doris/rpc/BackendServiceClient.java
index 13489e1894f..d61c1c3e00d 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/rpc/BackendServiceClient.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/rpc/BackendServiceClient.java
@@ -189,6 +189,11 @@ public class BackendServiceClient {
return stub.alterVaultSync(request);
}
+ public ListenableFuture<InternalService.PSyncTabletMetaResponse>
syncTabletMeta(
+ InternalService.PSyncTabletMetaRequest request) {
+ return stub.syncTabletMeta(request);
+ }
+
public Future<InternalService.PGetBeResourceResponse>
getBeResource(InternalService.PGetBeResourceRequest request,
int timeoutSec) {
return stub.withDeadlineAfter(timeoutSec,
TimeUnit.SECONDS).getBeResource(request);
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/rpc/BackendServiceProxy.java
b/fe/fe-core/src/main/java/org/apache/doris/rpc/BackendServiceProxy.java
index 15f35d37117..70e96650622 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/rpc/BackendServiceProxy.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/rpc/BackendServiceProxy.java
@@ -548,6 +548,18 @@ public class BackendServiceProxy {
}
}
+ public ListenableFuture<InternalService.PSyncTabletMetaResponse>
syncTabletMeta(TNetworkAddress address,
+ InternalService.PSyncTabletMetaRequest request) throws
RpcException {
+ try {
+ final BackendServiceClient client = getProxy(address);
+ return client.syncTabletMeta(request);
+ } catch (Throwable e) {
+ LOG.warn("failed to sync tablet meta from address={}:{}",
address.getHostname(),
+ address.getPort(), e);
+ throw new RpcException(address.getHostname(), e.getMessage());
+ }
+ }
+
public Future<InternalService.PFetchRemoteSchemaResponse>
fetchRemoteTabletSchemaAsync(
TNetworkAddress address, InternalService.PFetchRemoteSchemaRequest
request) throws RpcException {
try {
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
new file mode 100644
index 00000000000..467b7ae40b2
--- /dev/null
+++
b/fe/fe-core/src/test/java/org/apache/doris/cloud/alter/CloudSchemaChangeHandlerTest.java
@@ -0,0 +1,478 @@
+// 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.
+
+package org.apache.doris.cloud.alter;
+
+import org.apache.doris.catalog.Database;
+import org.apache.doris.catalog.Env;
+import org.apache.doris.catalog.KeysType;
+import org.apache.doris.catalog.MaterializedIndex;
+import org.apache.doris.catalog.OlapTable;
+import org.apache.doris.catalog.Partition;
+import org.apache.doris.catalog.Table;
+import org.apache.doris.cloud.proto.Cloud;
+import org.apache.doris.cloud.rpc.MetaServiceProxy;
+import org.apache.doris.common.Config;
+import org.apache.doris.common.UserException;
+import org.apache.doris.common.util.DebugPointUtil;
+import org.apache.doris.proto.InternalService;
+import org.apache.doris.rpc.BackendServiceProxy;
+import org.apache.doris.system.Backend;
+import org.apache.doris.system.SystemInfoService;
+import org.apache.doris.thrift.TNetworkAddress;
+import org.apache.doris.thrift.TStatusCode;
+
+import com.google.common.collect.ImmutableMap;
+import com.google.common.util.concurrent.Futures;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.ArgumentCaptor;
+import org.mockito.MockedStatic;
+import org.mockito.Mockito;
+
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+public class CloudSchemaChangeHandlerTest {
+ private int originalCloudTxnTabletBatchSize;
+ private String originalCloudUniqueId;
+ private String originalMetaServiceEndpoint;
+ private boolean originalEnableDebugPoints;
+
+ @Before
+ public void setUp() {
+ originalCloudTxnTabletBatchSize = Config.cloud_txn_tablet_batch_size;
+ originalCloudUniqueId = Config.cloud_unique_id;
+ originalMetaServiceEndpoint = Config.meta_service_endpoint;
+ originalEnableDebugPoints = Config.enable_debug_points;
+ Config.cloud_txn_tablet_batch_size = 2;
+ Config.cloud_unique_id = "cloud-test";
+ Config.meta_service_endpoint = "127.0.0.1:20121";
+ Config.enable_debug_points = false;
+ DebugPointUtil.clearDebugPoints();
+ }
+
+ @After
+ public void tearDown() {
+ Config.cloud_txn_tablet_batch_size = originalCloudTxnTabletBatchSize;
+ Config.cloud_unique_id = originalCloudUniqueId;
+ Config.meta_service_endpoint = originalMetaServiceEndpoint;
+ Config.enable_debug_points = originalEnableDebugPoints;
+ DebugPointUtil.clearDebugPoints();
+ }
+
+ @Test
+ public void testUpdateTablePropertiesNotifiesAllAliveBackendsInBatches()
throws Exception {
+ CloudSchemaChangeHandler handler = new CloudSchemaChangeHandler();
+ Database db = createMockDatabaseWithThreeTablets();
+
+ Backend aliveBackend1 = new Backend(1L, "be-1", 9050);
+ aliveBackend1.setAlive(true);
+ aliveBackend1.setBrpcPort(8060);
+ Backend aliveBackend2 = new Backend(2L, "be-2", 9050);
+ aliveBackend2.setAlive(true);
+ aliveBackend2.setBrpcPort(8061);
+ Backend deadBackend = new Backend(3L, "be-dead", 9050);
+ deadBackend.setAlive(false);
+ deadBackend.setBrpcPort(8062);
+
+ SystemInfoService systemInfoService =
Mockito.mock(SystemInfoService.class);
+
Mockito.when(systemInfoService.getAllBackendsByAllCluster()).thenReturn(ImmutableMap.of(
+ aliveBackend1.getId(), aliveBackend1,
+ aliveBackend2.getId(), aliveBackend2,
+ deadBackend.getId(), deadBackend));
+
+ Env env = Mockito.mock(Env.class);
+
+ MetaServiceProxy metaServiceProxy =
Mockito.mock(MetaServiceProxy.class);
+
Mockito.when(metaServiceProxy.updateTablet(Mockito.any())).thenReturn(okUpdateTabletResponse());
+
+ BackendServiceProxy backendServiceProxy =
Mockito.mock(BackendServiceProxy.class);
+ Mockito.when(backendServiceProxy.syncTabletMeta(Mockito.any(),
Mockito.any()))
+
.thenReturn(Futures.immediateFuture(okSyncTabletMetaResponse()));
+
+ try (MockedStatic<Env> envMock = Mockito.mockStatic(Env.class);
+ MockedStatic<MetaServiceProxy> metaProxyMock =
Mockito.mockStatic(MetaServiceProxy.class);
+ MockedStatic<BackendServiceProxy> backendProxyMock =
Mockito.mockStatic(BackendServiceProxy.class)) {
+ envMock.when(Env::getCurrentEnv).thenReturn(env);
+
envMock.when(Env::getCurrentSystemInfo).thenReturn(systemInfoService);
+
metaProxyMock.when(MetaServiceProxy::getInstance).thenReturn(metaServiceProxy);
+
backendProxyMock.when(BackendServiceProxy::getInstance).thenReturn(backendServiceProxy);
+
+ Map<String, String> properties = new HashMap<>();
+ properties.put("compaction_policy", "time_series");
+ handler.updateTableProperties(db, "tbl", properties);
+ }
+
+ ArgumentCaptor<Cloud.UpdateTabletRequest> updateCaptor =
+ ArgumentCaptor.forClass(Cloud.UpdateTabletRequest.class);
+ Mockito.verify(metaServiceProxy,
Mockito.times(2)).updateTablet(updateCaptor.capture());
+ List<Cloud.UpdateTabletRequest> updateRequests =
updateCaptor.getAllValues();
+ Assert.assertEquals(Arrays.asList(101L, 102L),
+ updateRequests.get(0).getTabletMetaInfosList().stream()
+
.map(Cloud.TabletMetaInfoPB::getTabletId).collect(Collectors.toList()));
+ Assert.assertEquals(Arrays.asList(103L),
+ updateRequests.get(1).getTabletMetaInfosList().stream()
+
.map(Cloud.TabletMetaInfoPB::getTabletId).collect(Collectors.toList()));
+
+ ArgumentCaptor<TNetworkAddress> addressCaptor =
ArgumentCaptor.forClass(TNetworkAddress.class);
+ ArgumentCaptor<InternalService.PSyncTabletMetaRequest> syncCaptor =
+
ArgumentCaptor.forClass(InternalService.PSyncTabletMetaRequest.class);
+ Mockito.verify(backendServiceProxy, Mockito.times(4))
+ .syncTabletMeta(addressCaptor.capture(), syncCaptor.capture());
+
+ List<TNetworkAddress> addresses = addressCaptor.getAllValues();
+ Assert.assertFalse(addresses.stream().anyMatch(addr ->
"be-dead".equals(addr.getHostname())));
+ Assert.assertEquals(2L, addresses.stream().filter(addr ->
"be-1".equals(addr.getHostname())).count());
+ Assert.assertEquals(2L, addresses.stream().filter(addr ->
"be-2".equals(addr.getHostname())).count());
+
+ List<InternalService.PSyncTabletMetaRequest> syncRequests =
syncCaptor.getAllValues();
+ Assert.assertEquals(Arrays.asList(101L, 102L),
syncRequests.get(0).getTabletIdsList());
+ Assert.assertEquals(Arrays.asList(101L, 102L),
syncRequests.get(1).getTabletIdsList());
+ Assert.assertEquals(Arrays.asList(103L),
syncRequests.get(2).getTabletIdsList());
+ Assert.assertEquals(Arrays.asList(103L),
syncRequests.get(3).getTabletIdsList());
+ }
+
+ @Test
+ public void testUpdateTablePropertiesThrowsWhenUpdateTabletFails() throws
Exception {
+ CloudSchemaChangeHandler handler = new CloudSchemaChangeHandler();
+ Database db = createMockDatabaseWithThreeTablets();
+
+ MetaServiceProxy metaServiceProxy =
Mockito.mock(MetaServiceProxy.class);
+ Mockito.when(metaServiceProxy.updateTablet(Mockito.any()))
+ .thenThrow(new RuntimeException("update failed"));
+
+ try (MockedStatic<MetaServiceProxy> metaProxyMock =
Mockito.mockStatic(MetaServiceProxy.class)) {
+
metaProxyMock.when(MetaServiceProxy::getInstance).thenReturn(metaServiceProxy);
+
+ Map<String, String> properties = new HashMap<>();
+ properties.put("compaction_policy", "time_series");
+ UserException exception = Assert.assertThrows(UserException.class,
+ () -> handler.updateTableProperties(db, "tbl",
properties));
+ Assert.assertTrue(exception.getMessage().contains("update
failed"));
+ }
+ }
+
+ @Test
+ public void testUpdateTablePropertiesThrowsWhenUpdateTabletResponseNotOk()
throws Exception {
+ CloudSchemaChangeHandler handler = new CloudSchemaChangeHandler();
+ Database db = createMockDatabaseWithThreeTablets();
+
+ MetaServiceProxy metaServiceProxy =
Mockito.mock(MetaServiceProxy.class);
+ Mockito.when(metaServiceProxy.updateTablet(Mockito.any()))
+ .thenReturn(Cloud.UpdateTabletResponse.newBuilder()
+ .setStatus(Cloud.MetaServiceResponseStatus.newBuilder()
+
.setCode(Cloud.MetaServiceCode.INVALID_ARGUMENT)
+ .setMsg("meta service rejected"))
+ .build());
+
+ try (MockedStatic<MetaServiceProxy> metaProxyMock =
Mockito.mockStatic(MetaServiceProxy.class)) {
+
metaProxyMock.when(MetaServiceProxy::getInstance).thenReturn(metaServiceProxy);
+
+ Map<String, String> properties = new HashMap<>();
+ properties.put("compaction_policy", "time_series");
+ UserException exception = Assert.assertThrows(UserException.class,
+ () -> handler.updateTableProperties(db, "tbl",
properties));
+ Assert.assertTrue(exception.getMessage().contains("meta service
rejected"));
+ }
+ }
+
+ @Test
+ public void testNotifyBackendsToSyncTabletMetaSwallowsDispatchFailure()
throws Exception {
+ CloudSchemaChangeHandler handler = new CloudSchemaChangeHandler();
+ Backend aliveBackend1 = new Backend(1L, "be-1", 9050);
+ aliveBackend1.setAlive(true);
+ aliveBackend1.setBrpcPort(8060);
+ Backend aliveBackend2 = new Backend(2L, "be-2", 9050);
+ aliveBackend2.setAlive(true);
+ aliveBackend2.setBrpcPort(8061);
+
+ SystemInfoService systemInfoService =
Mockito.mock(SystemInfoService.class);
+
Mockito.when(systemInfoService.getAllBackendsByAllCluster()).thenReturn(ImmutableMap.of(
+ aliveBackend1.getId(), aliveBackend1,
+ aliveBackend2.getId(), aliveBackend2));
+
+ BackendServiceProxy backendServiceProxy =
Mockito.mock(BackendServiceProxy.class);
+ Mockito.when(backendServiceProxy.syncTabletMeta(Mockito.argThat(
+ addr -> addr != null &&
"be-1".equals(addr.getHostname())), Mockito.any()))
+
.thenReturn(Futures.immediateFuture(okSyncTabletMetaResponse()));
+ Mockito.when(backendServiceProxy.syncTabletMeta(Mockito.argThat(
+ addr -> addr != null &&
"be-2".equals(addr.getHostname())), Mockito.any()))
+ .thenThrow(new org.apache.doris.rpc.RpcException("be-2",
"dispatch failed"));
+
+ try (MockedStatic<Env> envMock = Mockito.mockStatic(Env.class);
+ MockedStatic<BackendServiceProxy> backendProxyMock =
Mockito.mockStatic(BackendServiceProxy.class)) {
+
envMock.when(Env::getCurrentSystemInfo).thenReturn(systemInfoService);
+
backendProxyMock.when(BackendServiceProxy::getInstance).thenReturn(backendServiceProxy);
+
+ handler.notifyBackendsToSyncTabletMeta("test_table",
Arrays.asList(101L, 102L));
+ }
+
+ Mockito.verify(backendServiceProxy, Mockito.times(2))
+ .syncTabletMeta(Mockito.any(), Mockito.any());
+ }
+
+ @Test
+ public void testNotifyBackendsToSyncTabletMetaReturnsForEmptyTabletList() {
+ CloudSchemaChangeHandler handler = new CloudSchemaChangeHandler();
+ BackendServiceProxy backendServiceProxy =
Mockito.mock(BackendServiceProxy.class);
+
+ try (MockedStatic<BackendServiceProxy> backendProxyMock =
Mockito.mockStatic(BackendServiceProxy.class)) {
+
backendProxyMock.when(BackendServiceProxy::getInstance).thenReturn(backendServiceProxy);
+
+ handler.notifyBackendsToSyncTabletMeta("test_table",
Arrays.asList());
+ }
+
+ Mockito.verifyNoInteractions(backendServiceProxy);
+ }
+
+ @Test
+ public void
testNotifyBackendsToSyncTabletMetaReturnsWhenDebugPointEnabled() {
+ CloudSchemaChangeHandler handler = new CloudSchemaChangeHandler();
+ BackendServiceProxy backendServiceProxy =
Mockito.mock(BackendServiceProxy.class);
+ Config.enable_debug_points = true;
+
DebugPointUtil.addDebugPoint("CloudSchemaChangeHandler.notifyBackendsToSyncTabletMeta.skip");
+
+ try (MockedStatic<BackendServiceProxy> backendProxyMock =
Mockito.mockStatic(BackendServiceProxy.class)) {
+
backendProxyMock.when(BackendServiceProxy::getInstance).thenReturn(backendServiceProxy);
+
+ handler.notifyBackendsToSyncTabletMeta("test_table",
Arrays.asList(101L, 102L));
+ }
+
+ Mockito.verifyNoInteractions(backendServiceProxy);
+ }
+
+ @Test
+ public void testNotifyBackendsToSyncTabletMetaSkipsUnavailableBackends()
throws Exception {
+ CloudSchemaChangeHandler handler = new CloudSchemaChangeHandler();
+ Backend aliveBackend = new Backend(1L, "be-1", 9050);
+ aliveBackend.setAlive(true);
+ aliveBackend.setBrpcPort(8060);
+ Backend deadBackend = new Backend(2L, "be-dead", 9050);
+ deadBackend.setAlive(false);
+ deadBackend.setBrpcPort(8061);
+ Backend noBrpcBackend = new Backend(3L, "be-no-brpc", 9050);
+ noBrpcBackend.setAlive(true);
+ noBrpcBackend.setBrpcPort(0);
+
+ SystemInfoService systemInfoService =
Mockito.mock(SystemInfoService.class);
+
Mockito.when(systemInfoService.getAllBackendsByAllCluster()).thenReturn(ImmutableMap.of(
+ aliveBackend.getId(), aliveBackend,
+ deadBackend.getId(), deadBackend,
+ noBrpcBackend.getId(), noBrpcBackend));
+
+ BackendServiceProxy backendServiceProxy =
Mockito.mock(BackendServiceProxy.class);
+ Mockito.when(backendServiceProxy.syncTabletMeta(Mockito.any(),
Mockito.any()))
+
.thenReturn(Futures.immediateFuture(okSyncTabletMetaResponse()));
+
+ try (MockedStatic<Env> envMock = Mockito.mockStatic(Env.class);
+ MockedStatic<BackendServiceProxy> backendProxyMock =
Mockito.mockStatic(BackendServiceProxy.class)) {
+
envMock.when(Env::getCurrentSystemInfo).thenReturn(systemInfoService);
+
backendProxyMock.when(BackendServiceProxy::getInstance).thenReturn(backendServiceProxy);
+
+ handler.notifyBackendsToSyncTabletMeta("test_table",
Arrays.asList(101L, 102L));
+ }
+
+ ArgumentCaptor<InternalService.PSyncTabletMetaRequest> syncCaptor =
+
ArgumentCaptor.forClass(InternalService.PSyncTabletMetaRequest.class);
+ Mockito.verify(backendServiceProxy, Mockito.times(1))
+ .syncTabletMeta(Mockito.argThat(addr ->
"be-1".equals(addr.getHostname())), syncCaptor.capture());
+ Assert.assertEquals(Arrays.asList(101L, 102L),
syncCaptor.getValue().getTabletIdsList());
+ }
+
+ @Test
+ public void testNotifyBackendsToSyncTabletMetaSwallowsAsyncFailure()
throws Exception {
+ CloudSchemaChangeHandler handler = new CloudSchemaChangeHandler();
+ Backend aliveBackend = new Backend(1L, "be-1", 9050);
+ aliveBackend.setAlive(true);
+ aliveBackend.setBrpcPort(8060);
+
+ SystemInfoService systemInfoService =
Mockito.mock(SystemInfoService.class);
+
Mockito.when(systemInfoService.getAllBackendsByAllCluster()).thenReturn(ImmutableMap.of(
+ aliveBackend.getId(), aliveBackend));
+
+ BackendServiceProxy backendServiceProxy =
Mockito.mock(BackendServiceProxy.class);
+ Mockito.when(backendServiceProxy.syncTabletMeta(Mockito.any(),
Mockito.any()))
+ .thenReturn(Futures.immediateFailedFuture(new
RuntimeException("async failed")));
+
+ try (MockedStatic<Env> envMock = Mockito.mockStatic(Env.class);
+ MockedStatic<BackendServiceProxy> backendProxyMock =
Mockito.mockStatic(BackendServiceProxy.class)) {
+
envMock.when(Env::getCurrentSystemInfo).thenReturn(systemInfoService);
+
backendProxyMock.when(BackendServiceProxy::getInstance).thenReturn(backendServiceProxy);
+
+ handler.notifyBackendsToSyncTabletMeta("test_table",
Arrays.asList(101L, 102L));
+ }
+
+ Mockito.verify(backendServiceProxy, Mockito.times(1))
+ .syncTabletMeta(Mockito.any(), Mockito.any());
+ }
+
+ @Test
+ public void testNotifyBackendsToSyncTabletMetaSwallowsNullResponse()
throws Exception {
+ CloudSchemaChangeHandler handler = new CloudSchemaChangeHandler();
+ Backend aliveBackend = new Backend(1L, "be-1", 9050);
+ aliveBackend.setAlive(true);
+ aliveBackend.setBrpcPort(8060);
+
+ SystemInfoService systemInfoService =
Mockito.mock(SystemInfoService.class);
+
Mockito.when(systemInfoService.getAllBackendsByAllCluster()).thenReturn(ImmutableMap.of(
+ aliveBackend.getId(), aliveBackend));
+
+ BackendServiceProxy backendServiceProxy =
Mockito.mock(BackendServiceProxy.class);
+ Mockito.when(backendServiceProxy.syncTabletMeta(Mockito.any(),
Mockito.any()))
+ .thenReturn(Futures.immediateFuture(null));
+
+ try (MockedStatic<Env> envMock = Mockito.mockStatic(Env.class);
+ MockedStatic<BackendServiceProxy> backendProxyMock =
Mockito.mockStatic(BackendServiceProxy.class)) {
+
envMock.when(Env::getCurrentSystemInfo).thenReturn(systemInfoService);
+
backendProxyMock.when(BackendServiceProxy::getInstance).thenReturn(backendServiceProxy);
+
+ handler.notifyBackendsToSyncTabletMeta("test_table",
Arrays.asList(101L, 102L));
+ }
+
+ Mockito.verify(backendServiceProxy, Mockito.times(1))
+ .syncTabletMeta(Mockito.any(), Mockito.any());
+ }
+
+ @Test
+ public void
testNotifyBackendsToSyncTabletMetaSwallowsResponseWithoutStatus() throws
Exception {
+ CloudSchemaChangeHandler handler = new CloudSchemaChangeHandler();
+ Backend aliveBackend = new Backend(1L, "be-1", 9050);
+ aliveBackend.setAlive(true);
+ aliveBackend.setBrpcPort(8060);
+
+ SystemInfoService systemInfoService =
Mockito.mock(SystemInfoService.class);
+
Mockito.when(systemInfoService.getAllBackendsByAllCluster()).thenReturn(ImmutableMap.of(
+ aliveBackend.getId(), aliveBackend));
+
+ BackendServiceProxy backendServiceProxy =
Mockito.mock(BackendServiceProxy.class);
+ Mockito.when(backendServiceProxy.syncTabletMeta(Mockito.any(),
Mockito.any()))
+
.thenReturn(Futures.immediateFuture(InternalService.PSyncTabletMetaResponse.newBuilder()
+ .setSyncedTablets(2)
+ .build()));
+
+ try (MockedStatic<Env> envMock = Mockito.mockStatic(Env.class);
+ MockedStatic<BackendServiceProxy> backendProxyMock =
Mockito.mockStatic(BackendServiceProxy.class)) {
+
envMock.when(Env::getCurrentSystemInfo).thenReturn(systemInfoService);
+
backendProxyMock.when(BackendServiceProxy::getInstance).thenReturn(backendServiceProxy);
+
+ handler.notifyBackendsToSyncTabletMeta("test_table",
Arrays.asList(101L, 102L));
+ }
+
+ Mockito.verify(backendServiceProxy, Mockito.times(1))
+ .syncTabletMeta(Mockito.any(), Mockito.any());
+ }
+
+ @Test
+ public void testNotifyBackendsToSyncTabletMetaSwallowsNonOkResponse()
throws Exception {
+ CloudSchemaChangeHandler handler = new CloudSchemaChangeHandler();
+ Backend aliveBackend1 = new Backend(1L, "be-1", 9050);
+ aliveBackend1.setAlive(true);
+ aliveBackend1.setBrpcPort(8060);
+ Backend aliveBackend2 = new Backend(2L, "be-2", 9050);
+ aliveBackend2.setAlive(true);
+ aliveBackend2.setBrpcPort(8061);
+
+ SystemInfoService systemInfoService =
Mockito.mock(SystemInfoService.class);
+
Mockito.when(systemInfoService.getAllBackendsByAllCluster()).thenReturn(ImmutableMap.of(
+ aliveBackend1.getId(), aliveBackend1,
+ aliveBackend2.getId(), aliveBackend2));
+
+ BackendServiceProxy backendServiceProxy =
Mockito.mock(BackendServiceProxy.class);
+ Mockito.when(backendServiceProxy.syncTabletMeta(Mockito.any(),
Mockito.any()))
+
.thenReturn(Futures.immediateFuture(InternalService.PSyncTabletMetaResponse.newBuilder()
+
.setStatus(org.apache.doris.proto.Types.PStatus.newBuilder()
+
.setStatusCode(TStatusCode.CANCELLED.getValue()))
+ .setFailedTablets(2)
+ .build()));
+
+ try (MockedStatic<Env> envMock = Mockito.mockStatic(Env.class);
+ MockedStatic<BackendServiceProxy> backendProxyMock =
Mockito.mockStatic(BackendServiceProxy.class)) {
+
envMock.when(Env::getCurrentSystemInfo).thenReturn(systemInfoService);
+
backendProxyMock.when(BackendServiceProxy::getInstance).thenReturn(backendServiceProxy);
+
+ handler.notifyBackendsToSyncTabletMeta("test_table",
Arrays.asList(101L, 102L));
+ }
+
+ Mockito.verify(backendServiceProxy, Mockito.times(2))
+ .syncTabletMeta(Mockito.any(), Mockito.any());
+ }
+
+ @Test
+ public void
testNotifyBackendsToSyncTabletMetaReturnsWhenBackendDiscoveryFails() throws
Exception {
+ CloudSchemaChangeHandler handler = new CloudSchemaChangeHandler();
+ SystemInfoService systemInfoService =
Mockito.mock(SystemInfoService.class);
+ Mockito.when(systemInfoService.getAllBackendsByAllCluster())
+ .thenThrow(new
org.apache.doris.common.AnalysisException("backend discovery failed"));
+
+ BackendServiceProxy backendServiceProxy =
Mockito.mock(BackendServiceProxy.class);
+
+ try (MockedStatic<Env> envMock = Mockito.mockStatic(Env.class);
+ MockedStatic<BackendServiceProxy> backendProxyMock =
Mockito.mockStatic(BackendServiceProxy.class)) {
+
envMock.when(Env::getCurrentSystemInfo).thenReturn(systemInfoService);
+
backendProxyMock.when(BackendServiceProxy::getInstance).thenReturn(backendServiceProxy);
+
+ handler.notifyBackendsToSyncTabletMeta("test_table",
Arrays.asList(101L, 102L));
+ }
+
+ Mockito.verifyNoInteractions(backendServiceProxy);
+ }
+
+ private Cloud.UpdateTabletResponse okUpdateTabletResponse() {
+ return Cloud.UpdateTabletResponse.newBuilder()
+
.setStatus(Cloud.MetaServiceResponseStatus.newBuilder().setCode(Cloud.MetaServiceCode.OK))
+ .build();
+ }
+
+ private Database createMockDatabaseWithThreeTablets() throws Exception {
+ Database db = Mockito.mock(Database.class);
+ OlapTable table = Mockito.mock(OlapTable.class);
+ Partition partition = Mockito.mock(Partition.class);
+ MaterializedIndex index = Mockito.mock(MaterializedIndex.class);
+ org.apache.doris.catalog.Tablet tablet1 =
Mockito.mock(org.apache.doris.catalog.Tablet.class);
+ org.apache.doris.catalog.Tablet tablet2 =
Mockito.mock(org.apache.doris.catalog.Tablet.class);
+ org.apache.doris.catalog.Tablet tablet3 =
Mockito.mock(org.apache.doris.catalog.Tablet.class);
+
+ Mockito.when(db.getTableOrMetaException("tbl",
Table.TableType.OLAP)).thenReturn(table);
+ Mockito.when(table.getName()).thenReturn("tbl");
+ Mockito.when(table.getCompactionPolicy()).thenReturn("size_based");
+ Mockito.when(table.getKeysType()).thenReturn(KeysType.DUP_KEYS);
+
Mockito.when(table.getPartitions()).thenReturn(Arrays.asList(partition));
+ Mockito.when(table.getPartition("p1")).thenReturn(partition);
+ Mockito.when(partition.getName()).thenReturn("p1");
+
Mockito.when(partition.getMaterializedIndices(MaterializedIndex.IndexExtState.VISIBLE))
+ .thenReturn(Arrays.asList(index));
+ Mockito.when(index.getTablets()).thenReturn(Arrays.asList(tablet1,
tablet2, tablet3));
+ Mockito.when(tablet1.getId()).thenReturn(101L);
+ Mockito.when(tablet2.getId()).thenReturn(102L);
+ Mockito.when(tablet3.getId()).thenReturn(103L);
+ return db;
+ }
+
+ private InternalService.PSyncTabletMetaResponse okSyncTabletMetaResponse()
{
+ return InternalService.PSyncTabletMetaResponse.newBuilder()
+ .setStatus(org.apache.doris.proto.Types.PStatus.newBuilder()
+ .setStatusCode(TStatusCode.OK.getValue()))
+ .setSyncedTablets(1)
+ .build();
+ }
+}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/rpc/BackendServiceClientTest.java
b/fe/fe-core/src/test/java/org/apache/doris/rpc/BackendServiceClientTest.java
index 68aa36eee74..7992106cffd 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/rpc/BackendServiceClientTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/rpc/BackendServiceClientTest.java
@@ -19,13 +19,18 @@ package org.apache.doris.rpc;
import org.apache.doris.common.Config;
import org.apache.doris.common.jmockit.Deencapsulation;
+import org.apache.doris.proto.InternalService;
+import org.apache.doris.proto.PBackendServiceGrpc;
import org.apache.doris.thrift.TNetworkAddress;
+import com.google.common.util.concurrent.Futures;
+import com.google.common.util.concurrent.ListenableFuture;
import io.grpc.ManagedChannel;
import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
+import org.mockito.Mockito;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;
@@ -224,4 +229,29 @@ public class BackendServiceClientTest {
client1.shutdown();
client2.shutdown();
}
+
+ @Test
+ public void testSyncTabletMeta() {
+ TNetworkAddress address = new TNetworkAddress("localhost", 9060);
+ BackendServiceClient client = new BackendServiceClient(address,
"127.0.0.1", executor);
+
+ PBackendServiceGrpc.PBackendServiceFutureStub stub =
+
Mockito.mock(PBackendServiceGrpc.PBackendServiceFutureStub.class);
+ Deencapsulation.setField(client, "stub", stub);
+
+ InternalService.PSyncTabletMetaRequest request =
InternalService.PSyncTabletMetaRequest.newBuilder()
+ .addTabletIds(10001L)
+ .build();
+ ListenableFuture<InternalService.PSyncTabletMetaResponse>
expectedFuture = Futures.immediateFuture(
+ InternalService.PSyncTabletMetaResponse.newBuilder()
+ .setSyncedTablets(1)
+ .build());
+ Mockito.when(stub.syncTabletMeta(request)).thenReturn(expectedFuture);
+
+ ListenableFuture<InternalService.PSyncTabletMetaResponse> actualFuture
= client.syncTabletMeta(request);
+
+ Assert.assertSame(expectedFuture, actualFuture);
+ Mockito.verify(stub).syncTabletMeta(request);
+ client.shutdown();
+ }
}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/rpc/BackendServiceProxyTest.java
b/fe/fe-core/src/test/java/org/apache/doris/rpc/BackendServiceProxyTest.java
index 2a897f0e400..427ff457034 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/rpc/BackendServiceProxyTest.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/rpc/BackendServiceProxyTest.java
@@ -21,8 +21,11 @@ import org.apache.doris.catalog.Env;
import org.apache.doris.common.Config;
import org.apache.doris.common.DNSCache;
import org.apache.doris.common.jmockit.Deencapsulation;
+import org.apache.doris.proto.InternalService;
import org.apache.doris.thrift.TNetworkAddress;
+import com.google.common.util.concurrent.Futures;
+import com.google.common.util.concurrent.ListenableFuture;
import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
@@ -30,6 +33,7 @@ import org.junit.Test;
import org.mockito.MockedStatic;
import org.mockito.Mockito;
+import java.lang.reflect.Constructor;
import java.net.UnknownHostException;
import java.util.Map;
@@ -300,4 +304,54 @@ public class BackendServiceProxyTest {
Assert.assertTrue(serviceMap.containsKey(address1));
Assert.assertTrue(serviceMap.containsKey(address2));
}
+
+ @Test
+ public void testSyncTabletMeta() throws Exception {
+ String hostname = "backend-host.example.com";
+ String resolvedIp = "10.0.0.1";
+ int port = 9060;
+ TNetworkAddress address = new TNetworkAddress(hostname, port);
+ InternalService.PSyncTabletMetaRequest request =
InternalService.PSyncTabletMetaRequest.newBuilder()
+ .addTabletIds(10001L)
+ .build();
+
+ BackendServiceClient client = Mockito.mock(BackendServiceClient.class);
+ Mockito.when(client.isNormalState()).thenReturn(true);
+ Mockito.when(client.isUsingLatestChannelConfig()).thenReturn(true);
+ ListenableFuture<InternalService.PSyncTabletMetaResponse>
expectedFuture = Futures.immediateFuture(
+ InternalService.PSyncTabletMetaResponse.newBuilder()
+ .setSyncedTablets(1)
+ .build());
+
Mockito.when(client.syncTabletMeta(request)).thenReturn(expectedFuture);
+ Mockito.when(mockDnsCache.get(hostname)).thenReturn(resolvedIp);
+
+ Map<TNetworkAddress, Object> serviceMap =
Deencapsulation.getField(proxy, "serviceMap");
+ serviceMap.put(address, newBackendServiceClientExtIp(resolvedIp,
client));
+
+ ListenableFuture<InternalService.PSyncTabletMetaResponse> actualFuture
= proxy.syncTabletMeta(address, request);
+
+ Assert.assertSame(expectedFuture, actualFuture);
+ Mockito.verify(client).syncTabletMeta(request);
+ }
+
+ @Test
+ public void testSyncTabletMetaThrowsRpcException() {
+ String hostname = "non-existent-host.example.com";
+ TNetworkAddress address = new TNetworkAddress(hostname, 9060);
+ InternalService.PSyncTabletMetaRequest request =
InternalService.PSyncTabletMetaRequest.newBuilder()
+ .addTabletIds(10001L)
+ .build();
+ Mockito.when(mockDnsCache.get(hostname)).thenReturn("");
+
+ RpcException exception = Assert.assertThrows(RpcException.class, () ->
proxy.syncTabletMeta(address, request));
+ Assert.assertTrue(exception.getMessage().contains(hostname));
+ }
+
+ private Object newBackendServiceClientExtIp(String realIp,
BackendServiceClient client) throws Exception {
+ Class<?> clazz =
Class.forName("org.apache.doris.rpc.BackendServiceProxy$BackendServiceClientExtIp");
+ Constructor<?> constructor =
clazz.getDeclaredConstructor(BackendServiceProxy.class,
+ String.class, BackendServiceClient.class);
+ constructor.setAccessible(true);
+ return constructor.newInstance(proxy, realIp, client);
+ }
}
diff --git a/gensrc/proto/internal_service.proto
b/gensrc/proto/internal_service.proto
index aa8fbaa1478..ab666bda634 100644
--- a/gensrc/proto/internal_service.proto
+++ b/gensrc/proto/internal_service.proto
@@ -1118,6 +1118,17 @@ message PAlterVaultSyncRequest {
message PAlterVaultSyncResponse {
}
+message PSyncTabletMetaRequest {
+ repeated int64 tablet_ids = 1;
+}
+
+message PSyncTabletMetaResponse {
+ optional PStatus status = 1;
+ optional int64 synced_tablets = 2;
+ optional int64 skipped_tablets = 3;
+ optional int64 failed_tablets = 4;
+}
+
message PGetBeResourceRequest {
}
@@ -1283,6 +1294,7 @@ service PBackendService {
rpc fetch_remote_tablet_schema(PFetchRemoteSchemaRequest) returns
(PFetchRemoteSchemaResponse);
rpc test_jdbc_connection(PJdbcTestConnectionRequest) returns
(PJdbcTestConnectionResult);
rpc alter_vault_sync(PAlterVaultSyncRequest) returns
(PAlterVaultSyncResponse);
+ rpc sync_tablet_meta(PSyncTabletMetaRequest) returns
(PSyncTabletMetaResponse);
rpc get_be_resource(PGetBeResourceRequest) returns
(PGetBeResourceResponse);
rpc delete_dictionary(PDeleteDictionaryRequest) returns
(PDeleteDictionaryResponse);
rpc commit_refresh_dictionary(PCommitRefreshDictionaryRequest) returns
(PCommitRefreshDictionaryResponse);
diff --git
a/regression-test/suites/fault_injection_p0/cloud/test_cloud_sync_tablet_meta_after_alter.groovy
b/regression-test/suites/fault_injection_p0/cloud/test_cloud_sync_tablet_meta_after_alter.groovy
new file mode 100644
index 00000000000..17bdeae2d12
--- /dev/null
+++
b/regression-test/suites/fault_injection_p0/cloud/test_cloud_sync_tablet_meta_after_alter.groovy
@@ -0,0 +1,284 @@
+// 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.
+
+suite("test_cloud_sync_tablet_meta_after_alter", "nonConcurrent") {
+ if (!isCloudMode()) {
+ return
+ }
+ if (!getFeConfig("enable_debug_points").equalsIgnoreCase("true")) {
+ logger.info("enable_debug_points=false, skip")
+ return
+ }
+ if (context.config.multiClusterBes == null ||
context.config.multiClusterBes.trim().isEmpty()) {
+ logger.info("multiClusterBes is empty, skip")
+ return
+ }
+
+ GetDebugPoint().clearDebugPointsForAllFEs()
+ GetDebugPoint().clearDebugPointsForAllBEs()
+
+ def cluster0 = "sync_meta_cluster0"
+ def cluster1 = "sync_meta_cluster1"
+ def clusterId0 = "sync_meta_cluster_id0"
+ def clusterId1 = "sync_meta_cluster_id1"
+ def dbName = "reg_sync_tablet_meta_db"
+ def tableName = "test_cloud_sync_tablet_meta_after_alter"
+ def versionLimitTableName =
"test_cloud_sync_tablet_meta_version_limit_after_alter"
+
+ List<String> ipList = []
+ List<String> hbPortList = []
+ List<String> beUniqueIdList = []
+
+ String[] bes = context.config.multiClusterBes.split(',')
+ for (String values : bes) {
+ if (ipList.size() == 2) {
+ break
+ }
+ String[] beInfo = values.split(':')
+ ipList.add(beInfo[0])
+ hbPortList.add(beInfo[1])
+ beUniqueIdList.add(beInfo[3])
+ }
+ assertTrue(ipList.size() >= 2, "need at least two backends in
multiClusterBes")
+
+ def getBackendForCluster = { String clusterName ->
+ def backend = sql_return_maparray("show backends").find {
it.Tag.contains(clusterName) }
+ assert backend != null: "backend for cluster ${clusterName} not found"
+ return backend
+ }
+
+ def getBvar = { backend, String name ->
+ def url = "http://${backend.Host}:${backend.BrpcPort}/vars/${name}";
+ def (code, out, err) = curl("GET", url)
+ assertEquals(0, code)
+ def matcher = (out =~
/(?m)^${java.util.regex.Pattern.quote(name)}\\s*:\\s*(\\d+)$/)
+ assertTrue(matcher.find(), "failed to parse bvar ${name} from ${url},
out=${out}, err=${err}")
+ return matcher.group(1).toLong()
+ }
+
+ def waitForBvarIncrease = { backend, String name, long baseline, long
delta ->
+ for (int retry = 0; retry < 20; retry++) {
+ def current = getBvar(backend, name)
+ logger.info("wait bvar ${name} on
${backend.Host}:${backend.BrpcPort}, baseline=${baseline}, current=${current}")
+ if (current >= baseline + delta) {
+ return current
+ }
+ sleep(500)
+ }
+ return getBvar(backend, name)
+ }
+
+ def getCompactionPolicy = { String currentTableName ->
+ def tablets = sql_return_maparray("""show tablets from
${currentTableName};""")
+ assertEquals(1, tablets.size())
+ def (code, out, err) = curl("GET", tablets[0].CompactionStatus)
+ assertEquals(0, code)
+ def tabletJson = parseJson(out.trim())
+ return tabletJson["compaction policy"]
+ }
+
+ def useCluster = { String clusterName ->
+ sql """use @${clusterName}"""
+ sql """use ${dbName}"""
+ }
+
+ def cleanupClusters = {
+ for (uniqueId in beUniqueIdList) {
+ def resp = get_cluster.call(uniqueId)
+ for (cluster in resp) {
+ if (cluster.type == "COMPUTE") {
+ drop_cluster.call(cluster.cluster_name, cluster.cluster_id)
+ }
+ }
+ }
+ wait_cluster_change()
+ }
+
+ setBeConfigTemporary([
+ tablet_sync_interval_s: 18000,
+ schedule_sync_tablets_interval_s: 18000
+ ]) {
+ try {
+ cleanupClusters.call()
+ add_cluster.call(beUniqueIdList[0], ipList[0], hbPortList[0],
cluster0, clusterId0)
+ add_cluster.call(beUniqueIdList[1], ipList[1], hbPortList[1],
cluster1, clusterId1)
+ wait_cluster_change()
+
+ sql """use @${cluster0}"""
+ sql """create database if not exists ${dbName}"""
+ sql """use ${dbName}"""
+ sql """drop table if exists ${tableName} force"""
+ sql """drop table if exists ${versionLimitTableName} force"""
+ sql """
+ CREATE TABLE ${tableName} (
+ id INT NOT NULL,
+ value INT
+ ) ENGINE=OLAP
+ DUPLICATE KEY(id)
+ DISTRIBUTED BY HASH(id) BUCKETS 1
+ PROPERTIES (
+ "replication_num" = "1",
+ "disable_auto_compaction" = "true"
+ )
+ """
+ sql """insert into ${tableName} values (1, 10), (2, 20), (3, 30)"""
+
+ def be0 = getBackendForCluster(cluster0)
+ def be1 = getBackendForCluster(cluster1)
+
+ // Case 1: warm only cluster0, alter property, and verify cluster1
is notified but skipped.
+ useCluster.call(cluster0)
+ qt_case1_cluster0_warmup """select count(*) from ${tableName}"""
+ assertEquals("size_based", getCompactionPolicy(tableName))
+
+ def case1Synced0 = getBvar(be0,
"cloud_sync_tablet_meta_synced_total")
+ def case1Skipped1 = getBvar(be1,
"cloud_sync_tablet_meta_skipped_total")
+
+ sql """alter table ${tableName} set ("compaction_policy" =
"time_series")"""
+
+ assertTrue(waitForBvarIncrease(be0,
"cloud_sync_tablet_meta_synced_total", case1Synced0, 1) >= case1Synced0 + 1)
+ assertTrue(waitForBvarIncrease(be1,
"cloud_sync_tablet_meta_skipped_total", case1Skipped1, 1) >= case1Skipped1 + 1)
+
+ useCluster.call(cluster0)
+ qt_case1_cluster0_after_alter """select count(*) from
${tableName}"""
+ assertEquals("time_series", getCompactionPolicy(tableName))
+
+ useCluster.call(cluster1)
+ qt_case1_cluster1_first_access """select count(*) from
${tableName}"""
+ assertEquals("time_series", getCompactionPolicy(tableName))
+
+ // Case 2: warm both clusters, alter property again, and verify
both clusters sync immediately.
+ useCluster.call(cluster0)
+ qt_case2_cluster0_warmup """select count(*) from ${tableName}"""
+ useCluster.call(cluster1)
+ qt_case2_cluster1_warmup """select count(*) from ${tableName}"""
+
+ def case2Synced0 = getBvar(be0,
"cloud_sync_tablet_meta_synced_total")
+ def case2Synced1 = getBvar(be1,
"cloud_sync_tablet_meta_synced_total")
+
+ useCluster.call(cluster0)
+ sql """alter table ${tableName} set ("compaction_policy" =
"size_based")"""
+
+ assertTrue(waitForBvarIncrease(be0,
"cloud_sync_tablet_meta_synced_total", case2Synced0, 1) >= case2Synced0 + 1)
+ assertTrue(waitForBvarIncrease(be1,
"cloud_sync_tablet_meta_synced_total", case2Synced1, 1) >= case2Synced1 + 1)
+
+ useCluster.call(cluster0)
+ assertEquals("size_based", getCompactionPolicy(tableName))
+ useCluster.call(cluster1)
+ assertEquals("size_based", getCompactionPolicy(tableName))
+
+ // Case 3: hit the size_based version limit first, then alter to
time_series and verify writes succeed immediately.
+ useCluster.call(cluster0)
+ sql """drop table if exists ${versionLimitTableName} force"""
+ sql """
+ CREATE TABLE ${versionLimitTableName} (
+ id BIGINT NOT NULL,
+ value BIGINT
+ ) ENGINE=OLAP
+ DUPLICATE KEY(id)
+ DISTRIBUTED BY HASH(id) BUCKETS 1
+ PROPERTIES (
+ "replication_num" = "1",
+ "disable_auto_compaction" = "true"
+ )
+ """
+ qt_case3_version_limit_warmup """select count(*) from
${versionLimitTableName}"""
+ assertEquals("size_based",
getCompactionPolicy(versionLimitTableName))
+
+ int successfulBeforeAlter = 0
+ String versionLimitFailure = null
+ for (int i = 0; i < 2400; i++) {
+ try {
+ sql """insert into ${versionLimitTableName} values (${i},
${i})"""
+ successfulBeforeAlter++
+ if ((i + 1) % 200 == 0) {
+ logger.info("inserted {} versions into {}", i + 1,
versionLimitTableName)
+ }
+ } catch (Exception e) {
+ versionLimitFailure = e.getMessage()
+ logger.info("hit version limit after {} successful
inserts, msg={}",
+ successfulBeforeAlter, versionLimitFailure)
+ break
+ }
+ }
+ assertTrue(versionLimitFailure != null, "expected version limit
failure before alter")
+ assertTrue(versionLimitFailure.contains("-235")
+ || versionLimitFailure.toLowerCase().contains("too many
versions")
+ || versionLimitFailure.toLowerCase().contains("version
count")
+ || versionLimitFailure.toLowerCase().contains("exceed
limit"),
+ "unexpected failure message: ${versionLimitFailure}")
+
+ def versionLimitTablets = sql_return_maparray("""show tablets from
${versionLimitTableName};""")
+ assertEquals(1, versionLimitTablets.size())
+ long versionCountBeforeAlter =
versionLimitTablets[0].VersionCount.toLong()
+ assertTrue(versionCountBeforeAlter >= 2000,
+ "expected version count near max limit,
actual=${versionCountBeforeAlter}")
+
+ sql """alter table ${versionLimitTableName} set
("compaction_policy" = "time_series")"""
+ assertEquals("time_series",
getCompactionPolicy(versionLimitTableName))
+
+ int successfulAfterAlter = 0
+ for (int i = 0; i < 50; i++) {
+ long rowId = 100000L + i
+ sql """insert into ${versionLimitTableName} values (${rowId},
${rowId})"""
+ successfulAfterAlter++
+ }
+ assertEquals(50, successfulAfterAlter)
+ def versionLimitTabletsAfterAlter = sql_return_maparray("""show
tablets from ${versionLimitTableName};""")
+ assertEquals(1, versionLimitTabletsAfterAlter.size())
+ assertTrue(versionLimitTabletsAfterAlter[0].VersionCount.toLong()
> versionCountBeforeAlter,
+ "expected version count to keep increasing immediately
after alter")
+ qt_case3_version_limit_after_alter """select count(*) from
${versionLimitTableName}"""
+
+ // Case 4: disable FE proactive notification and verify cached
tablets remain stale.
+
GetDebugPoint().enableDebugPointForAllFEs("CloudSchemaChangeHandler.notifyBackendsToSyncTabletMeta.skip")
+ def case4Request0 = getBvar(be0,
"cloud_sync_tablet_meta_requests_total")
+ def case4Request1 = getBvar(be1,
"cloud_sync_tablet_meta_requests_total")
+ def case4Synced0 = getBvar(be0,
"cloud_sync_tablet_meta_synced_total")
+ def case4Synced1 = getBvar(be1,
"cloud_sync_tablet_meta_synced_total")
+ def case4Skipped0 = getBvar(be0,
"cloud_sync_tablet_meta_skipped_total")
+ def case4Skipped1 = getBvar(be1,
"cloud_sync_tablet_meta_skipped_total")
+
+ useCluster.call(cluster0)
+ sql """alter table ${tableName} set ("compaction_policy" =
"time_series")"""
+ sleep(2000)
+
+ assertEquals(case4Request0, getBvar(be0,
"cloud_sync_tablet_meta_requests_total"))
+ assertEquals(case4Request1, getBvar(be1,
"cloud_sync_tablet_meta_requests_total"))
+ assertEquals(case4Synced0, getBvar(be0,
"cloud_sync_tablet_meta_synced_total"))
+ assertEquals(case4Synced1, getBvar(be1,
"cloud_sync_tablet_meta_synced_total"))
+ assertEquals(case4Skipped0, getBvar(be0,
"cloud_sync_tablet_meta_skipped_total"))
+ assertEquals(case4Skipped1, getBvar(be1,
"cloud_sync_tablet_meta_skipped_total"))
+
+ useCluster.call(cluster0)
+ assertEquals("size_based", getCompactionPolicy(tableName))
+ useCluster.call(cluster1)
+ assertEquals("size_based", getCompactionPolicy(tableName))
+ } finally {
+ GetDebugPoint().clearDebugPointsForAllFEs()
+ GetDebugPoint().clearDebugPointsForAllBEs()
+ try {
+ sql """use @${cluster0}"""
+ sql """drop table if exists ${dbName}.${tableName} force"""
+ sql """drop table if exists ${dbName}.${versionLimitTableName}
force"""
+ } catch (Exception e) {
+ logger.info("drop table in finally failed: ${e.getMessage()}")
+ }
+ cleanupClusters.call()
+ }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
