-
Notifications
You must be signed in to change notification settings - Fork 3.9k
branch-4.1: [feat](cache-memory) add external metadata cache memory governance #66717
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.
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
Changes from all commits
b89e044
41e63e0
6443730
3d97895
053a91a
c302b68
b8fcc6d
ec807ce
f5b9250
29ffdd6
b6e8939
950d87d
f681235
66a94dd
819a391
34a96a0
6b8415c
ba960d8
3c53138
b8d6f7e
7789a16
7ad7815
e97404b
233b501
b59da07
94da403
a85a3e5
da6f21e
1b62b6b
02002f7
76ecb3e
d18acde
9c193c7
ad5a42d
7fe5e76
da14048
8cb50ad
fc76ea6
a4cbf89
246b544
1d22313
1057395
bfca15a
b090024
e2e998c
a25dbfc
2df6c43
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -53,8 +53,24 @@ 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}, | ||
| {"WEIGHT_BOUNDED", TYPE_BOOLEAN, sizeof(bool), true}, | ||
| {"MAX_WEIGHT", TYPE_BIGINT, sizeof(int64_t), true}, | ||
| {"ESTIMATED_WEIGHT", TYPE_BIGINT, sizeof(int64_t), true}, | ||
| {"EVICTION_WEIGHT", TYPE_BIGINT, sizeof(int64_t), true}, | ||
| {"WEIGHT_REJECT_COUNT", TYPE_BIGINT, sizeof(int64_t), true}, | ||
| {"CATALOG_MAX_WEIGHT", TYPE_BIGINT, sizeof(int64_t), true}, | ||
| {"CATALOG_ESTIMATED_WEIGHT", TYPE_BIGINT, sizeof(int64_t), true}, | ||
| {"GLOBAL_MAX_WEIGHT", TYPE_BIGINT, sizeof(int64_t), true}, | ||
| {"GLOBAL_ESTIMATED_WEIGHT", TYPE_BIGINT, sizeof(int64_t), true}, | ||
| {"LAST_WEIGHT_REJECT_REASON", TYPE_STRING, sizeof(StringRef), true}, | ||
| }; | ||
|
|
||
| // Columns that every FE knows. The weight statistics columns appended after LAST_ERROR are | ||
| // only served by FEs that carry the memory-governance change; during a rolling upgrade an older | ||
| // FE rejects a projection that names them, so the scanner falls back to this prefix and leaves | ||
| // the newer columns NULL. | ||
| static constexpr size_t kLegacyMetaCacheStatsColumnCount = 23; | ||
|
|
||
| SchemaCatalogMetaCacheStatsScanner::SchemaCatalogMetaCacheStatsScanner() | ||
| : SchemaScanner(_s_tbls_columns, TSchemaTableType::SCH_CATALOG_META_CACHE_STATISTICS) {} | ||
|
|
||
|
|
@@ -67,9 +83,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); | ||
| } | ||
|
|
@@ -79,20 +98,47 @@ 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; | ||
|
|
||
| // A transport-level failure says nothing about the FE's column support and must be | ||
| // surfaced to the caller instead of being retried with the legacy projection. | ||
| 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 fe_status = Status::create(result->status); | ||
| // The RPC itself succeeded, so a non-OK status was produced by the FE handler: this is the | ||
| // signal a legacy FE without the weight columns emits for an unknown projection name. | ||
| *fe_rejected = !fe_status.ok(); | ||
| return fe_status; | ||
| } | ||
|
|
||
| Status status(Status::create(result.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()) { | ||
|
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. [P2] Restrict the legacy fallback to unsupported columns This retries the 23-column projection for every non-OK result, including transport errors and unrelated failures from a current FE. If that immediate retry succeeds, the scanner treats the server as legacy and returns all ten supported weight columns as NULL, masking the original failure and silently hiding quota/rejection telemetry. Please select the legacy request from an explicit FE capability/version (or a positively identified unsupported-column status) and preserve unrelated first-request errors; add a current-FE transient-error regression alongside the old-FE fallback case. |
||
| LOG(WARNING) << "fetch catalog meta cache stats from FE(" << _fe_addr.hostname | ||
| << ") failed, errmsg=" << status; | ||
| return status; | ||
| if (!fe_rejected) { | ||
| // Transport error from a current FE: do not mask it with a legacy retry that would | ||
| // silently blank the weight telemetry columns. | ||
| LOG(WARNING) << "fetch catalog meta cache stats from FE(" << _fe_addr.hostname | ||
| << ") failed, errmsg=" << status; | ||
| return status; | ||
| } | ||
| Status first_status = status; | ||
| LOG(INFO) << "FE(" << _fe_addr.hostname | ||
| << ") rejected the full catalog meta cache stats projection, retrying with the " | ||
| "legacy column set: " | ||
| << first_status; | ||
| result = TFetchSchemaTableDataResult(); | ||
| status = _fetch_from_fe(kLegacyMetaCacheStatsColumnCount, &result, &fe_rejected); | ||
| if (!status.ok()) { | ||
| // The legacy projection failed too, so the first rejection was not a legacy FE; | ||
| // surface the original error rather than the retry artifact. | ||
| 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; | ||
|
|
||
|
|
@@ -106,19 +152,29 @@ Status SchemaCatalogMetaCacheStatsScanner::_get_meta_cache_from_fe() { | |
|
|
||
| _block->reserve(_block_rows_limit); | ||
|
|
||
| size_t col_size = _s_tbls_columns.size(); | ||
| if (result_data.size() > 0) { | ||
| auto col_size = result_data[0].column_value.size(); | ||
| if (col_size != _s_tbls_columns.size()) { | ||
| 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 { | ||
| // Column unknown to the serving FE: NULL. | ||
| auto column_guard = _block->mutate_column_scoped(j); | ||
| column_guard.mutable_column()->insert_default(); | ||
| column_guard.restore(); | ||
| } | ||
| } | ||
| } | ||
| return Status::OK(); | ||
|
|
||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,42 @@ | ||
| <?xml version="1.0" encoding="UTF-8"?> | ||
| <!-- | ||
| 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. | ||
| --> | ||
| <project xmlns="http://maven.apache.org/POM/4.0.0" | ||
| xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" | ||
| xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd"> | ||
| <modelVersion>4.0.0</modelVersion> | ||
|
|
||
| <parent> | ||
| <groupId>org.apache.doris</groupId> | ||
| <artifactId>fe</artifactId> | ||
| <version>${revision}</version> | ||
| <relativePath>../pom.xml</relativePath> | ||
| </parent> | ||
|
|
||
| <artifactId>fe-benchmark</artifactId> | ||
| <name>Doris FE Benchmarks</name> | ||
|
|
||
| <dependencies> | ||
| <dependency> | ||
| <groupId>${project.groupId}</groupId> | ||
| <artifactId>fe-core</artifactId> | ||
| <version>${project.version}</version> | ||
| </dependency> | ||
| </dependencies> | ||
| </project> |
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,59 @@ | ||
| #!/usr/bin/env bash | ||
|
|
||
| # 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. | ||
|
|
||
| set -euo pipefail | ||
|
|
||
| BENCHMARK_DIR=$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")" && pwd) | ||
| FE_DIR=$(cd -- "${BENCHMARK_DIR}/.." && pwd) | ||
| CLASSPATH_FILE=$(mktemp) | ||
| trap 'rm -f "${CLASSPATH_FILE}"' EXIT | ||
|
|
||
| ( | ||
| cd "${FE_DIR}" | ||
| mvn -Pbenchmark -pl fe-benchmark -am compile -DskipTests -Dskip.clean=true | ||
| mvn -Pbenchmark -pl fe-benchmark -am dependency:build-classpath \ | ||
| -Dskip.clean=true \ | ||
| -DincludeScope=test \ | ||
| -Dmdep.outputFile="${CLASSPATH_FILE}" | ||
| ) | ||
|
|
||
| REACTOR_CLASSES= | ||
| while IFS= read -r -d '' CLASSES_DIR; do | ||
| REACTOR_CLASSES+="${CLASSES_DIR}:" | ||
| done < <(find "${FE_DIR}" -type d -path '*/target/classes' -print0) | ||
| DEPENDENCY_CLASSES=$(tr -d '\n' < "${CLASSPATH_FILE}") | ||
| BENCHMARK_FILTER=${BENCHMARK_FILTER:-'HivePartitionValuesSizeBenchmark|IcebergCacheSizeBenchmark|PaimonCacheSizeBenchmark|MetaCacheSoftValueBenchmark'} | ||
|
|
||
| BENCHMARK_CLASSES=( | ||
| org.apache.doris.datasource.hive.HivePartitionValuesSizeBenchmark | ||
| org.apache.doris.datasource.iceberg.IcebergCacheSizeBenchmark | ||
| org.apache.doris.datasource.paimon.PaimonCacheSizeBenchmark | ||
| org.apache.doris.datasource.metacache.MetaCacheSoftValueBenchmark | ||
| ) | ||
|
|
||
| for BENCHMARK_CLASS in "${BENCHMARK_CLASSES[@]}"; do | ||
| if [[ "${BENCHMARK_CLASS##*.}" =~ ${BENCHMARK_FILTER} ]]; then | ||
| java \ | ||
| -Xms1g \ | ||
| -Xmx4g \ | ||
| -classpath "${REACTOR_CLASSES}${DEPENDENCY_CLASSES}" \ | ||
| "${BENCHMARK_CLASS}" \ | ||
| "$@" | ||
| fi | ||
| done |
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,101 @@ | ||
| // 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.benchmark; | ||
|
|
||
| import java.util.Locale; | ||
| import java.util.concurrent.TimeUnit; | ||
|
|
||
| /** Small dependency-free harness for opt-in FE microbenchmarks. */ | ||
| public final class BenchmarkHarness { | ||
| private static final long WARMUP_MILLIS = Long.getLong("benchmark.warmup.millis", 500L); | ||
| private static final long MEASUREMENT_MILLIS = Long.getLong("benchmark.measurement.millis", 500L); | ||
| private static final int MEASUREMENT_ITERATIONS = Integer.getInteger("benchmark.iterations", 3); | ||
| private static final boolean PRINT_RESULT = Boolean.getBoolean("benchmark.print.result"); | ||
| private static volatile Object sink; | ||
|
|
||
| private BenchmarkHarness() { | ||
| } | ||
|
|
||
| @FunctionalInterface | ||
| public interface Operation { | ||
| Object run() throws Exception; | ||
| } | ||
|
|
||
| public static void measure(String name, TimeUnit outputUnit, Operation operation) throws Exception { | ||
| runWindow(operation, WARMUP_MILLIS); | ||
| double totalNanosPerOperation = 0.0D; | ||
| long totalOperations = 0L; | ||
| for (int iteration = 0; iteration < MEASUREMENT_ITERATIONS; iteration++) { | ||
| Window result = runWindow(operation, MEASUREMENT_MILLIS); | ||
| totalNanosPerOperation += result.nanosPerOperation; | ||
| totalOperations += result.operations; | ||
| } | ||
| double averageNanos = totalNanosPerOperation / MEASUREMENT_ITERATIONS; | ||
| String result = PRINT_RESULT ? ", result=" + sink : ""; | ||
| System.out.printf(Locale.ROOT, "%-72s %12.3f %s/op (%d ops%s)%n", | ||
| name, convertFromNanos(averageNanos, outputUnit), unitName(outputUnit), totalOperations, result); | ||
| } | ||
|
|
||
| private static Window runWindow(Operation operation, long minimumMillis) throws Exception { | ||
| long start = System.nanoTime(); | ||
| long deadline = start + TimeUnit.MILLISECONDS.toNanos(minimumMillis); | ||
| long operations = 0L; | ||
| do { | ||
| sink = operation.run(); | ||
| operations++; | ||
| } while (System.nanoTime() < deadline); | ||
| long elapsed = System.nanoTime() - start; | ||
| return new Window(operations, (double) elapsed / operations); | ||
| } | ||
|
|
||
| private static double convertFromNanos(double nanos, TimeUnit outputUnit) { | ||
| if (outputUnit == TimeUnit.NANOSECONDS) { | ||
| return nanos; | ||
| } else if (outputUnit == TimeUnit.MICROSECONDS) { | ||
| return nanos / 1_000.0D; | ||
| } else if (outputUnit == TimeUnit.MILLISECONDS) { | ||
| return nanos / 1_000_000.0D; | ||
| } else if (outputUnit == TimeUnit.SECONDS) { | ||
| return nanos / 1_000_000_000.0D; | ||
| } | ||
| throw new IllegalArgumentException("unsupported benchmark time unit: " + outputUnit); | ||
| } | ||
|
|
||
| private static String unitName(TimeUnit outputUnit) { | ||
| if (outputUnit == TimeUnit.NANOSECONDS) { | ||
| return "ns"; | ||
| } else if (outputUnit == TimeUnit.MICROSECONDS) { | ||
| return "us"; | ||
| } else if (outputUnit == TimeUnit.MILLISECONDS) { | ||
| return "ms"; | ||
| } else if (outputUnit == TimeUnit.SECONDS) { | ||
| return "s"; | ||
| } | ||
| return outputUnit.name().toLowerCase(Locale.ROOT); | ||
| } | ||
|
|
||
| private static final class Window { | ||
| private final long operations; | ||
| private final double nanosPerOperation; | ||
|
|
||
| private Window(long operations, double nanosPerOperation) { | ||
| this.operations = operations; | ||
| this.nanosPerOperation = nanosPerOperation; | ||
| } | ||
| } | ||
| } |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
[P2] Preserve this RPC across a rolling upgrade
A new BE sends every local column name to fetchSchemaTableData, but an old FE's META_CACHE_STATS_COLUMN_TO_INDEX has none of this appended suffix; its filterColumns() looks up a null index and throws. The inverse pair also cannot execute a new-column projection on an old BE descriptor. Please add a version-gated legacy request/fallback that fills unsupported cells with NULL/defaults (and gate new-column planning as needed), with mixed-version tests in both directions.