branch-4.1: [feat](cache-memory) add external metadata cache memory governance

be/src/information_schema/schema_catalog_meta_cache_stats_scanner.cpp

@@ -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,10 @@ 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) {
     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 +96,28 @@ 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));
+    return Status::create(result->status);
+}
 
-    Status status(Status::create(result.status));
+Status SchemaCatalogMetaCacheStatsScanner::_get_meta_cache_from_fe() {
+    TFetchSchemaTableDataResult result;
+    Status status = _fetch_from_fe(_s_tbls_columns.size(), &result);
     if (!status.ok()) {
-        LOG(WARNING) << "fetch catalog meta cache stats from FE(" << _fe_addr.hostname
-                     << ") failed, errmsg=" << status;
-        return status;
+        LOG(INFO) << "fetch catalog meta cache stats from FE(" << _fe_addr.hostname
+                  << ") with all columns failed, retrying with the legacy column set: " << status;
+        result = TFetchSchemaTableDataResult();
+        status = _fetch_from_fe(kLegacyMetaCacheStatsColumnCount, &result);
+        if (!status.ok()) {
+            LOG(WARNING) << "fetch catalog meta cache stats from FE(" << _fe_addr.hostname
+                         << ") failed, errmsg=" << status;
+            return status;
+        }
     }
     std::vector<TRow> result_data = result.data_batch;
 
@@ -106,19 +131,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();

be/src/information_schema/schema_catalog_meta_cache_stats_scanner.h

@@ -25,6 +25,7 @@
 namespace doris {
 class RuntimeState;
 class Block;
+class TFetchSchemaTableDataResult;
 
 class SchemaCatalogMetaCacheStatsScanner : public SchemaScanner {
     ENABLE_FACTORY_CREATOR(SchemaCatalogMetaCacheStatsScanner);
@@ -40,6 +41,7 @@ class SchemaCatalogMetaCacheStatsScanner : public SchemaScanner {
 
 private:
     Status _get_meta_cache_from_fe();
+    Status _fetch_from_fe(size_t column_count, TFetchSchemaTableDataResult* result);
 
     TNetworkAddress _fe_addr;
 

fe/fe-benchmark/pom.xml

@@ -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>

fe/fe-benchmark/run-external-meta-cache-size-benchmark.sh

@@ -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

fe/fe-benchmark/src/main/java/org/apache/doris/benchmark/BenchmarkHarness.java

@@ -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;
+        }
+    }
+}