Skip to content

branch-4.1: [enhance](cloud) proactively sync tablet meta after alter #61585 #66798

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
github-actions wants to merge 1 commit into
base: branch-4.1
Choose a base branch
from
Open

branch-4.1: [enhance](cloud) proactively sync tablet meta after alter #61585 #66798

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
77 changes: 77 additions & 0 deletions be/src/cloud/cloud_internal_service.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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};
Expand All @@ -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,
Expand Down
8 changes: 8 additions & 0 deletions be/src/cloud/cloud_internal_service.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand All @@ -48,6 +52,10 @@ class CloudInternalServiceImpl final : public PInternalService {
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,
Expand Down
95 changes: 54 additions & 41 deletions be/src/cloud/cloud_tablet.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -954,6 +954,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());
Expand Down Expand Up @@ -1463,9 +1469,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);
Expand All @@ -1477,57 +1483,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);
}
}

return Status::OK();
Expand Down
4 changes: 4 additions & 0 deletions be/src/cloud/cloud_tablet_mgr.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -287,6 +287,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());
Expand Down
2 changes: 2 additions & 0 deletions be/src/cloud/cloud_tablet_mgr.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -62,6 +62,8 @@ class CloudTabletMgr {
// 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);
Expand Down
9 changes: 9 additions & 0 deletions be/src/service/internal_service.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2568,5 +2568,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
4 changes: 4 additions & 0 deletions be/src/service/internal_service.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -223,6 +223,10 @@ class PInternalService : public PBackendService {
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,
Expand Down
Loading
Loading