diff --git a/fe/fe-benchmark/pom.xml b/fe/fe-benchmark/pom.xml
new file mode 100644
index 00000000000000..2de619284a9f69
--- /dev/null
+++ b/fe/fe-benchmark/pom.xml
@@ -0,0 +1,42 @@
+
+
+
+ 4.0.0
+
+
+ org.apache.doris
+ fe
+ ${revision}
+ ../pom.xml
+
+
+ fe-benchmark
+ Doris FE Benchmarks
+
+
+
+ ${project.groupId}
+ fe-core
+ ${project.version}
+
+
+
diff --git a/fe/fe-benchmark/run-external-meta-cache-size-benchmark.sh b/fe/fe-benchmark/run-external-meta-cache-size-benchmark.sh
new file mode 100755
index 00000000000000..7ce2e0be8e3d4f
--- /dev/null
+++ b/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
diff --git a/fe/fe-benchmark/src/main/java/org/apache/doris/benchmark/BenchmarkHarness.java b/fe/fe-benchmark/src/main/java/org/apache/doris/benchmark/BenchmarkHarness.java
new file mode 100644
index 00000000000000..c1e9234c70fa67
--- /dev/null
+++ b/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;
+ }
+ }
+}
diff --git a/fe/fe-benchmark/src/main/java/org/apache/doris/datasource/hive/HivePartitionValuesSizeBenchmark.java b/fe/fe-benchmark/src/main/java/org/apache/doris/datasource/hive/HivePartitionValuesSizeBenchmark.java
new file mode 100644
index 00000000000000..7e3acd9e2f97ac
--- /dev/null
+++ b/fe/fe-benchmark/src/main/java/org/apache/doris/datasource/hive/HivePartitionValuesSizeBenchmark.java
@@ -0,0 +1,295 @@
+// 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.datasource.hive;
+
+import org.apache.doris.analysis.PartitionValue;
+import org.apache.doris.benchmark.BenchmarkHarness;
+import org.apache.doris.catalog.ListPartitionItem;
+import org.apache.doris.catalog.PartitionItem;
+import org.apache.doris.catalog.PartitionKey;
+import org.apache.doris.catalog.Type;
+import org.apache.doris.datasource.NameMapping;
+import org.apache.doris.datasource.hive.HiveExternalMetaCache.HivePartitionValues;
+import org.apache.doris.datasource.hive.HiveExternalMetaCache.PartitionValueCacheKey;
+import org.apache.doris.datasource.metacache.CacheSpec;
+import org.apache.doris.datasource.metacache.ExternalMetaCacheBudgetManager;
+import org.apache.doris.datasource.metacache.ExternalMetaCacheBudgetManager.EntryBudget;
+import org.apache.doris.datasource.metacache.MetaCacheEntry;
+import org.apache.doris.datasource.metacache.MetaCacheSizeEstimate;
+
+import com.google.common.collect.HashBiMap;
+import com.google.common.collect.Maps;
+import com.google.common.util.concurrent.MoreExecutors;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.OptionalLong;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.TimeUnit;
+
+/** Measures count publication, weighted preparation, and prepared-value admission separately. */
+public class HivePartitionValuesSizeBenchmark {
+ private static final int TAIL_PAYLOAD_BYTES = 1024 * 1024;
+ private static final long MAX_WEIGHT_BYTES = 4L * 1024L * 1024L * 1024L;
+
+ public int countPublicationBaseline(UnsealedState state) {
+ state.partitionValues.rebuildSortedPartitionRangesForPublication();
+ return state.partitionValues.getSortedPartitionRanges()
+ .map(ranges -> ranges.sortedPartitions.size() + ranges.defaultPartitions.size())
+ .orElse(0);
+ }
+
+ public int sealPublicationWithoutEstimate(UnsealedState state) {
+ state.partitionValues.sealForPublication();
+ return state.partitionValues.getIdToPartitionItem().size();
+ }
+
+ public long weightedPublication(UnsealedState state) {
+ state.partitionValues.rebuildSortedPartitionRangesForPublication();
+ state.partitionValues.prepareForCachePublication(state.key);
+ return requireComplete(state.partitionValues);
+ }
+
+ public long eventCopySealAndEstimate(PreparedState state) {
+ HivePartitionValues copy = state.partitionValues.mutableCopy();
+ copy.rebuildSortedPartitionRangesForPublication();
+ copy.prepareForCachePublication(state.key);
+ return requireComplete(copy);
+ }
+
+ public long preparedSizeProvider(PreparedState state) {
+ return state.partitionValues.getSizeEstimate().getBytes();
+ }
+
+ public long estimateFormula(PreparedState state) {
+ state.partitionValues.prepareSizeEstimate(state.key);
+ return requireComplete(state.partitionValues);
+ }
+
+ public void replacementAdmission(PreparedState state) {
+ MetaCacheEntry.ReplaceResult result = state.cacheEntry.tryReplace(
+ state.key, state.currentPartitionValues, state.nextPartitionValues);
+ if (result != MetaCacheEntry.ReplaceResult.REPLACED) {
+ throw new IllegalStateException("replacement failed: " + result);
+ }
+ state.currentPartitionValues = state.nextPartitionValues;
+ state.nextPartitionValues = state.nextPartitionValues == state.partitionValues
+ ? state.replacementPartitionValues : state.partitionValues;
+ }
+
+ public HivePartitionValues countStrongCacheHit(PreparedState state) {
+ return state.countCacheEntry.getIfPresent(state.key);
+ }
+
+ public HivePartitionValues weightedSoftCacheHit(PreparedState state) {
+ return state.cacheEntry.getIfPresent(state.key);
+ }
+
+ public static void main(String[] args) throws Exception {
+ HivePartitionValuesSizeBenchmark benchmark = new HivePartitionValuesSizeBenchmark();
+ for (int partitionCount : new int[] {1000, 10000, 100000}) {
+ for (String distribution : new String[] {"uniform", "tail_skew"}) {
+ String suffix = "[partitions=" + partitionCount + ",distribution=" + distribution + "]";
+ BenchmarkHarness.measure("hive.countPublicationBaseline" + suffix, TimeUnit.MILLISECONDS, () -> {
+ UnsealedState state = unsealedState(partitionCount, distribution);
+ return benchmark.countPublicationBaseline(state);
+ });
+ BenchmarkHarness.measure("hive.sealPublicationWithoutEstimate" + suffix,
+ TimeUnit.MILLISECONDS, () -> {
+ UnsealedState state = unsealedState(partitionCount, distribution);
+ return benchmark.sealPublicationWithoutEstimate(state);
+ });
+ BenchmarkHarness.measure("hive.weightedPublication" + suffix, TimeUnit.MILLISECONDS, () -> {
+ UnsealedState state = unsealedState(partitionCount, distribution);
+ return benchmark.weightedPublication(state);
+ });
+
+ PreparedState prepared = new PreparedState();
+ prepared.partitionCount = partitionCount;
+ prepared.distribution = distribution;
+ prepared.setup();
+ try {
+ BenchmarkHarness.measure("hive.eventCopySealAndEstimate" + suffix,
+ TimeUnit.MILLISECONDS, () -> benchmark.eventCopySealAndEstimate(prepared));
+ BenchmarkHarness.measure("hive.preparedSizeProvider" + suffix,
+ TimeUnit.NANOSECONDS, () -> benchmark.preparedSizeProvider(prepared));
+ BenchmarkHarness.measure("hive.estimateFormula" + suffix,
+ TimeUnit.NANOSECONDS, () -> benchmark.estimateFormula(prepared));
+ BenchmarkHarness.measure("hive.countStrongCacheHit" + suffix,
+ TimeUnit.NANOSECONDS, () -> benchmark.countStrongCacheHit(prepared));
+ BenchmarkHarness.measure("hive.weightedSoftCacheHit" + suffix,
+ TimeUnit.NANOSECONDS, () -> benchmark.weightedSoftCacheHit(prepared));
+ BenchmarkHarness.measure("hive.replacementAdmission" + suffix,
+ TimeUnit.NANOSECONDS, () -> {
+ benchmark.replacementAdmission(prepared);
+ return null;
+ });
+ } finally {
+ prepared.tearDown();
+ }
+ }
+ }
+ }
+
+ private static UnsealedState unsealedState(int partitionCount, String distribution) throws Exception {
+ UnsealedState state = new UnsealedState();
+ state.partitionCount = partitionCount;
+ state.distribution = distribution;
+ state.setupInvocation();
+ return state;
+ }
+
+ /** Fresh graph per invocation so all publication work stays inside the measured method. */
+ public static class UnsealedState {
+ public int partitionCount;
+
+ public String distribution;
+
+ private PartitionValueCacheKey key;
+ private HivePartitionValues partitionValues;
+
+ public void setupInvocation() throws Exception {
+ List types = benchmarkTypes();
+ key = benchmarkKey(types);
+ partitionValues = createPartitionValues(partitionCount, distribution, types);
+ }
+ }
+
+ public static class PreparedState {
+ public int partitionCount;
+
+ public String distribution;
+
+ private PartitionValueCacheKey key;
+ private HivePartitionValues partitionValues;
+ private HivePartitionValues replacementPartitionValues;
+ private HivePartitionValues currentPartitionValues;
+ private HivePartitionValues nextPartitionValues;
+ private MetaCacheEntry cacheEntry;
+ private MetaCacheEntry countCacheEntry;
+ private ExecutorService cacheExecutor;
+
+ public void setup() throws Exception {
+ List types = benchmarkTypes();
+ key = benchmarkKey(types);
+ partitionValues = createPartitionValues(partitionCount, distribution, types);
+ partitionValues.prepareForCachePublication(key);
+ requireComplete(partitionValues);
+
+ // A distinct value root makes Caffeine perform a real replacement while sharing
+ // immutable payload objects to keep the fixture's resident heap bounded.
+ replacementPartitionValues = new HivePartitionValues(
+ partitionValues.getIdToPartitionItem(),
+ partitionValues.getPartitionNameToIdMap(),
+ partitionValues.getPartitionValuesMap());
+ replacementPartitionValues.prepareForCachePublication(key);
+ requireComplete(replacementPartitionValues);
+
+ cacheExecutor = MoreExecutors.newDirectExecutorService();
+ ExternalMetaCacheBudgetManager budgetManager =
+ new ExternalMetaCacheBudgetManager(OptionalLong.of(MAX_WEIGHT_BYTES));
+ EntryBudget entryBudget = budgetManager.createEntryBudget(
+ 1L, "hive", "partition_values_benchmark", OptionalLong.empty(), OptionalLong.empty());
+ cacheEntry = new MetaCacheEntry<>(
+ "partition_values_benchmark",
+ ignored -> partitionValues,
+ CacheSpec.ofWeight(true, CacheSpec.CACHE_NO_TTL, 1L, MAX_WEIGHT_BYTES),
+ cacheExecutor,
+ false,
+ false,
+ (entryKey, value) -> value.prepareForCachePublication(entryKey),
+ entryBudget);
+ cacheEntry.put(key, partitionValues);
+ countCacheEntry = new MetaCacheEntry<>(
+ "partition_values_count_benchmark",
+ ignored -> partitionValues,
+ CacheSpec.of(true, CacheSpec.CACHE_NO_TTL, 1L),
+ cacheExecutor,
+ false,
+ false);
+ countCacheEntry.put(key, partitionValues);
+ currentPartitionValues = partitionValues;
+ nextPartitionValues = replacementPartitionValues;
+ }
+
+ public void tearDown() {
+ if (cacheEntry != null) {
+ cacheEntry.close();
+ }
+ if (countCacheEntry != null) {
+ countCacheEntry.close();
+ }
+ if (cacheExecutor != null) {
+ cacheExecutor.shutdownNow();
+ }
+ }
+ }
+
+ private static List benchmarkTypes() {
+ return Collections.singletonList(Type.STRING);
+ }
+
+ private static PartitionValueCacheKey benchmarkKey(List types) {
+ return new PartitionValueCacheKey(
+ NameMapping.createForTest(1L, "benchmark_db", "benchmark_table"), types);
+ }
+
+ private static long requireComplete(HivePartitionValues value) {
+ MetaCacheSizeEstimate estimate = value.getSizeEstimate();
+ if (!estimate.isComplete()) {
+ throw new IllegalStateException("benchmark graph is not fully measurable: "
+ + estimate.getIncompleteReason());
+ }
+ return estimate.getBytes();
+ }
+
+ private static HivePartitionValues createPartitionValues(
+ int count, String distribution, List types) throws Exception {
+ HashBiMap nameToId = HashBiMap.create(count);
+ Map idToItem = Maps.newHashMapWithExpectedSize(count);
+ Map> idToValues = Maps.newHashMapWithExpectedSize(count);
+ String tailPayload = "tail_skew".equals(distribution) ? repeat('x', TAIL_PAYLOAD_BYTES) : null;
+ long partitionNameCharacterCount = 0L;
+
+ for (int i = 0; i < count; i++) {
+ long id = i;
+ String value = i == count - 1 && tailPayload != null ? tailPayload : "value-" + i;
+ String name = "p=" + value;
+ partitionNameCharacterCount += name.length();
+ List rawValues = Collections.singletonList(new PartitionValue(value));
+ PartitionKey partitionKey = PartitionKey.createListPartitionKeyWithTypes(rawValues, types, true);
+ List keys = new ArrayList<>(1);
+ keys.add(partitionKey);
+
+ nameToId.put(name, id);
+ idToItem.put(id, new ListPartitionItem(keys));
+ idToValues.put(id, new ArrayList<>(Collections.singletonList(value)));
+ }
+ return new HivePartitionValues(
+ idToItem, nameToId, idToValues, partitionNameCharacterCount, types.size());
+ }
+
+ private static String repeat(char value, int count) {
+ char[] chars = new char[count];
+ Arrays.fill(chars, value);
+ return new String(chars);
+ }
+}
diff --git a/fe/fe-benchmark/src/main/java/org/apache/doris/datasource/iceberg/IcebergCacheSizeBenchmark.java b/fe/fe-benchmark/src/main/java/org/apache/doris/datasource/iceberg/IcebergCacheSizeBenchmark.java
new file mode 100644
index 00000000000000..2cca17a879b625
--- /dev/null
+++ b/fe/fe-benchmark/src/main/java/org/apache/doris/datasource/iceberg/IcebergCacheSizeBenchmark.java
@@ -0,0 +1,411 @@
+// 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.datasource.iceberg;
+
+import org.apache.doris.benchmark.BenchmarkHarness;
+import org.apache.doris.datasource.NameMapping;
+import org.apache.doris.datasource.iceberg.cache.ManifestCacheValue;
+import org.apache.doris.datasource.metacache.MetaCacheSizeEstimate;
+import org.apache.doris.datasource.metacache.MetaCacheWeightUtils;
+
+import com.google.common.collect.ImmutableList;
+import org.apache.iceberg.BaseTable;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.DataFiles;
+import org.apache.iceberg.ManifestContent;
+import org.apache.iceberg.Metrics;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.Snapshot;
+import org.apache.iceberg.SnapshotParser;
+import org.apache.iceberg.SnapshotRef;
+import org.apache.iceberg.StaticTableOperations;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.TableMetadata;
+import org.apache.iceberg.TableMetadataParser;
+import org.apache.iceberg.inmemory.InMemoryFileIO;
+import org.apache.iceberg.types.Types;
+
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Optional;
+import java.util.concurrent.TimeUnit;
+
+/** Measures Iceberg table, long-history table, snapshot and manifest publication. */
+public class IcebergCacheSizeBenchmark {
+ public int tableValueConstruction(TablePublicationState state) {
+ return new IcebergTableCacheValue(state.table).getIcebergTable().schema().schemaId();
+ }
+
+ public long tablePublication(TablePublicationState state) {
+ IcebergTableCacheValue value = new IcebergTableCacheValue(state.table);
+ value.prepareForCachePublication(state.mapping);
+ return requireComplete(value.getSizeEstimate());
+ }
+
+ public long tablePayloadCounter(TablePublicationState state) {
+ return IcebergCacheSizeEstimator.retainedTablePayloadBytes(state.table);
+ }
+
+ public long longHistoryTablePublication(LongHistoryTablePublicationState state) {
+ IcebergTableCacheValue value = new IcebergTableCacheValue(state.table);
+ value.prepareForCachePublication(state.mapping);
+ return requireComplete(value.getSizeEstimate());
+ }
+
+ public long snapshotPublication(SnapshotPublicationState state) {
+ IcebergSnapshotCacheValue value = new IcebergSnapshotCacheValue(
+ state.partitionInfo, new IcebergSnapshot(7L, 0L), Optional.empty(), state.table);
+ value.prepareForCachePublication(state.snapshotKey);
+ return requireComplete(value.getSizeEstimate());
+ }
+
+ public int snapshotValueConstruction(SnapshotPublicationState state) {
+ return new IcebergSnapshotCacheValue(
+ state.partitionInfo, new IcebergSnapshot(7L, 0L), Optional.empty(), state.table)
+ .getPartitionInfo().getNameToIcebergPartition().size();
+ }
+
+ public long preparedSnapshotCacheHit(SnapshotPublicationState state) {
+ return state.preparedSnapshotValue.getIcebergTable().get()
+ .currentSnapshot().snapshotId();
+ }
+
+ public long manifestPublication(ManifestState state) {
+ return requireComplete(IcebergCacheSizeEstimator.estimateManifestEntry(state.key, state.value));
+ }
+
+ public int manifestValueConstruction(ManifestState state) {
+ return ManifestCacheValue.forDataFiles(state.files).getDataFiles().size();
+ }
+
+ public int denseManifestReaderBaseline(DenseManifestState state) {
+ List collected = new ArrayList<>();
+ for (DataFile file : state.files) {
+ collected.add(file.copy());
+ }
+ return ImmutableList.copyOf(collected).size();
+ }
+
+ public int denseManifestValueConstruction(DenseManifestState state) {
+ ManifestCacheValue.Builder builder = ManifestCacheValue.dataFilesBuilder();
+ for (DataFile file : state.files) {
+ builder.addDataFile(file.copy());
+ }
+ return builder.build().getDataFiles().size();
+ }
+
+ public long preparedWeightProvider(PreparedState state) {
+ return state.preparedTableValue.getSizeEstimate().getBytes();
+ }
+
+ public long tableFormula(PreparedState state) {
+ return requireComplete(IcebergCacheSizeEstimator.estimateTableEntry(
+ state.mapping, state.preparedTableValue));
+ }
+
+ public int preparedTableCacheHit(PreparedState state) {
+ return state.preparedTableValue.getIcebergTable().schema().schemaId();
+ }
+
+ public static void main(String[] args) throws Exception {
+ IcebergCacheSizeBenchmark benchmark = new IcebergCacheSizeBenchmark();
+ for (int fieldCount : new int[] {10, 100}) {
+ String suffix = "[fields=" + fieldCount + "]";
+ TablePublicationState tableState = new TablePublicationState();
+ tableState.fieldCount = fieldCount;
+ tableState.setup();
+ BenchmarkHarness.measure("iceberg.tableValueConstruction" + suffix,
+ TimeUnit.NANOSECONDS, () -> benchmark.tableValueConstruction(tableState));
+ BenchmarkHarness.measure("iceberg.tablePayloadCounter" + suffix,
+ TimeUnit.MICROSECONDS, () -> benchmark.tablePayloadCounter(tableState));
+ BenchmarkHarness.measure("iceberg.tablePublication" + suffix,
+ TimeUnit.MICROSECONDS, () -> benchmark.tablePublication(tableState));
+
+ PreparedState prepared = new PreparedState();
+ prepared.fieldCount = fieldCount;
+ prepared.setup();
+ BenchmarkHarness.measure("iceberg.preparedWeightProvider" + suffix,
+ TimeUnit.NANOSECONDS, () -> benchmark.preparedWeightProvider(prepared));
+ BenchmarkHarness.measure("iceberg.tableFormula" + suffix,
+ TimeUnit.NANOSECONDS, () -> benchmark.tableFormula(prepared));
+ BenchmarkHarness.measure("iceberg.preparedTableCacheHit" + suffix,
+ TimeUnit.MICROSECONDS, () -> benchmark.preparedTableCacheHit(prepared));
+ }
+ for (int snapshotCount : new int[] {1000, 10000}) {
+ String suffix = "[snapshots=" + snapshotCount + "]";
+ LongHistoryTablePublicationState state = new LongHistoryTablePublicationState();
+ state.snapshotCount = snapshotCount;
+ state.setup();
+ BenchmarkHarness.measure("iceberg.longHistoryTablePublication" + suffix,
+ TimeUnit.MICROSECONDS, () -> benchmark.longHistoryTablePublication(state));
+ }
+ for (int fieldCount : new int[] {10, 100}) {
+ for (int partitionCount : new int[] {1000, 10000}) {
+ String suffix = "[fields=" + fieldCount + ",partitions=" + partitionCount + "]";
+ SnapshotPublicationState state = new SnapshotPublicationState();
+ state.fieldCount = fieldCount;
+ state.partitionCount = partitionCount;
+ state.setup();
+ BenchmarkHarness.measure("iceberg.snapshotValueConstruction" + suffix,
+ TimeUnit.NANOSECONDS, () -> benchmark.snapshotValueConstruction(state));
+ BenchmarkHarness.measure("iceberg.snapshotPublication" + suffix,
+ TimeUnit.MICROSECONDS, () -> benchmark.snapshotPublication(state));
+ BenchmarkHarness.measure("iceberg.preparedSnapshotCacheHit" + suffix,
+ TimeUnit.MICROSECONDS, () -> benchmark.preparedSnapshotCacheHit(state));
+ }
+ }
+ for (int fileCount : new int[] {100, 10000}) {
+ ManifestState state = new ManifestState();
+ state.fileCount = fileCount;
+ state.setup();
+ BenchmarkHarness.measure("iceberg.manifestPublication[files=" + fileCount + "]",
+ TimeUnit.MICROSECONDS, () -> benchmark.manifestPublication(state));
+ BenchmarkHarness.measure("iceberg.manifestValueConstruction[files=" + fileCount + "]",
+ TimeUnit.MICROSECONDS, () -> benchmark.manifestValueConstruction(state));
+ }
+ for (int metricColumns : new int[] {100, 1000}) {
+ for (int fileCount : new int[] {100, 10000}) {
+ DenseManifestState state = new DenseManifestState();
+ state.metricColumns = metricColumns;
+ state.fileCount = fileCount;
+ state.setup();
+ String suffix = "[files=" + fileCount + ",metricColumns=" + metricColumns + "]";
+ BenchmarkHarness.measure("iceberg.denseManifestReaderBaseline" + suffix,
+ TimeUnit.MICROSECONDS, () -> benchmark.denseManifestReaderBaseline(state));
+ BenchmarkHarness.measure("iceberg.denseManifestValueConstruction" + suffix,
+ TimeUnit.MICROSECONDS, () -> benchmark.denseManifestValueConstruction(state));
+ }
+ }
+ }
+
+ public static class TablePublicationState {
+ public int fieldCount;
+
+ private NameMapping mapping;
+ private Table table;
+
+ public void setup() {
+ mapping = NameMapping.createForTest(1L, "benchmark_db", "benchmark_table");
+ table = newTable(fieldCount);
+ }
+ }
+
+ public static class SnapshotPublicationState {
+ public int fieldCount;
+
+ public int partitionCount;
+
+ private Table table;
+ private IcebergSnapshotEntryKey snapshotKey;
+ private IcebergPartitionInfo partitionInfo;
+ private IcebergSnapshotCacheValue preparedSnapshotValue;
+
+ public void setup() {
+ NameMapping mapping = NameMapping.createForTest(1L, "benchmark_db", "benchmark_table");
+ table = newTable(fieldCount);
+ partitionInfo = newPartitionInfo(partitionCount);
+ snapshotKey = IcebergSnapshotEntryKey.tryCreate(mapping, table)
+ .orElseThrow(() -> new IllegalStateException("benchmark table has no generation key"));
+ preparedSnapshotValue = new IcebergSnapshotCacheValue(
+ partitionInfo, new IcebergSnapshot(7L, 0L), Optional.empty(), table);
+ preparedSnapshotValue.prepareForCachePublication(snapshotKey);
+ requireComplete(preparedSnapshotValue.getSizeEstimate());
+ }
+ }
+
+ public static class LongHistoryTablePublicationState {
+ public int snapshotCount;
+
+ private NameMapping mapping;
+ private Table table;
+
+ public void setup() {
+ mapping = NameMapping.createForTest(1L, "benchmark_db", "benchmark_table");
+ table = newLongHistoryTable(snapshotCount);
+ }
+ }
+
+ public static class PreparedState {
+ public int fieldCount;
+
+ private IcebergTableCacheValue preparedTableValue;
+ private NameMapping mapping;
+
+ public void setup() {
+ mapping = NameMapping.createForTest(1L, "benchmark_db", "benchmark_table");
+ Table table = newTable(fieldCount);
+ preparedTableValue = new IcebergTableCacheValue(table);
+ preparedTableValue.prepareForCachePublication(mapping);
+ requireComplete(preparedTableValue.getSizeEstimate());
+ }
+ }
+
+ public static class ManifestState {
+ public int fileCount;
+
+ private IcebergManifestEntryKey key;
+ private ManifestCacheValue value;
+ private List files;
+
+ public void setup() {
+ key = new IcebergManifestEntryKey("/benchmark/manifest.avro", ManifestContent.DATA);
+ files = new ArrayList<>(fileCount);
+ for (int index = 0; index < fileCount; index++) {
+ files.add(DataFiles.builder(PartitionSpec.unpartitioned())
+ .withPath("/benchmark/data/file-" + index + ".parquet")
+ .withFileSizeInBytes(1024L + index)
+ .withRecordCount(10L + index)
+ .build());
+ }
+ value = ManifestCacheValue.forDataFiles(files);
+ }
+ }
+
+ public static class DenseManifestState {
+ public int fileCount;
+
+ public int metricColumns;
+
+ private List files;
+
+ public void setup() {
+ int poolSize = Math.min(fileCount, 256);
+ List filePool = new ArrayList<>(poolSize);
+ for (int fileIndex = 0; fileIndex < poolSize; fileIndex++) {
+ HashMap lowerBounds = new HashMap<>(metricColumns);
+ HashMap upperBounds = new HashMap<>(metricColumns);
+ for (int columnIndex = 0; columnIndex < metricColumns; columnIndex++) {
+ lowerBounds.put(columnIndex, ByteBuffer.allocate(16));
+ upperBounds.put(columnIndex, ByteBuffer.allocate(32));
+ }
+ Metrics metrics = new Metrics(
+ 10L,
+ Collections.emptyMap(),
+ Collections.emptyMap(),
+ Collections.emptyMap(),
+ Collections.emptyMap(),
+ lowerBounds,
+ upperBounds);
+ filePool.add(DataFiles.builder(PartitionSpec.unpartitioned())
+ .withPath("/benchmark/data/dense-" + fileIndex + ".parquet")
+ .withFileSizeInBytes(1024L)
+ .withMetrics(metrics)
+ .build());
+ }
+ files = new ArrayList<>(fileCount);
+ for (int index = 0; index < fileCount; index++) {
+ files.add(filePool.get(index % poolSize));
+ }
+ }
+ }
+
+ private static Table newTable(int fieldCount) {
+ List fields = new ArrayList<>(fieldCount);
+ for (int index = 0; index < fieldCount; index++) {
+ fields.add(Types.NestedField.optional(index + 1, "field_" + index, Types.StringType.get()));
+ }
+ Schema schema = new Schema(fields);
+ TableMetadata metadata = TableMetadata.newTableMetadata(
+ schema, PartitionSpec.unpartitioned(), "file:/benchmark/table", Collections.emptyMap());
+ InMemoryFileIO fileIO = new InMemoryFileIO();
+ StringBuilder snapshotJson = new StringBuilder("{\"snapshot-id\":7,\"timestamp-ms\":1,")
+ .append("\"summary\":{\"operation\":\"append\"},\"manifests\":[");
+ for (int index = 0; index < 10; index++) {
+ if (index > 0) {
+ snapshotJson.append(',');
+ }
+ String manifestPath = "/benchmark/manifest-" + index + ".avro";
+ snapshotJson.append('"').append(manifestPath).append('"');
+ fileIO.addFile(manifestPath, new byte[0]);
+ }
+ Snapshot snapshot = SnapshotParser.fromJson(snapshotJson.append("],\"schema-id\":0}").toString());
+ metadata = TableMetadata.buildFrom(metadata).setBranchSnapshot(snapshot, SnapshotRef.MAIN_BRANCH)
+ .discardChanges()
+ .withMetadataLocation("file:/benchmark/table/metadata/v1.json").build();
+ return new BaseTable(new StaticTableOperations(metadata, fileIO), "benchmark.table");
+ }
+
+ private static Table newLongHistoryTable(int snapshotCount) {
+ long currentSnapshotId = 1000L + snapshotCount - 1L;
+ StringBuilder json = new StringBuilder()
+ .append("{\"format-version\":2,\"table-uuid\":\"benchmark-table\",")
+ .append("\"location\":\"file:/benchmark/table\",\"last-sequence-number\":")
+ .append(snapshotCount).append(",\"last-updated-ms\":").append(snapshotCount)
+ .append(",\"last-column-id\":1,\"current-schema-id\":0,")
+ .append("\"schemas\":[{\"type\":\"struct\",\"schema-id\":0,\"fields\":[")
+ .append("{\"id\":1,\"name\":\"field\",\"required\":false,\"type\":\"string\"}]}],")
+ .append("\"default-spec-id\":0,\"partition-specs\":[{\"spec-id\":0,\"fields\":[]}],")
+ .append("\"last-partition-id\":999,\"default-sort-order-id\":0,")
+ .append("\"sort-orders\":[{\"order-id\":0,\"fields\":[]}],\"properties\":{},")
+ .append("\"current-snapshot-id\":").append(currentSnapshotId)
+ .append(",\"refs\":{\"main\":{\"snapshot-id\":").append(currentSnapshotId)
+ .append(",\"type\":\"branch\"}},\"snapshots\":[");
+ for (int index = 0; index < snapshotCount; index++) {
+ if (index > 0) {
+ json.append(',');
+ }
+ json.append("{\"sequence-number\":").append(index + 1L)
+ .append(",\"snapshot-id\":").append(1000L + index)
+ .append(",\"timestamp-ms\":").append(index + 1L)
+ .append(",\"summary\":{\"operation\":\"append\"},")
+ .append("\"manifest-list\":\"/benchmark/history/list-").append(index)
+ .append(".avro\",\"schema-id\":0}");
+ }
+ json.append("],\"statistics\":[],\"partition-statistics\":[],\"snapshot-log\":[");
+ for (int index = 0; index < snapshotCount; index++) {
+ if (index > 0) {
+ json.append(',');
+ }
+ json.append("{\"timestamp-ms\":").append(index + 1L)
+ .append(",\"snapshot-id\":").append(1000L + index).append('}');
+ }
+ json.append("],\"metadata-log\":[]}");
+ TableMetadata metadata = TableMetadataParser.fromJson(
+ "file:/benchmark/table/metadata/v1.json", json.toString());
+ return new BaseTable(
+ new StaticTableOperations(metadata, new InMemoryFileIO()), "benchmark.table");
+ }
+
+ private static IcebergPartitionInfo newPartitionInfo(int partitionCount) {
+ HashMap partitions = new HashMap<>(partitionCount);
+ long retainedPayloadBytes = 0L;
+ for (int index = 0; index < partitionCount; index++) {
+ String name = "partition_key=value_" + index;
+ IcebergPartition partition = new IcebergPartition(name, 0, 10L + index, 1024L + index,
+ 1L, 1_700_000_000_000L + index, 7L,
+ Collections.singletonList("value_" + index), Collections.singletonList("identity"));
+ partitions.put(name, partition);
+ retainedPayloadBytes = MetaCacheWeightUtils.saturatedAdd(
+ retainedPayloadBytes, partition.getRetainedPayloadBytes());
+ }
+ return new IcebergPartitionInfo(
+ Collections.emptyMap(), partitions, Collections.emptyMap(), retainedPayloadBytes);
+ }
+
+ private static long requireComplete(MetaCacheSizeEstimate estimate) {
+ if (!estimate.isComplete()) {
+ throw new IllegalStateException("incomplete benchmark estimate: " + estimate.getIncompleteReason());
+ }
+ return estimate.getBytes();
+ }
+
+}
diff --git a/fe/fe-benchmark/src/main/java/org/apache/doris/datasource/metacache/MetaCacheSoftValueBenchmark.java b/fe/fe-benchmark/src/main/java/org/apache/doris/datasource/metacache/MetaCacheSoftValueBenchmark.java
new file mode 100644
index 00000000000000..b488f428e8b9a9
--- /dev/null
+++ b/fe/fe-benchmark/src/main/java/org/apache/doris/datasource/metacache/MetaCacheSoftValueBenchmark.java
@@ -0,0 +1,175 @@
+// 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.datasource.metacache;
+
+import org.apache.doris.datasource.metacache.ExternalMetaCacheBudgetManager.EntryBudget;
+
+import com.github.benmanes.caffeine.cache.LoadingCache;
+import com.google.common.util.concurrent.MoreExecutors;
+
+import java.lang.ref.Reference;
+import java.lang.reflect.Field;
+import java.lang.reflect.Method;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.OptionalLong;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.LockSupport;
+
+/** Measures reservation cleanup after Caffeine reports soft values as COLLECTED. */
+public final class MetaCacheSoftValueBenchmark {
+ private static final int SAMPLE_COUNT = 3;
+ private static final int VALUE_COUNT = 10_000;
+ private static final long MAX_WEIGHT_BYTES = 16L * 1024L * 1024L;
+
+ private MetaCacheSoftValueBenchmark() {
+ }
+
+ public static void main(String[] args) throws Exception {
+ long totalNanos = 0L;
+ for (int sample = 0; sample < SAMPLE_COUNT; sample++) {
+ CollectedState state = CollectedState.create(VALUE_COUNT);
+ try {
+ state.enqueueAll();
+ long start = System.nanoTime();
+ state.cleanUp();
+ totalNanos += System.nanoTime() - start;
+ } finally {
+ state.close();
+ }
+ }
+ double averageNanos = (double) totalNanos / SAMPLE_COUNT;
+ System.out.printf(Locale.ROOT,
+ "%-72s %12.3f us/batch (%.3f ns/value, %d samples)%n",
+ "metacache.collectedCleanup[values=" + VALUE_COUNT + "]",
+ averageNanos / TimeUnit.MICROSECONDS.toNanos(1L),
+ averageNanos / VALUE_COUNT,
+ SAMPLE_COUNT);
+ }
+
+ private static final class CollectedState implements AutoCloseable {
+ private final ExecutorService executor;
+ private final ExternalMetaCacheBudgetManager budgetManager;
+ private final MetaCacheEntry entry;
+ private final LoadingCache, ?> loadingCache;
+ private final List> references;
+
+ private CollectedState(ExecutorService executor,
+ ExternalMetaCacheBudgetManager budgetManager,
+ MetaCacheEntry entry,
+ LoadingCache, ?> loadingCache,
+ List> references) {
+ this.executor = executor;
+ this.budgetManager = budgetManager;
+ this.entry = entry;
+ this.loadingCache = loadingCache;
+ this.references = references;
+ }
+
+ private static CollectedState create(int valueCount) throws Exception {
+ ExecutorService executor = MoreExecutors.newDirectExecutorService();
+ ExternalMetaCacheBudgetManager budgetManager =
+ new ExternalMetaCacheBudgetManager(OptionalLong.of(MAX_WEIGHT_BYTES));
+ EntryBudget budget = budgetManager.createEntryBudget(
+ 1L, "benchmark", "soft_cleanup", OptionalLong.empty(), OptionalLong.empty());
+ MetaCacheEntry entry = new MetaCacheEntry<>(
+ "soft_cleanup",
+ key -> new byte[1],
+ CacheSpec.ofWeight(true, CacheSpec.CACHE_NO_TTL, valueCount, MAX_WEIGHT_BYTES),
+ executor,
+ false,
+ false,
+ (key, value) -> MetaCacheSizeEstimate.complete(value.length),
+ budget);
+ for (int index = 0; index < valueCount; index++) {
+ entry.put("key-" + index, new byte[1]);
+ }
+
+ LoadingCache, ?> loadingCache = (LoadingCache, ?>) readField(entry, "loadingData");
+ Object boundedLocalCache = readField(loadingCache, "cache");
+ Map, ?> nodes = (Map, ?>) readField(boundedLocalCache, "data");
+ List> references = new ArrayList<>(nodes.size());
+ for (Object node : nodes.values()) {
+ Method valueReferenceMethod = findMethod(node.getClass(), "getValueReference");
+ references.add((Reference>) valueReferenceMethod.invoke(node));
+ }
+ if (references.size() != valueCount) {
+ entry.close();
+ executor.shutdownNow();
+ throw new IllegalStateException(
+ "benchmark admission retained " + references.size() + " of " + valueCount + " values");
+ }
+ return new CollectedState(executor, budgetManager, entry, loadingCache, references);
+ }
+
+ private void enqueueAll() {
+ for (Reference> reference : references) {
+ reference.clear();
+ reference.enqueue();
+ }
+ }
+
+ private void cleanUp() {
+ long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(30L);
+ while (budgetManager.getGlobalUsedWeight() != 0L
+ && System.nanoTime() < deadline) {
+ loadingCache.cleanUp();
+ LockSupport.parkNanos(TimeUnit.MICROSECONDS.toNanos(100L));
+ }
+ if (budgetManager.getGlobalUsedWeight() != 0L) {
+ throw new IllegalStateException(
+ "COLLECTED cleanup retained " + budgetManager.getGlobalUsedWeight() + " bytes");
+ }
+ }
+
+ @Override
+ public void close() {
+ entry.close();
+ executor.shutdownNow();
+ }
+ }
+
+ private static Object readField(Object target, String name) throws Exception {
+ for (Class> type = target.getClass(); type != null; type = type.getSuperclass()) {
+ try {
+ Field field = type.getDeclaredField(name);
+ field.setAccessible(true);
+ return field.get(target);
+ } catch (NoSuchFieldException ignored) {
+ // Continue through Caffeine's generated cache hierarchy.
+ }
+ }
+ throw new NoSuchFieldException(name);
+ }
+
+ private static Method findMethod(Class> type, String name) throws Exception {
+ for (Class> current = type; current != null; current = current.getSuperclass()) {
+ try {
+ Method method = current.getDeclaredMethod(name);
+ method.setAccessible(true);
+ return method;
+ } catch (NoSuchMethodException ignored) {
+ // Continue through Caffeine's generated node hierarchy.
+ }
+ }
+ throw new NoSuchMethodException(name);
+ }
+}
diff --git a/fe/fe-benchmark/src/main/java/org/apache/doris/datasource/paimon/PaimonCacheSizeBenchmark.java b/fe/fe-benchmark/src/main/java/org/apache/doris/datasource/paimon/PaimonCacheSizeBenchmark.java
new file mode 100644
index 00000000000000..f73eca412f3c44
--- /dev/null
+++ b/fe/fe-benchmark/src/main/java/org/apache/doris/datasource/paimon/PaimonCacheSizeBenchmark.java
@@ -0,0 +1,272 @@
+// 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.datasource.paimon;
+
+import org.apache.doris.benchmark.BenchmarkHarness;
+import org.apache.doris.datasource.NameMapping;
+import org.apache.doris.datasource.metacache.MetaCacheSizeEstimate;
+import org.apache.doris.datasource.metacache.MetaCacheWeightUtils;
+
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.fs.local.LocalFileIO;
+import org.apache.paimon.partition.Partition;
+import org.apache.paimon.schema.TableSchema;
+import org.apache.paimon.table.AppendOnlyFileStoreTable;
+import org.apache.paimon.table.CatalogEnvironment;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.IntType;
+import org.apache.paimon.types.RowType;
+import org.apache.paimon.types.VarCharType;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+
+/** Measures Paimon 1.4.2 nested-schema/non-empty snapshot publication and prepared weight lookup. */
+public class PaimonCacheSizeBenchmark {
+ public long snapshotPublication(PublicationState state) {
+ PaimonSnapshotCacheValue value = new PaimonSnapshotCacheValue(state.partitionInfo, state.snapshot);
+ value.prepareForCachePublication(state.key);
+ return requireComplete(value.getSizeEstimate());
+ }
+
+ public long tablePayloadCounter(PublicationState state) {
+ return PaimonCacheSizeEstimator.retainedTablePayloadBytes(state.snapshot.getTable());
+ }
+
+ public long preparedWeightProvider(PreparedState state) {
+ return state.value.getSizeEstimate().getBytes();
+ }
+
+ public long snapshotFormula(PreparedState state) {
+ return requireComplete(PaimonCacheSizeEstimator.estimateSnapshotEntry(state.key, state.value));
+ }
+
+ public long partitionMapBaseline(PartitionPayloadState state) {
+ return buildPartitionInfo(state, false);
+ }
+
+ public long partitionMapWithRetainedCounter(PartitionPayloadState state) {
+ return buildPartitionInfo(state, true);
+ }
+
+ public static void main(String[] args) throws Exception {
+ PaimonCacheSizeBenchmark benchmark = new PaimonCacheSizeBenchmark();
+ for (int fieldCount : new int[] {10, 100}) {
+ for (int partitionCount : new int[] {1000, 10000}) {
+ String suffix = "[fields=" + fieldCount + ",partitions=" + partitionCount + "]";
+ PublicationState publication = new PublicationState();
+ publication.fieldCount = fieldCount;
+ publication.partitionCount = partitionCount;
+ publication.setup();
+ BenchmarkHarness.measure("paimon.tablePayloadCounter" + suffix,
+ TimeUnit.MICROSECONDS, () -> benchmark.tablePayloadCounter(publication));
+ BenchmarkHarness.measure("paimon.snapshotPublication" + suffix,
+ TimeUnit.MICROSECONDS, () -> benchmark.snapshotPublication(publication));
+
+ PreparedState prepared = new PreparedState();
+ prepared.fieldCount = fieldCount;
+ prepared.partitionCount = partitionCount;
+ prepared.setup();
+ BenchmarkHarness.measure("paimon.preparedWeightProvider" + suffix,
+ TimeUnit.NANOSECONDS, () -> benchmark.preparedWeightProvider(prepared));
+ BenchmarkHarness.measure("paimon.snapshotFormula" + suffix,
+ TimeUnit.NANOSECONDS, () -> benchmark.snapshotFormula(prepared));
+ }
+ }
+ for (int partitionCount : new int[] {1000, 10000}) {
+ for (boolean tailSkew : new boolean[] {false, true}) {
+ PartitionPayloadState state = new PartitionPayloadState();
+ state.partitionCount = partitionCount;
+ state.tailSkew = tailSkew;
+ state.setup();
+ String suffix = "[partitions=" + partitionCount
+ + ",distribution=" + (tailSkew ? "tail-skew" : "uniform") + "]";
+ BenchmarkHarness.measure("paimon.partitionMapBaseline" + suffix,
+ TimeUnit.MICROSECONDS, () -> benchmark.partitionMapBaseline(state));
+ BenchmarkHarness.measure("paimon.partitionMapWithRetainedCounter" + suffix,
+ TimeUnit.MICROSECONDS, () -> benchmark.partitionMapWithRetainedCounter(state));
+ }
+ }
+ }
+
+ public static class PublicationState {
+ public int fieldCount;
+
+ public int partitionCount;
+
+ private PaimonSnapshotEntryKey key;
+ private PaimonPartitionInfo partitionInfo;
+ private PaimonSnapshot snapshot;
+
+ public void setup() throws Exception {
+ Fixture fixture = newFixture(fieldCount, partitionCount);
+ key = fixture.key;
+ partitionInfo = fixture.value.getPartitionInfo();
+ snapshot = fixture.value.getSnapshot();
+ }
+ }
+
+ public static class PreparedState {
+ public int fieldCount;
+
+ public int partitionCount;
+
+ private PaimonSnapshotEntryKey key;
+ private PaimonSnapshotCacheValue value;
+
+ public void setup() throws Exception {
+ Fixture fixture = newFixture(fieldCount, partitionCount);
+ key = fixture.key;
+ value = fixture.value;
+ value.prepareForCachePublication(fixture.key);
+ requireComplete(value.getSizeEstimate());
+ }
+ }
+
+ public static class PartitionPayloadState {
+ public int partitionCount;
+
+ public boolean tailSkew;
+
+ private List partitions;
+
+ public void setup() {
+ partitions = new ArrayList<>(partitionCount);
+ String longTail = String.join("", Collections.nCopies(64 * 1024, "x"));
+ for (int index = 0; index < partitionCount; index++) {
+ String value = tailSkew && index % 997 == 0 ? longTail : String.valueOf(index);
+ LinkedHashMap typedSpec = new LinkedHashMap<>();
+ for (int field = 0; field < 4; field++) {
+ typedSpec.put("partition_key_" + field, value + '_' + field);
+ }
+ String displayName = "partition_key_0=" + value;
+ partitions.add(new PartitionPayload(
+ displayName, new ArrayList<>(typedSpec.values()), index));
+ }
+ }
+ }
+
+ private static Fixture newFixture(int fieldCount, int partitionCount) throws Exception {
+ List fields = new ArrayList<>(fieldCount + 2);
+ fields.add(new DataField(0, "partition_key", new IntType()));
+ for (int index = 0; index < fieldCount; index++) {
+ fields.add(new DataField(index + 1, "field_" + index, new VarCharType()));
+ }
+ List nestedFields = new ArrayList<>();
+ for (int index = 0; index < 8; index++) {
+ nestedFields.add(DataTypes.FIELD(fieldCount + index + 1,
+ "nested_field_" + index, DataTypes.STRING()));
+ }
+ fields.add(new DataField(fieldCount + 9, "nested_payload", new RowType(nestedFields)));
+ TableSchema schema = new TableSchema(
+ 0L, fields, fieldCount + 9, Collections.singletonList("partition_key"),
+ Collections.emptyList(), Collections.emptyMap(), null);
+ FileStoreTable table = new AppendOnlyFileStoreTable(
+ LocalFileIO.create(), new Path("file:/tmp/doris-paimon-cache-size-benchmark"),
+ schema, CatalogEnvironment.empty());
+ NameMapping mapping = NameMapping.createForTest(1L, "benchmark_db", "benchmark_table");
+ PaimonSnapshotEntryKey key = new PaimonSnapshotEntryKey(mapping, 7L, schema.id(), 1L);
+ HashMap partitions = new HashMap<>(partitionCount);
+ long retainedPartitionPayloadBytes = 0L;
+ for (int index = 0; index < partitionCount; index++) {
+ String name = "partition_key=" + index;
+ String value = String.valueOf(index);
+ partitions.put(name, new Partition(Collections.singletonMap("partition_key", value),
+ 10L + index, 1024L + index, 1L, 1_700_000_000_000L + index, 1, true,
+ 1_700_000_000_000L, "benchmark", 1_700_000_000_000L + index, "benchmark",
+ Collections.singletonMap("source", "benchmark")));
+ retainedPartitionPayloadBytes = MetaCacheWeightUtils.saturatedAdd(
+ retainedPartitionPayloadBytes, MetaCacheWeightUtils.estimatedStringBytes(name));
+ retainedPartitionPayloadBytes = MetaCacheWeightUtils.saturatedAdd(
+ retainedPartitionPayloadBytes, MetaCacheWeightUtils.estimatedStringBytes("partition_key"));
+ retainedPartitionPayloadBytes = MetaCacheWeightUtils.saturatedAdd(
+ retainedPartitionPayloadBytes, MetaCacheWeightUtils.estimatedStringBytes(value));
+ }
+ PaimonSnapshotCacheValue value = new PaimonSnapshotCacheValue(
+ new PaimonPartitionInfo(Collections.emptyMap(), partitions, retainedPartitionPayloadBytes),
+ new PaimonSnapshot(7L, schema.id(), table));
+ return new Fixture(key, value);
+ }
+
+ private static long requireComplete(MetaCacheSizeEstimate estimate) {
+ if (!estimate.isComplete()) {
+ throw new IllegalStateException("incomplete benchmark estimate: " + estimate.getIncompleteReason());
+ }
+ return estimate.getBytes();
+ }
+
+ private long buildPartitionInfo(PartitionPayloadState state, boolean countPayload) {
+ HashMap partitions = new HashMap<>(state.partitionCount);
+ long retainedPayloadBytes = 0L;
+ for (PartitionPayload payload : state.partitions) {
+ LinkedHashMap typedSpec = new LinkedHashMap<>();
+ for (int field = 0; field < payload.values.size(); field++) {
+ String fieldName = "partition_key_" + field;
+ String fieldValue = payload.values.get(field);
+ typedSpec.put(fieldName, fieldValue);
+ if (countPayload) {
+ retainedPayloadBytes = PaimonPartitionInfo.addRetainedStringPayload(
+ retainedPayloadBytes, fieldName);
+ retainedPayloadBytes = PaimonPartitionInfo.addRetainedStringPayload(
+ retainedPayloadBytes, fieldValue);
+ }
+ }
+ if (countPayload) {
+ retainedPayloadBytes = PaimonPartitionInfo.addRetainedStringPayload(
+ retainedPayloadBytes, payload.displayName);
+ }
+ int index = payload.index;
+ partitions.put(payload.displayName, new Partition(
+ typedSpec, 10L + index, 1024L + index, 1L,
+ 1_700_000_000_000L + index, 1, false));
+ }
+ PaimonPartitionInfo partitionInfo = new PaimonPartitionInfo(
+ Collections.emptyMap(), partitions, retainedPayloadBytes);
+ return MetaCacheWeightUtils.saturatedAdd(
+ partitionInfo.getNameToPartition().size(), partitionInfo.getRetainedPayloadBytes());
+ }
+
+ private static class PartitionPayload {
+ private final String displayName;
+ private final List values;
+ private final int index;
+
+ private PartitionPayload(
+ String displayName, List values, int index) {
+ this.displayName = displayName;
+ this.values = values;
+ this.index = index;
+ }
+ }
+
+ private static class Fixture {
+ private final PaimonSnapshotEntryKey key;
+ private final PaimonSnapshotCacheValue value;
+
+ private Fixture(PaimonSnapshotEntryKey key, PaimonSnapshotCacheValue value) {
+ this.key = key;
+ this.value = value;
+ }
+ }
+}
diff --git a/fe/fe-common/src/main/java/org/apache/doris/common/Config.java b/fe/fe-common/src/main/java/org/apache/doris/common/Config.java
index 4e535950aae0e3..c3e95e2d4ff24a 100644
--- a/fe/fe-common/src/main/java/org/apache/doris/common/Config.java
+++ b/fe/fe-common/src/main/java/org/apache/doris/common/Config.java
@@ -2462,6 +2462,11 @@ public class Config extends ConfigBase {
})
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."})
diff --git a/fe/fe-core/pom.xml b/fe/fe-core/pom.xml
index a880c5ffad4db4..93ceeb49b58144 100644
--- a/fe/fe-core/pom.xml
+++ b/fe/fe-core/pom.xml
@@ -851,6 +851,11 @@ under the License.
mockito-inline
test
+
+ org.openjdk.jol
+ jol-core
+ test
+
diff --git a/fe/fe-core/src/main/java/org/apache/doris/common/CacheFactory.java b/fe/fe-core/src/main/java/org/apache/doris/common/CacheFactory.java
index 674bf0aa39cd5b..7ad55174e303e0 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/common/CacheFactory.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/common/CacheFactory.java
@@ -24,6 +24,7 @@
import com.github.benmanes.caffeine.cache.LoadingCache;
import com.github.benmanes.caffeine.cache.RemovalListener;
import com.github.benmanes.caffeine.cache.Ticker;
+import com.github.benmanes.caffeine.cache.Weigher;
import org.jetbrains.annotations.NotNull;
import java.time.Duration;
@@ -49,7 +50,10 @@ public class CacheFactory {
private OptionalLong expireAfterAccessSec;
private OptionalLong refreshAfterWriteSec;
private long maxSize;
+ private OptionalLong maxWeight;
+ private Weigher