Skip to content

[feature](fe) Add external metadata cache memory governance #67726

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
CalvinKirs wants to merge 1 commit into
base: master
Choose a base branch
from
Open

[feature](fe) Add external metadata cache memory governance #67726

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
66 changes: 50 additions & 16 deletions be/src/information_schema/schema_catalog_meta_cache_stats_scanner.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -53,8 +53,14 @@ std::vector<SchemaScanner::ColumnDesc> SchemaCatalogMetaCacheStatsScanner::_s_tb
{"LAST_LOAD_SUCCESS_TIME", TYPE_STRING, sizeof(StringRef), true},
{"LAST_LOAD_FAILURE_TIME", TYPE_STRING, sizeof(StringRef), true},
{"LAST_ERROR", TYPE_STRING, sizeof(StringRef), true},
{"MAX_WEIGHT", TYPE_BIGINT, sizeof(int64_t), true},
{"ESTIMATED_WEIGHT", TYPE_BIGINT, sizeof(int64_t), true},
{"WEIGHT_REJECT_COUNT", TYPE_BIGINT, sizeof(int64_t), true},
{"LAST_WEIGHT_REJECT_REASON", TYPE_STRING, sizeof(StringRef), true},
};

static constexpr size_t kLegacyMetaCacheStatsColumnCount = 24;

SchemaCatalogMetaCacheStatsScanner::SchemaCatalogMetaCacheStatsScanner()
: SchemaScanner(_s_tbls_columns, TSchemaTableType::SCH_CATALOG_META_CACHE_STATISTICS) {}

Expand All @@ -67,9 +73,12 @@ Status SchemaCatalogMetaCacheStatsScanner::start(RuntimeState* state) {
return Status::OK();
}

Status SchemaCatalogMetaCacheStatsScanner::_get_meta_cache_from_fe() {
Status SchemaCatalogMetaCacheStatsScanner::_fetch_from_fe(size_t column_count,
TFetchSchemaTableDataResult* result,
bool* fe_rejected) {
*fe_rejected = false;
TSchemaTableRequestParams schema_table_request_params;
for (int i = 0; i < _s_tbls_columns.size(); i++) {
for (size_t i = 0; i < column_count; i++) {
schema_table_request_params.__isset.columns_name = true;
schema_table_request_params.columns_name.emplace_back(_s_tbls_columns[i].name);
}
Expand All @@ -79,20 +88,36 @@ Status SchemaCatalogMetaCacheStatsScanner::_get_meta_cache_from_fe() {
request.__set_schema_table_name(TSchemaTableName::CATALOG_META_CACHE_STATS);
request.__set_schema_table_params(schema_table_request_params);

TFetchSchemaTableDataResult result;

RETURN_IF_ERROR(ThriftRpcHelper::rpc<FrontendServiceClient>(
_fe_addr.hostname, _fe_addr.port,
[&request, &result](FrontendServiceConnection& client) {
client->fetchSchemaTableData(result, request);
[&request, result](FrontendServiceConnection& client) {
client->fetchSchemaTableData(*result, request);
},
_rpc_timeout));

Status status(Status::create(result.status));
Status fe_status = Status::create(result->status);
*fe_rejected = !fe_status.ok();
return fe_status;
}

Status SchemaCatalogMetaCacheStatsScanner::_get_meta_cache_from_fe() {
TFetchSchemaTableDataResult result;
bool fe_rejected = false;
Status status = _fetch_from_fe(_s_tbls_columns.size(), &result, &fe_rejected);
if (!status.ok()) {
LOG(WARNING) << "fetch catalog meta cache stats from FE(" << _fe_addr.hostname
<< ") failed, errmsg=" << status;
return status;
if (!fe_rejected) {
LOG(WARNING) << "fetch catalog meta cache stats from FE(" << _fe_addr.hostname
<< ") failed, errmsg=" << status;
return status;
}
Status first_status = status;
result = TFetchSchemaTableDataResult();
status = _fetch_from_fe(kLegacyMetaCacheStatsColumnCount, &result, &fe_rejected);
if (!status.ok()) {
LOG(WARNING) << "fetch catalog meta cache stats from FE(" << _fe_addr.hostname
<< ") failed, errmsg=" << first_status;
return first_status;
}
}
std::vector<TRow> result_data = result.data_batch;

Expand All @@ -106,19 +131,28 @@ Status SchemaCatalogMetaCacheStatsScanner::_get_meta_cache_from_fe() {

_block->reserve(_block_rows_limit);

if (result_data.size() > 0) {
auto col_size = result_data[0].column_value.size();
if (col_size != _s_tbls_columns.size()) {
size_t col_size = _s_tbls_columns.size();
if (!result_data.empty()) {
col_size = result_data[0].column_value.size();
if (col_size != _s_tbls_columns.size() && col_size != kLegacyMetaCacheStatsColumnCount) {
return Status::InternalError<false>(
"catalog meta cache stats schema is not match for FE and BE");
}
}

int available_columns = static_cast<int>(col_size);
int total_columns = static_cast<int>(_s_tbls_columns.size());
for (int i = 0; i < result_data.size(); i++) {
TRow row = result_data[i];
for (int j = 0; j < _s_tbls_columns.size(); j++) {
RETURN_IF_ERROR(insert_block_column(row.column_value[j], j, _block.get(),
_s_tbls_columns[j].type));
for (int j = 0; j < total_columns; j++) {
if (j < available_columns) {
RETURN_IF_ERROR(insert_block_column(row.column_value[j], j, _block.get(),
_s_tbls_columns[j].type));
} else {
auto column_guard = _block->mutate_column_scoped(j);
column_guard.mutable_column()->insert_default();
column_guard.restore();
}
}
}
return Status::OK();
Expand Down
3 changes: 3 additions & 0 deletions be/src/information_schema/schema_catalog_meta_cache_stats_scanner.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,7 @@
namespace doris {
class RuntimeState;
class Block;
class TFetchSchemaTableDataResult;

class SchemaCatalogMetaCacheStatsScanner : public SchemaScanner {
ENABLE_FACTORY_CREATOR(SchemaCatalogMetaCacheStatsScanner);
Expand All @@ -40,6 +41,8 @@ class SchemaCatalogMetaCacheStatsScanner : public SchemaScanner {

private:
Status _get_meta_cache_from_fe();
Status _fetch_from_fe(size_t column_count, TFetchSchemaTableDataResult* result,
bool* fe_rejected);

TNetworkAddress _fe_addr;

Expand Down
5 changes: 5 additions & 0 deletions fe/fe-common/src/main/java/org/apache/doris/common/Config.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2204,6 +2204,11 @@ public class Config extends ConfigBase {
@ConfField(description = "The auto-refresh interval of the external meta cache.")
public static long external_cache_refresh_time_minutes = 10; // 10 mins

@ConfField(mutable = false, masterOnly = false,
description = "FE-wide maximum weight for managed external metadata caches. Supports byte units "
+ "or a percentage of the JVM max heap; 0 disables the global quota.")
public static String external_meta_cache_max_weight = "0";

// Enable manual miss load for external meta cache to avoid blocking replayer on slow loaders.
@ConfField(mutable = true, masterOnly = false,
description = "Whether external meta cache uses manual miss load instead of Caffeine sync load.")
Expand Down
Loading
Loading