properties, String key, OptionalLong defaultValue) {
+ if (key == null) {
+ return defaultValue;
+ }
+ String value = properties.get(key);
+ if (value == null) {
+ return defaultValue;
+ }
+ try {
+ return OptionalLong.of(parseDataSizeAllowZero(value));
+ } catch (IllegalArgumentException e) {
+ throw invalidDataSizeProperty(key, value);
+ }
+ }
+
+ private static long parseDataSizeAllowZero(String value) {
+ Matcher matcher = DATA_SIZE_PATTERN.matcher(value);
+ if (!matcher.matches()) {
+ throw new IllegalArgumentException("invalid data size");
+ }
+
+ long number;
+ try {
+ number = Long.parseLong(matcher.group(1));
+ } catch (NumberFormatException e) {
+ throw new IllegalArgumentException("invalid data size", e);
+ }
+
+ long multiplier;
+ switch (matcher.group(2).toUpperCase(Locale.ROOT)) {
+ case "":
+ case "B":
+ multiplier = 1L;
+ break;
+ case "K":
+ case "KB":
+ multiplier = 1L << 10;
+ break;
+ case "M":
+ case "MB":
+ multiplier = 1L << 20;
+ break;
+ case "G":
+ case "GB":
+ multiplier = 1L << 30;
+ break;
+ case "T":
+ case "TB":
+ multiplier = 1L << 40;
+ break;
+ case "P":
+ case "PB":
+ multiplier = 1L << 50;
+ break;
+ default:
+ throw new IllegalArgumentException("invalid data size unit");
+ }
+ try {
+ return Math.multiplyExact(number, multiplier);
+ } catch (ArithmeticException e) {
+ throw new IllegalArgumentException("data size is too large", e);
+ }
+ }
+
+ private static IllegalArgumentException invalidDataSizeProperty(String key, String value) {
+ return new IllegalArgumentException("The parameter " + key + " is wrong, value is " + value);
+ }
+
public boolean isEnable() {
return enable;
}
@@ -246,6 +349,20 @@ public long getCapacity() {
return capacity;
}
+ public OptionalLong getMaxWeight() {
+ return maxWeight;
+ }
+
+ public boolean isWeightBounded() {
+ return maxWeight.isPresent();
+ }
+
+ public boolean isCacheEnabled() {
+ return enable
+ && ttlSecond != CACHE_TTL_DISABLE_CACHE
+ && maxWeight.orElse(capacity) != 0L;
+ }
+
public static final class PropertySpec {
private final String enableKey;
private final boolean defaultEnable;
@@ -253,15 +370,20 @@ public static final class PropertySpec {
private final long defaultTtlSecond;
private final String capacityKey;
private final long defaultCapacity;
+ private final String maxWeightKey;
+ private final OptionalLong defaultMaxWeight;
private PropertySpec(String enableKey, boolean defaultEnable, String ttlKey,
- long defaultTtlSecond, String capacityKey, long defaultCapacity) {
+ long defaultTtlSecond, String capacityKey, long defaultCapacity,
+ String maxWeightKey, OptionalLong defaultMaxWeight) {
this.enableKey = enableKey;
this.defaultEnable = defaultEnable;
this.ttlKey = ttlKey;
this.defaultTtlSecond = defaultTtlSecond;
this.capacityKey = capacityKey;
this.defaultCapacity = defaultCapacity;
+ this.maxWeightKey = maxWeightKey;
+ this.defaultMaxWeight = defaultMaxWeight;
}
public String getEnableKey() {
@@ -288,6 +410,14 @@ public long getDefaultCapacity() {
return defaultCapacity;
}
+ public String getMaxWeightKey() {
+ return maxWeightKey;
+ }
+
+ public OptionalLong getDefaultMaxWeight() {
+ return defaultMaxWeight;
+ }
+
public static final class Builder {
private String enableKey;
private boolean defaultEnable;
@@ -295,6 +425,8 @@ public static final class Builder {
private long defaultTtlSecond;
private String capacityKey;
private long defaultCapacity;
+ private String maxWeightKey;
+ private OptionalLong defaultMaxWeight = OptionalLong.empty();
public Builder enable(String key, boolean defaultValue) {
this.enableKey = key;
@@ -314,6 +446,12 @@ public Builder capacity(String key, long defaultValue) {
return this;
}
+ public Builder maxWeight(String key, OptionalLong defaultValue) {
+ this.maxWeightKey = Objects.requireNonNull(key, "maxWeightKey is required");
+ this.defaultMaxWeight = Objects.requireNonNull(defaultValue, "defaultMaxWeight is required");
+ return this;
+ }
+
public PropertySpec build() {
return new PropertySpec(
Objects.requireNonNull(enableKey, "enableKey is required"),
@@ -321,7 +459,9 @@ public PropertySpec build() {
Objects.requireNonNull(ttlKey, "ttlKey is required"),
defaultTtlSecond,
Objects.requireNonNull(capacityKey, "capacityKey is required"),
- defaultCapacity);
+ defaultCapacity,
+ maxWeightKey,
+ defaultMaxWeight);
}
}
}
diff --git a/fe/fe-connector/fe-connector-cache/src/main/java/org/apache/doris/connector/cache/ConnectorMetadataCache.java b/fe/fe-connector/fe-connector-cache/src/main/java/org/apache/doris/connector/cache/ConnectorMetadataCache.java
index fde68573e73c50..1bbbb6be93d38f 100644
--- a/fe/fe-connector/fe-connector-cache/src/main/java/org/apache/doris/connector/cache/ConnectorMetadataCache.java
+++ b/fe/fe-connector/fe-connector-cache/src/main/java/org/apache/doris/connector/cache/ConnectorMetadataCache.java
@@ -33,10 +33,12 @@
* the hive/iceberg/paimon derived partition-view caches (entry {@code "partition_view"}); a connector may hold
* several instances under distinct entry names.
*
- * Config : {@code meta.cache...(enable|ttl-second|capacity)}, default ON / 86400s / 1000
- * entries (matching {@code IcebergPartitionCache}'s {@code DEFAULT_TABLE_CACHE_CAPACITY}). {@code enable=false} /
- * {@code ttl-second=0} / {@code capacity=0} each disable the cache (see {@link CacheSpec#isCacheEnabled}): {@link
- * #get} then calls the loader on every call, matching {@code IcebergPartitionCache}'s disabled-cache bypass.
+ * Config : {@code meta.cache...(enable|ttl-second|capacity|max-weight)}, default ON /
+ * 86400s / 1000 entries (matching {@code IcebergPartitionCache}'s {@code DEFAULT_TABLE_CACHE_CAPACITY}).
+ * {@code enable=false} / {@code ttl-second=0} / an effective bound of zero each disable the cache (see
+ * {@link CacheSpec#isCacheEnabled}): {@link #get} then calls the loader on every call, matching
+ * {@code IcebergPartitionCache}'s disabled-cache bypass. A caller that supports {@code max-weight} must use the
+ * estimator constructor; count-bounded callers can keep using the original constructor.
*
* Concurrency : mirrors {@code IcebergPartitionCache} / {@code MaxComputePartitionCache} exactly — the
* entry is contextual-only (no built-in loader; the caller supplies one per {@link #get} call) with manual miss
@@ -51,25 +53,38 @@ public final class ConnectorMetadataCache {
static final long DEFAULT_CAPACITY = 1000L;
private final MetaCacheEntry entry;
+ private final boolean weightBounded;
/**
* @param engine engine token for the {@code meta.cache...*} property namespace, e.g.
* {@code "iceberg"}/{@code "paimon"}/{@code "hive"}/{@code "max_compute"}.
* @param entryName the entry name within that namespace (e.g. {@code "partition_view"}); a connector may hold
* several {@code ConnectorMetadataCache}s under distinct entry names.
- * @param props the catalog properties; drives the {@link CacheSpec} (enable/ttl-second/capacity). May be
- * {@code null}, treated as empty (defaults apply).
+ * @param props the catalog properties; drives the {@link CacheSpec}
+ * (enable/ttl-second/capacity/max-weight). May be {@code null}, treated as empty.
*/
public ConnectorMetadataCache(String engine, String entryName, Map props) {
+ this(engine, entryName, props, null);
+ }
+
+ /**
+ * Builds a generic connector metadata cache with an optional type-specific byte estimator.
+ *
+ * @param sizeEstimator required when {@code meta.cache...max-weight} is configured; ignored by
+ * the count-bounded branch
+ */
+ public ConnectorMetadataCache(String engine, String entryName, Map props,
+ MetaCacheSizeEstimator sizeEstimator) {
Objects.requireNonNull(engine, "engine can not be null");
Objects.requireNonNull(entryName, "entryName can not be null");
Map properties = props == null ? Collections.emptyMap() : props;
CacheSpec spec = CacheSpec.fromProperties(properties, engine, entryName,
CacheSpec.of(true, DEFAULT_TTL_SECOND, DEFAULT_CAPACITY));
+ this.weightBounded = spec.isWeightBounded();
// contextual-only (loader == null, supplied per-call by get()) + manual-miss-load, no auto-refresh --
// identical shape to IcebergPartitionCache / MaxComputePartitionCache's entry construction.
this.entry = new MetaCacheEntry<>(engine + "." + entryName, null, spec,
- ForkJoinPool.commonPool(), false, true, 0L, true);
+ ForkJoinPool.commonPool(), false, true, 0L, true, sizeEstimator);
}
/** Caching is on only when the resolved {@link CacheSpec} is effectively enabled (see {@link #entry}'s spec). */
@@ -77,6 +92,11 @@ public boolean isEnabled() {
return entry.stats().isEffectiveEnabled();
}
+ /** Whether this cache is configured with {@code max-weight} instead of the legacy entry-count capacity. */
+ public boolean isWeightBounded() {
+ return weightBounded;
+ }
+
/**
* Returns the cached value for {@code key} if present, else runs {@code loader}, caches and returns it.
* Disabled cache -> {@code loader} runs on every call. The loader runs OUTSIDE Caffeine's compute lock
diff --git a/fe/fe-connector/fe-connector-cache/src/main/java/org/apache/doris/connector/cache/JvmSizeUtils.java b/fe/fe-connector/fe-connector-cache/src/main/java/org/apache/doris/connector/cache/JvmSizeUtils.java
new file mode 100644
index 00000000000000..67e5fc7f1e81e3
--- /dev/null
+++ b/fe/fe-connector/fe-connector-cache/src/main/java/org/apache/doris/connector/cache/JvmSizeUtils.java
@@ -0,0 +1,189 @@
+// 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.connector.cache;
+
+import com.sun.management.HotSpotDiagnosticMXBean;
+
+import java.lang.management.ManagementFactory;
+import java.lang.reflect.Field;
+import java.lang.reflect.Modifier;
+
+/**
+ * Low-cost JVM heap layout formulas for type-specific connector cache estimators.
+ *
+ * This class reflects class declarations once to calculate shallow sizes. It never reads fields from a runtime
+ * object and never walks an object graph.
+ */
+public final class JvmSizeUtils {
+ private static final VmLayout VM_LAYOUT = VmLayout.detect();
+ private static final boolean COMPACT_STRINGS = vmBoolean("CompactStrings");
+
+ private static final ClassValue SHALLOW_SIZES = new ClassValue<>() {
+ @Override
+ protected Long computeValue(Class> type) {
+ if (type.isArray()) {
+ throw new IllegalArgumentException("Array size depends on its length: " + type);
+ }
+ long size = VM_LAYOUT.objectHeaderBytes();
+ for (Class> current = type; current != null; current = current.getSuperclass()) {
+ for (Field field : current.getDeclaredFields()) {
+ if (!Modifier.isStatic(field.getModifiers())) {
+ size = saturatedAdd(size, fieldSize(field.getType()));
+ }
+ }
+ }
+ return align(size);
+ }
+ };
+ private static final long STRING_SHALLOW_BYTES = instanceSize(String.class);
+ private static final long ARRAY_LIST_SHALLOW_BYTES = instanceSize(java.util.ArrayList.class);
+
+ private JvmSizeUtils() {
+ }
+
+ public static long instanceSize(Class> type) {
+ return SHALLOW_SIZES.get(type);
+ }
+
+ public static long objectArraySize(int length) {
+ return arraySize(length, VM_LAYOUT.referenceBytes());
+ }
+
+ public static long byteArraySize(int length) {
+ return arraySize(length, Byte.BYTES);
+ }
+
+ public static long intArraySize(int length) {
+ return arraySize(length, Integer.BYTES);
+ }
+
+ public static long longArraySize(int length) {
+ return arraySize(length, Long.BYTES);
+ }
+
+ /** Estimate an array whose component is a primitive type. */
+ public static long primitiveArraySize(Class> componentType, int length) {
+ if (!componentType.isPrimitive() || componentType == void.class) {
+ throw new IllegalArgumentException("Not an array component primitive: " + componentType);
+ }
+ return arraySize(length, Math.toIntExact(fieldSize(componentType)));
+ }
+
+ public static long arrayListSize(int backingArrayCapacity) {
+ return saturatedAdd(ARRAY_LIST_SHALLOW_BYTES, objectArraySize(backingArrayCapacity));
+ }
+
+ /** Estimate the heap retained by a Java 17 String and its compact-string byte array. */
+ public static long stringSize(String value) {
+ if (value == null) {
+ return 0L;
+ }
+ int bytesPerCharacter = COMPACT_STRINGS && isLatin1(value) ? Byte.BYTES : Character.BYTES;
+ long valueBytes = saturatedMultiply(value.length(), bytesPerCharacter);
+ int arrayLength = valueBytes >= Integer.MAX_VALUE ? Integer.MAX_VALUE : (int) valueBytes;
+ return saturatedAdd(STRING_SHALLOW_BYTES, byteArraySize(arrayLength));
+ }
+
+ public static long saturatedAdd(long left, long right) {
+ if (right > 0L && left > Long.MAX_VALUE - right) {
+ return Long.MAX_VALUE;
+ }
+ return left + right;
+ }
+
+ public static long saturatedMultiply(long left, long right) {
+ if (left == 0L || right == 0L) {
+ return 0L;
+ }
+ if (left > Long.MAX_VALUE / right) {
+ return Long.MAX_VALUE;
+ }
+ return left * right;
+ }
+
+ private static long arraySize(int length, int elementBytes) {
+ long elements = saturatedMultiply(length, elementBytes);
+ return align(saturatedAdd(VM_LAYOUT.arrayHeaderBytes(), elements));
+ }
+
+ private static long fieldSize(Class> type) {
+ if (!type.isPrimitive()) {
+ return VM_LAYOUT.referenceBytes();
+ }
+ if (type == long.class || type == double.class) {
+ return Long.BYTES;
+ }
+ if (type == int.class || type == float.class) {
+ return Integer.BYTES;
+ }
+ if (type == short.class || type == char.class) {
+ return Short.BYTES;
+ }
+ return Byte.BYTES;
+ }
+
+ private static boolean isLatin1(String value) {
+ for (int i = 0; i < value.length(); i++) {
+ if (value.charAt(i) > 0xff) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ private static long align(long value) {
+ long remainder = value % VM_LAYOUT.objectAlignmentBytes();
+ return remainder == 0L
+ ? value
+ : saturatedAdd(value, VM_LAYOUT.objectAlignmentBytes() - remainder);
+ }
+
+ private static boolean vmBoolean(String option) {
+ return Boolean.parseBoolean(hotSpotDiagnostic().getVMOption(option).getValue());
+ }
+
+ private static int vmInt(String option) {
+ return Integer.parseInt(hotSpotDiagnostic().getVMOption(option).getValue());
+ }
+
+ private static HotSpotDiagnosticMXBean hotSpotDiagnostic() {
+ return ManagementFactory.getPlatformMXBean(HotSpotDiagnosticMXBean.class);
+ }
+
+ private record VmLayout(
+ int referenceBytes,
+ int objectHeaderBytes,
+ int arrayHeaderBytes,
+ int objectAlignmentBytes) {
+
+ private static VmLayout detect() {
+ int referenceBytes = vmBoolean("UseCompressedOops") ? Integer.BYTES : Long.BYTES;
+ int classPointerBytes = vmBoolean("UseCompressedClassPointers") ? Integer.BYTES : Long.BYTES;
+ int alignment = vmInt("ObjectAlignmentInBytes");
+ int objectHeaderBytes = Long.BYTES + classPointerBytes;
+ int arrayHeaderBytes = Math.toIntExact(
+ alignWithoutLayout(objectHeaderBytes + Integer.BYTES, alignment));
+ return new VmLayout(referenceBytes, objectHeaderBytes, arrayHeaderBytes, alignment);
+ }
+
+ private static long alignWithoutLayout(long value, int alignment) {
+ long remainder = value % alignment;
+ return remainder == 0L ? value : value + alignment - remainder;
+ }
+ }
+}
diff --git a/fe/fe-connector/fe-connector-cache/src/main/java/org/apache/doris/connector/cache/MetaCacheEntry.java b/fe/fe-connector/fe-connector-cache/src/main/java/org/apache/doris/connector/cache/MetaCacheEntry.java
index 425697ef9b1098..d0dc2657cc9f4c 100644
--- a/fe/fe-connector/fe-connector-cache/src/main/java/org/apache/doris/connector/cache/MetaCacheEntry.java
+++ b/fe/fe-connector/fe-connector-cache/src/main/java/org/apache/doris/connector/cache/MetaCacheEntry.java
@@ -19,6 +19,7 @@
import com.github.benmanes.caffeine.cache.Cache;
import com.github.benmanes.caffeine.cache.LoadingCache;
+import com.github.benmanes.caffeine.cache.Policy;
import com.github.benmanes.caffeine.cache.stats.CacheStats;
import java.util.Objects;
@@ -84,6 +85,14 @@ public MetaCacheEntry(String name, Function loader, CacheSpec cacheSpec, E
public MetaCacheEntry(String name, Function loader, CacheSpec cacheSpec,
ExecutorService refreshExecutor, boolean autoRefresh, boolean contextualOnly,
long refreshAfterWriteSeconds, boolean manualMissLoadEnabled) {
+ this(name, loader, cacheSpec, refreshExecutor, autoRefresh, contextualOnly,
+ refreshAfterWriteSeconds, manualMissLoadEnabled, null);
+ }
+
+ public MetaCacheEntry(String name, Function loader, CacheSpec cacheSpec,
+ ExecutorService refreshExecutor, boolean autoRefresh, boolean contextualOnly,
+ long refreshAfterWriteSeconds, boolean manualMissLoadEnabled,
+ MetaCacheSizeEstimator sizeEstimator) {
this.name = name;
if (contextualOnly) {
if (loader != null) {
@@ -101,22 +110,35 @@ public MetaCacheEntry(String name, Function loader, CacheSpec cacheSpec,
this.refreshAfterWriteSeconds = refreshAfterWriteSeconds;
this.manualMissLoadEnabled = manualMissLoadEnabled;
Objects.requireNonNull(refreshExecutor, "refreshExecutor can not be null");
- this.effectiveEnabled = CacheSpec.isCacheEnabled(
- this.cacheSpec.isEnable(), this.cacheSpec.getTtlSecond(), this.cacheSpec.getCapacity());
+ if (this.cacheSpec.isWeightBounded()) {
+ Objects.requireNonNull(sizeEstimator, "sizeEstimator is required when max-weight is configured");
+ }
+ this.effectiveEnabled = this.cacheSpec.isCacheEnabled();
OptionalLong expireAfterAccessSec =
effectiveEnabled ? CacheSpec.toExpireAfterAccess(this.cacheSpec.getTtlSecond()) : OptionalLong.empty();
OptionalLong refreshAfterWriteSec =
effectiveEnabled && autoRefresh
? OptionalLong.of(refreshAfterWriteSeconds)
: OptionalLong.empty();
- long maxSize = effectiveEnabled ? this.cacheSpec.getCapacity() : 0L;
+ long maxSize = effectiveEnabled && !this.cacheSpec.isWeightBounded()
+ ? this.cacheSpec.getCapacity()
+ : 0L;
CacheFactory cacheFactory = new CacheFactory(
expireAfterAccessSec,
refreshAfterWriteSec,
maxSize,
true,
null);
- this.loadingData = cacheFactory.buildCache(this::loadFromDefaultLoader, refreshExecutor);
+ if (this.cacheSpec.isWeightBounded()) {
+ long maxWeight = effectiveEnabled ? this.cacheSpec.getMaxWeight().getAsLong() : 0L;
+ this.loadingData = cacheFactory.buildCacheWithWeight(
+ this::loadFromDefaultLoader,
+ refreshExecutor,
+ maxWeight,
+ (key, value) -> toCaffeineWeight(sizeEstimator.estimateBytes(key, value)));
+ } else {
+ this.loadingData = cacheFactory.buildCache(this::loadFromDefaultLoader, refreshExecutor);
+ }
this.data = loadingData;
// Initialize striped locks eagerly to keep the hot path allocation-free.
for (int i = 0; i < loadLocks.length; i++) {
@@ -227,6 +249,8 @@ public void forEach(BiConsumer consumer) {
public MetaCacheEntryStats stats() {
CacheStats cacheStats = loadingData.stats();
+ Policy.Eviction evictionPolicy = loadingData.policy().eviction()
+ .orElseThrow(() -> new IllegalStateException("cache has no eviction policy: " + name));
long successCount = loadSuccessCount.get();
long failureCount = loadFailureCount.get();
long totalLoadTime = totalLoadTimeNanos.get();
@@ -237,7 +261,10 @@ public MetaCacheEntryStats stats() {
autoRefresh,
cacheSpec.getTtlSecond(),
cacheSpec.getCapacity(),
+ cacheSpec.isWeightBounded(),
+ cacheSpec.getMaxWeight().orElse(-1L),
data.estimatedSize(),
+ evictionPolicy.weightedSize().orElse(-1L),
cacheStats.requestCount(),
cacheStats.hitCount(),
cacheStats.missCount(),
@@ -247,12 +274,20 @@ public MetaCacheEntryStats stats() {
totalLoadTime,
totalLoadCount == 0 ? 0D : (double) totalLoadTime / totalLoadCount,
cacheStats.evictionCount(),
+ cacheStats.evictionWeight(),
invalidateCount.get(),
lastLoadSuccessTimeMs.get(),
lastLoadFailureTimeMs.get(),
lastError.get());
}
+ private static int toCaffeineWeight(long estimatedBytes) {
+ if (estimatedBytes < 0L) {
+ throw new IllegalArgumentException("estimated cache entry bytes can not be negative: " + estimatedBytes);
+ }
+ return estimatedBytes >= Integer.MAX_VALUE ? Integer.MAX_VALUE : (int) estimatedBytes;
+ }
+
// Injected at construction (fe-core reads Config.enable_external_meta_cache_manual_miss_load dynamically).
private boolean isManualMissLoadEnabled() {
return manualMissLoadEnabled;
diff --git a/fe/fe-connector/fe-connector-cache/src/main/java/org/apache/doris/connector/cache/MetaCacheEntryStats.java b/fe/fe-connector/fe-connector-cache/src/main/java/org/apache/doris/connector/cache/MetaCacheEntryStats.java
index 41c8b89192cd1b..1c8ef49cc59e35 100644
--- a/fe/fe-connector/fe-connector-cache/src/main/java/org/apache/doris/connector/cache/MetaCacheEntryStats.java
+++ b/fe/fe-connector/fe-connector-cache/src/main/java/org/apache/doris/connector/cache/MetaCacheEntryStats.java
@@ -35,6 +35,8 @@
*
* For last-load timestamps, {@code -1} means no corresponding event happened yet.
* {@code lastError} keeps the latest load failure message; empty string means no failure recorded.
+ * Snapshots do not trigger Caffeine maintenance, so size, weight, and eviction values may be briefly stale while
+ * asynchronous maintenance is pending.
*/
public final class MetaCacheEntryStats {
private final boolean configEnabled;
@@ -42,7 +44,10 @@ public final class MetaCacheEntryStats {
private final boolean autoRefresh;
private final long ttlSecond;
private final long capacity;
+ private final boolean weightBounded;
+ private final long maxWeight;
private final long estimatedSize;
+ private final long estimatedWeight;
private final long requestCount;
private final long hitCount;
private final long missCount;
@@ -52,6 +57,7 @@ public final class MetaCacheEntryStats {
private final long totalLoadTimeNanos;
private final double averageLoadPenaltyNanos;
private final long evictionCount;
+ private final long evictionWeight;
private final long invalidateCount;
private final long lastLoadSuccessTimeMs;
private final long lastLoadFailureTimeMs;
@@ -66,7 +72,10 @@ public MetaCacheEntryStats(
boolean autoRefresh,
long ttlSecond,
long capacity,
+ boolean weightBounded,
+ long maxWeight,
long estimatedSize,
+ long estimatedWeight,
long requestCount,
long hitCount,
long missCount,
@@ -76,6 +85,7 @@ public MetaCacheEntryStats(
long totalLoadTimeNanos,
double averageLoadPenaltyNanos,
long evictionCount,
+ long evictionWeight,
long invalidateCount,
long lastLoadSuccessTimeMs,
long lastLoadFailureTimeMs,
@@ -85,7 +95,10 @@ public MetaCacheEntryStats(
this.autoRefresh = autoRefresh;
this.ttlSecond = ttlSecond;
this.capacity = capacity;
+ this.weightBounded = weightBounded;
+ this.maxWeight = maxWeight;
this.estimatedSize = estimatedSize;
+ this.estimatedWeight = estimatedWeight;
this.requestCount = requestCount;
this.hitCount = hitCount;
this.missCount = missCount;
@@ -95,6 +108,7 @@ public MetaCacheEntryStats(
this.totalLoadTimeNanos = totalLoadTimeNanos;
this.averageLoadPenaltyNanos = averageLoadPenaltyNanos;
this.evictionCount = evictionCount;
+ this.evictionWeight = evictionWeight;
this.invalidateCount = invalidateCount;
this.lastLoadSuccessTimeMs = lastLoadSuccessTimeMs;
this.lastLoadFailureTimeMs = lastLoadFailureTimeMs;
@@ -106,7 +120,7 @@ public boolean isConfigEnabled() {
}
/**
- * Effective cache enable state evaluated by {@link CacheSpec#isCacheEnabled(boolean, long, long)}.
+ * Effective cache enable state evaluated by {@link CacheSpec#isCacheEnabled()}.
*/
public boolean isEffectiveEnabled() {
return effectiveEnabled;
@@ -124,10 +138,28 @@ public long getCapacity() {
return capacity;
}
+ public boolean isWeightBounded() {
+ return weightBounded;
+ }
+
+ /**
+ * Returns the configured maximum weight in bytes, or -1 for a count-bounded cache.
+ */
+ public long getMaxWeight() {
+ return maxWeight;
+ }
+
public long getEstimatedSize() {
return estimatedSize;
}
+ /**
+ * Returns Caffeine's current weighted size in bytes, or -1 for a count-bounded cache.
+ */
+ public long getEstimatedWeight() {
+ return estimatedWeight;
+ }
+
public long getRequestCount() {
return requestCount;
}
@@ -167,6 +199,10 @@ public long getEvictionCount() {
return evictionCount;
}
+ public long getEvictionWeight() {
+ return evictionWeight;
+ }
+
public double getEvictionRate() {
if (requestCount == 0) {
return 0D;
diff --git a/fe/fe-connector/fe-connector-cache/src/main/java/org/apache/doris/connector/cache/MetaCacheSizeEstimator.java b/fe/fe-connector/fe-connector-cache/src/main/java/org/apache/doris/connector/cache/MetaCacheSizeEstimator.java
new file mode 100644
index 00000000000000..bf824ad7c45e0e
--- /dev/null
+++ b/fe/fe-connector/fe-connector-cache/src/main/java/org/apache/doris/connector/cache/MetaCacheSizeEstimator.java
@@ -0,0 +1,30 @@
+// 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.connector.cache;
+
+/**
+ * Estimates the retained bytes owned by one connector metadata cache entry.
+ *
+ *
Implementations should be type-specific, deterministic, non-blocking, and normally O(1) at cache admission.
+ * Large object graphs should precompute their owned size when the value is built instead of being traversed again
+ * whenever Caffeine inserts, replaces, refreshes, or computes an entry.
+ */
+@FunctionalInterface
+public interface MetaCacheSizeEstimator {
+ long estimateBytes(K key, V value);
+}
diff --git a/fe/fe-connector/fe-connector-cache/src/main/java/org/apache/doris/connector/cache/ReflectiveObjectSizeEstimator.java b/fe/fe-connector/fe-connector-cache/src/main/java/org/apache/doris/connector/cache/ReflectiveObjectSizeEstimator.java
new file mode 100644
index 00000000000000..4d55d15e3b32da
--- /dev/null
+++ b/fe/fe-connector/fe-connector-cache/src/main/java/org/apache/doris/connector/cache/ReflectiveObjectSizeEstimator.java
@@ -0,0 +1,359 @@
+// 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.connector.cache;
+
+import java.lang.reflect.Array;
+import java.lang.reflect.Field;
+import java.lang.reflect.Modifier;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.IdentityHashMap;
+import java.util.Iterator;
+import java.util.LinkedHashMap;
+import java.util.LinkedHashSet;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.RandomAccess;
+import java.util.Set;
+import java.util.TreeMap;
+import java.util.TreeSet;
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * Bounded reflective object-graph estimator for validating type-specific cache estimators.
+ *
+ * The expensive class discovery is cached in a {@link ClassValue}: each class hierarchy is reflected once to
+ * retain its shallow size and accessible reference fields. Runtime estimation still reads each new root's actual
+ * fields because two instances of the same class can retain very different lists, maps, arrays, and strings.
+ * Collections and object arrays are sampled to keep this fallback bounded on large metadata values.
+ *
+ *
This is intentionally a construction-time safety net, not a Caffeine hit-path weigher. Cache values should
+ * combine it with a type-specific estimate once, store the larger result, and expose that stored number to
+ * Caffeine in O(1).
+ */
+public final class ReflectiveObjectSizeEstimator {
+ private static final int DEFAULT_SAMPLE_SIZE = 5;
+ private static final int DEFAULT_MAX_DEPTH = 20;
+
+ private static final long HASH_MAP_NODE_BYTES = classSize("java.util.HashMap$Node");
+ private static final long LINKED_HASH_MAP_ENTRY_BYTES = classSize("java.util.LinkedHashMap$Entry");
+ private static final long TREE_MAP_ENTRY_BYTES = classSize("java.util.TreeMap$Entry");
+ private static final long LINKED_LIST_NODE_BYTES = classSize("java.util.LinkedList$Node");
+ private static final long CONCURRENT_HASH_MAP_NODE_BYTES = classSize("java.util.concurrent.ConcurrentHashMap$Node");
+
+ private static final Set> SHALLOW_LEAF_TYPES = Set.of(
+ Boolean.class,
+ Byte.class,
+ Character.class,
+ Short.class,
+ Integer.class,
+ Float.class,
+ Long.class,
+ Double.class);
+
+ private static final ClassValue CLASS_PLANS = new ClassValue<>() {
+ @Override
+ protected ClassPlan computeValue(Class> type) {
+ List referenceFields = new ArrayList<>();
+ for (Class> current = type; current != null; current = current.getSuperclass()) {
+ for (Field field : current.getDeclaredFields()) {
+ if (Modifier.isStatic(field.getModifiers())
+ || field.getType().isPrimitive()
+ || field.getType().isEnum()) {
+ continue;
+ }
+ try {
+ if (field.trySetAccessible()) {
+ referenceFields.add(field);
+ }
+ } catch (RuntimeException ignored) {
+ // Strongly encapsulated JDK fields are expected; container storage is modeled separately.
+ }
+ }
+ }
+ return new ClassPlan(
+ JvmSizeUtils.instanceSize(type),
+ referenceFields.toArray(new Field[0]));
+ }
+ };
+
+ private ReflectiveObjectSizeEstimator() {
+ }
+
+ public static long estimate(Object root) {
+ return estimate(root, DEFAULT_SAMPLE_SIZE, DEFAULT_MAX_DEPTH);
+ }
+
+ public static long estimate(Object root, int sampleSize, int maxDepth) {
+ if (sampleSize <= 0) {
+ throw new IllegalArgumentException("sampleSize must be positive: " + sampleSize);
+ }
+ if (maxDepth < 0) {
+ throw new IllegalArgumentException("maxDepth can not be negative: " + maxDepth);
+ }
+ return new Walker(sampleSize).estimate(root, maxDepth);
+ }
+
+ private static final class Walker {
+ private final int sampleSize;
+ private final Map visited = new IdentityHashMap<>();
+
+ private Walker(int sampleSize) {
+ this.sampleSize = sampleSize;
+ }
+
+ private long estimate(Object value, int depth) {
+ if (value == null || visited.put(value, Boolean.TRUE) != null) {
+ return 0L;
+ }
+
+ Class> type = value.getClass();
+ if (type.isEnum() || value instanceof Class>) {
+ return 0L;
+ }
+ if (value instanceof String) {
+ return JvmSizeUtils.stringSize((String) value);
+ }
+ if (SHALLOW_LEAF_TYPES.contains(type) || type.isHidden()) {
+ return JvmSizeUtils.instanceSize(type);
+ }
+ if (type.isArray()) {
+ return estimateArray(value, depth);
+ }
+ if (value instanceof ByteBuffer) {
+ return estimateByteBuffer((ByteBuffer) value, depth);
+ }
+ if (value instanceof Optional>) {
+ return estimateOptional((Optional>) value, depth);
+ }
+ if (value instanceof Map, ?>) {
+ return estimateMap((Map, ?>) value, depth);
+ }
+ if (value instanceof Collection>) {
+ return estimateCollection((Collection>) value, depth);
+ }
+
+ ClassPlan plan = CLASS_PLANS.get(type);
+ long bytes = plan.shallowBytes;
+ if (depth == 0) {
+ return bytes;
+ }
+ for (Field field : plan.referenceFields) {
+ bytes = add(bytes, estimateField(value, field, depth - 1));
+ }
+ return bytes;
+ }
+
+ private long estimateArray(Object value, int depth) {
+ int length = Array.getLength(value);
+ Class> componentType = value.getClass().getComponentType();
+ if (componentType.isPrimitive()) {
+ return JvmSizeUtils.primitiveArraySize(componentType, length);
+ }
+
+ long bytes = JvmSizeUtils.objectArraySize(length);
+ if (depth == 0 || length == 0) {
+ return bytes;
+ }
+ int samples = Math.min(length, sampleSize);
+ long sampledBytes = 0L;
+ for (int i = 0; i < samples; i++) {
+ int index = sampleIndex(i, samples, length);
+ sampledBytes = add(sampledBytes, estimate(Array.get(value, index), depth - 1));
+ }
+ return add(bytes, scale(sampledBytes, samples, length));
+ }
+
+ private long estimateByteBuffer(ByteBuffer value, int depth) {
+ long bytes = JvmSizeUtils.instanceSize(value.getClass());
+ if (depth > 0 && value.hasArray()) {
+ bytes = add(bytes, estimate(value.array(), depth - 1));
+ }
+ return bytes;
+ }
+
+ private long estimateOptional(Optional> value, int depth) {
+ long bytes = JvmSizeUtils.instanceSize(value.getClass());
+ return depth == 0 || value.isEmpty()
+ ? bytes
+ : add(bytes, estimate(value.get(), depth - 1));
+ }
+
+ private long estimateCollection(Collection> values, int depth) {
+ int size = values.size();
+ long bytes = add(
+ JvmSizeUtils.instanceSize(values.getClass()),
+ collectionStorageBytes(values, size));
+ if (depth == 0 || size == 0) {
+ return bytes;
+ }
+
+ int samples = Math.min(size, sampleSize);
+ long sampledBytes = 0L;
+ if (values instanceof List> && values instanceof RandomAccess) {
+ List> list = (List>) values;
+ for (int i = 0; i < samples; i++) {
+ sampledBytes = add(sampledBytes,
+ estimate(list.get(sampleIndex(i, samples, size)), depth - 1));
+ }
+ } else {
+ sampledBytes = estimateIterableSamples(values, samples, depth - 1);
+ }
+ return add(bytes, scale(sampledBytes, samples, size));
+ }
+
+ private long estimateMap(Map, ?> values, int depth) {
+ int size = values.size();
+ long bytes = add(
+ JvmSizeUtils.instanceSize(values.getClass()),
+ mapStorageBytes(values, size));
+ if (depth == 0 || size == 0) {
+ return bytes;
+ }
+
+ int samples = Math.min(size, sampleSize);
+ long sampledBytes = 0L;
+ Iterator extends Map.Entry, ?>> iterator = values.entrySet().iterator();
+ for (int i = 0; i < samples; i++) {
+ Map.Entry, ?> entry = iterator.next();
+ sampledBytes = add(sampledBytes, estimate(entry.getKey(), depth - 1));
+ sampledBytes = add(sampledBytes, estimate(entry.getValue(), depth - 1));
+ }
+ return add(bytes, scale(sampledBytes, samples, size));
+ }
+
+ private long estimateIterableSamples(Collection> values, int samples, int depth) {
+ Iterator> iterator = values.iterator();
+ long sampledBytes = 0L;
+ for (int i = 0; i < samples; i++) {
+ sampledBytes = add(sampledBytes, estimate(iterator.next(), depth));
+ }
+ return sampledBytes;
+ }
+
+ private long estimateField(Object owner, Field field, int depth) {
+ try {
+ return estimate(field.get(owner), depth);
+ } catch (IllegalAccessException e) {
+ throw new IllegalStateException("Cached reference field is no longer accessible: " + field, e);
+ }
+ }
+ }
+
+ private static long collectionStorageBytes(Collection> values, int size) {
+ if (size == 0) {
+ return 0L;
+ }
+ if (values instanceof LinkedHashSet>) {
+ return hashStorageBytes(size, LINKED_HASH_MAP_ENTRY_BYTES);
+ }
+ if (values instanceof HashSet>) {
+ return hashStorageBytes(size, HASH_MAP_NODE_BYTES);
+ }
+ if (values instanceof TreeSet>) {
+ return multiply(size, TREE_MAP_ENTRY_BYTES);
+ }
+ if (values instanceof LinkedList>) {
+ return multiply(size, LINKED_LIST_NODE_BYTES);
+ }
+ return JvmSizeUtils.objectArraySize(size);
+ }
+
+ private static long mapStorageBytes(Map, ?> values, int size) {
+ if (size == 0) {
+ return 0L;
+ }
+ if (values instanceof LinkedHashMap, ?>) {
+ return hashStorageBytes(size, LINKED_HASH_MAP_ENTRY_BYTES);
+ }
+ if (values instanceof HashMap, ?>) {
+ return hashStorageBytes(size, HASH_MAP_NODE_BYTES);
+ }
+ if (values instanceof ConcurrentHashMap, ?>) {
+ return hashStorageBytes(size, CONCURRENT_HASH_MAP_NODE_BYTES);
+ }
+ if (values instanceof TreeMap, ?>) {
+ return multiply(size, TREE_MAP_ENTRY_BYTES);
+ }
+ return JvmSizeUtils.objectArraySize(saturatedDouble(size));
+ }
+
+ private static long hashStorageBytes(int size, long nodeBytes) {
+ return add(
+ JvmSizeUtils.objectArraySize(hashCapacity(size)),
+ multiply(size, nodeBytes));
+ }
+
+ private static int hashCapacity(int size) {
+ long needed = (size * 4L + 2L) / 3L;
+ int capacity = 16;
+ while (capacity < needed && capacity < 1 << 30) {
+ capacity <<= 1;
+ }
+ return capacity;
+ }
+
+ private static int sampleIndex(int sample, int sampleCount, int totalCount) {
+ return sampleCount == 1
+ ? 0
+ : (int) ((long) sample * (totalCount - 1) / (sampleCount - 1));
+ }
+
+ private static long scale(long sampledBytes, int sampleCount, int totalCount) {
+ if (sampleCount == totalCount) {
+ return sampledBytes;
+ }
+ double scaled = (double) sampledBytes * totalCount / sampleCount;
+ return scaled >= Long.MAX_VALUE ? Long.MAX_VALUE : (long) scaled;
+ }
+
+ private static int saturatedDouble(int value) {
+ return value > Integer.MAX_VALUE / 2 ? Integer.MAX_VALUE : value * 2;
+ }
+
+ private static long classSize(String className) {
+ try {
+ return JvmSizeUtils.instanceSize(Class.forName(className));
+ } catch (ClassNotFoundException e) {
+ throw new IllegalStateException("Required JVM collection class is missing: " + className, e);
+ }
+ }
+
+ private static long multiply(long left, long right) {
+ return JvmSizeUtils.saturatedMultiply(left, right);
+ }
+
+ private static long add(long left, long right) {
+ return JvmSizeUtils.saturatedAdd(left, right);
+ }
+
+ private static final class ClassPlan {
+ private final long shallowBytes;
+ private final Field[] referenceFields;
+
+ private ClassPlan(long shallowBytes, Field[] referenceFields) {
+ this.shallowBytes = shallowBytes;
+ this.referenceFields = referenceFields;
+ }
+ }
+}
diff --git a/fe/fe-connector/fe-connector-cache/src/test/java/org/apache/doris/connector/cache/CacheSpecTest.java b/fe/fe-connector/fe-connector-cache/src/test/java/org/apache/doris/connector/cache/CacheSpecTest.java
index 276735b40c2f9b..27ced4beb40377 100644
--- a/fe/fe-connector/fe-connector-cache/src/test/java/org/apache/doris/connector/cache/CacheSpecTest.java
+++ b/fe/fe-connector/fe-connector-cache/src/test/java/org/apache/doris/connector/cache/CacheSpecTest.java
@@ -247,4 +247,42 @@ public void isMetaCacheKeyForEngine() {
CacheSpec.isMetaCacheKeyForEngine("meta.cache.paimon.table.ttl-second", "iceberg"));
Assertions.assertFalse(CacheSpec.isMetaCacheKeyForEngine(null, "iceberg"));
}
+
+ @Test
+ public void fromPropertiesParsesMaximumWeight() {
+ Map properties = new HashMap<>();
+ properties.put("meta.cache.hive.file.max-weight", "512MB");
+
+ CacheSpec spec = CacheSpec.fromProperties(
+ properties, "hive", "file", CacheSpec.of(true, 60L, 100L));
+
+ Assertions.assertTrue(spec.isWeightBounded());
+ Assertions.assertEquals(512L * 1024 * 1024, spec.getMaxWeight().getAsLong());
+ Assertions.assertTrue(spec.isCacheEnabled());
+ }
+
+ @Test
+ public void maximumWeightOverridesCountCapacity() {
+ CacheSpec spec = CacheSpec.ofWeight(true, CacheSpec.CACHE_NO_TTL, 0L, 1024L);
+ Assertions.assertTrue(spec.isCacheEnabled());
+
+ CacheSpec disabled = CacheSpec.ofWeight(true, CacheSpec.CACHE_NO_TTL, 100L, 0L);
+ Assertions.assertFalse(disabled.isCacheEnabled());
+ }
+
+ @Test
+ public void maximumWeightRejectsInvalidValues() {
+ String key = "meta.cache.iceberg.table.max-weight";
+ for (String value : new String[] {"abc", "-1", "9223372036854775807KB"}) {
+ Map properties = new HashMap<>();
+ properties.put(key, value);
+ IllegalArgumentException error = Assertions.assertThrows(IllegalArgumentException.class,
+ () -> CacheSpec.fromProperties(
+ properties, "iceberg", "table", CacheSpec.of(true, 60L, 100L)));
+ Assertions.assertEquals("The parameter " + key + " is wrong, value is " + value,
+ error.getMessage());
+ Assertions.assertThrows(IllegalArgumentException.class,
+ () -> CacheSpec.checkDataSizeProperty(value, key));
+ }
+ }
}
diff --git a/fe/fe-connector/fe-connector-cache/src/test/java/org/apache/doris/connector/cache/ConnectorMetadataCacheTest.java b/fe/fe-connector/fe-connector-cache/src/test/java/org/apache/doris/connector/cache/ConnectorMetadataCacheTest.java
index 8a6c1596fccad4..224b29e8da8c95 100644
--- a/fe/fe-connector/fe-connector-cache/src/test/java/org/apache/doris/connector/cache/ConnectorMetadataCacheTest.java
+++ b/fe/fe-connector/fe-connector-cache/src/test/java/org/apache/doris/connector/cache/ConnectorMetadataCacheTest.java
@@ -263,6 +263,33 @@ public void defaultsAreEnabledWithNoProperties() {
Assertions.assertTrue(cache.isEnabled(), "the cache must be ON by default with no override properties");
}
+ @Test
+ public void maxWeightRequiresEstimator() {
+ Map props = new HashMap<>();
+ props.put("meta.cache." + ENGINE + ".partition_view.max-weight", "1KB");
+
+ Assertions.assertThrows(NullPointerException.class,
+ () -> new ConnectorMetadataCache<>(ENGINE, "partition_view", props));
+ }
+
+ @Test
+ public void maxWeightUsesProvidedEstimator() {
+ Map props = new HashMap<>();
+ props.put("meta.cache." + ENGINE + ".partition_view.max-weight", "1B");
+ AtomicInteger estimates = new AtomicInteger();
+ ConnectorMetadataCache cache = new ConnectorMetadataCache<>(
+ ENGINE, "partition_view", props, (key, value) -> {
+ estimates.incrementAndGet();
+ return value.length();
+ });
+
+ String value = cache.get(key("db", "t", 5L, 1L), () -> "weighted");
+
+ Assertions.assertEquals("weighted", value);
+ Assertions.assertEquals(1, estimates.get(), "the max-weight branch must call the provided estimator");
+ Assertions.assertTrue(cache.isWeightBounded());
+ }
+
@Test
public void loaderExceptionPropagatesUnwrappedAndIsNotCached() {
ConnectorMetadataCache cache = newCache();
diff --git a/fe/fe-connector/fe-connector-cache/src/test/java/org/apache/doris/connector/cache/MetaCacheEntryTest.java b/fe/fe-connector/fe-connector-cache/src/test/java/org/apache/doris/connector/cache/MetaCacheEntryTest.java
index 8284f4ae1ed672..287f98098f53f8 100644
--- a/fe/fe-connector/fe-connector-cache/src/test/java/org/apache/doris/connector/cache/MetaCacheEntryTest.java
+++ b/fe/fe-connector/fe-connector-cache/src/test/java/org/apache/doris/connector/cache/MetaCacheEntryTest.java
@@ -243,6 +243,66 @@ public void putIfNotInvalidatedSinceHonorsGenerationGuard() {
}
}
+ @Test
+ public void weightedEntryEvictsAndReportsWeight() throws Exception {
+ ExecutorService refreshExecutor = Executors.newSingleThreadExecutor();
+ try {
+ CacheSpec cacheSpec = CacheSpec.ofWeight(true, CacheSpec.CACHE_NO_TTL, 100L, 4L);
+ MetaCacheEntry entry = new MetaCacheEntry<>(
+ "weighted", key -> key, cacheSpec, refreshExecutor,
+ false, false, 0L, false, (key, value) -> value.length());
+
+ entry.get("aaa");
+ entry.get("bbb");
+ extractLoadingCache(entry).cleanUp();
+
+ MetaCacheEntryStats stats = entry.stats();
+ Assertions.assertTrue(stats.isWeightBounded());
+ Assertions.assertEquals(4L, stats.getMaxWeight());
+ Assertions.assertTrue(stats.getEstimatedWeight() <= 4L);
+ Assertions.assertEquals(1L, stats.getEvictionCount());
+ Assertions.assertEquals(3L, stats.getEvictionWeight());
+ } finally {
+ refreshExecutor.shutdownNow();
+ }
+ }
+
+ @Test
+ public void weightedEntryRequiresEstimator() {
+ ExecutorService refreshExecutor = Executors.newSingleThreadExecutor();
+ try {
+ CacheSpec cacheSpec = CacheSpec.ofWeight(true, CacheSpec.CACHE_NO_TTL, 100L, 4L);
+ NullPointerException error = Assertions.assertThrows(NullPointerException.class,
+ () -> new MetaCacheEntry<>("weighted", String::length, cacheSpec, refreshExecutor));
+ Assertions.assertEquals("sizeEstimator is required when max-weight is configured", error.getMessage());
+ } finally {
+ refreshExecutor.shutdownNow();
+ }
+ }
+
+ @Test
+ public void weightedEntryRejectsNegativeAndSaturatesLargeEstimate() throws Exception {
+ ExecutorService refreshExecutor = Executors.newSingleThreadExecutor();
+ try {
+ CacheSpec smallSpec = CacheSpec.ofWeight(true, CacheSpec.CACHE_NO_TTL, 100L, 4L);
+ MetaCacheEntry negative = new MetaCacheEntry<>(
+ "negative", key -> key, smallSpec, refreshExecutor,
+ false, false, 0L, false, (key, value) -> -1L);
+ Assertions.assertThrows(IllegalArgumentException.class, () -> negative.get("a"));
+
+ CacheSpec largeSpec = CacheSpec.ofWeight(
+ true, CacheSpec.CACHE_NO_TTL, 100L, Integer.MAX_VALUE);
+ MetaCacheEntry saturated = new MetaCacheEntry<>(
+ "saturated", key -> key, largeSpec, refreshExecutor,
+ false, false, 0L, false, (key, value) -> Long.MAX_VALUE);
+ saturated.get("a");
+ extractLoadingCache(saturated).cleanUp();
+ Assertions.assertEquals(Integer.MAX_VALUE, saturated.stats().getEstimatedWeight());
+ } finally {
+ refreshExecutor.shutdownNow();
+ }
+ }
+
@SuppressWarnings("unchecked")
private static LoadingCache extractLoadingCache(MetaCacheEntry, ?> entry) throws Exception {
Field field = MetaCacheEntry.class.getDeclaredField("loadingData");
diff --git a/fe/fe-connector/fe-connector-cache/src/test/java/org/apache/doris/connector/cache/ReflectiveObjectSizeEstimatorTest.java b/fe/fe-connector/fe-connector-cache/src/test/java/org/apache/doris/connector/cache/ReflectiveObjectSizeEstimatorTest.java
new file mode 100644
index 00000000000000..2ec71bf27611cf
--- /dev/null
+++ b/fe/fe-connector/fe-connector-cache/src/test/java/org/apache/doris/connector/cache/ReflectiveObjectSizeEstimatorTest.java
@@ -0,0 +1,123 @@
+// 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.connector.cache;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+public class ReflectiveObjectSizeEstimatorTest {
+
+ @Test
+ public void usesActualRuntimeCollectionSize() {
+ Holder small = new Holder(strings(1));
+ Holder large = new Holder(strings(100));
+
+ long smallBytes = ReflectiveObjectSizeEstimator.estimate(small);
+ long largeBytes = ReflectiveObjectSizeEstimator.estimate(large);
+
+ Assertions.assertTrue(largeBytes > smallBytes);
+ }
+
+ @Test
+ public void usesActualRuntimeMapSizeWithCachedClassPlan() {
+ Map values = new HashMap<>();
+ values.put("key-0", "value-0");
+ MapHolder holder = new MapHolder(values);
+ long smallBytes = ReflectiveObjectSizeEstimator.estimate(holder);
+
+ for (int i = 1; i < 100; i++) {
+ values.put("key-" + i, "value-" + i);
+ }
+
+ Assertions.assertTrue(ReflectiveObjectSizeEstimator.estimate(holder) > smallBytes);
+ }
+
+ @Test
+ public void countsSharedReferenceOnceAndStopsCycles() {
+ String shared = new String("shared-value");
+ SharedHolder holder = new SharedHolder(shared, shared);
+ long expected = JvmSizeUtils.instanceSize(SharedHolder.class) + JvmSizeUtils.stringSize(shared);
+
+ Assertions.assertEquals(expected, ReflectiveObjectSizeEstimator.estimate(holder));
+
+ Cycle cycle = new Cycle();
+ cycle.next = cycle;
+ Assertions.assertEquals(
+ JvmSizeUtils.instanceSize(Cycle.class),
+ ReflectiveObjectSizeEstimator.estimate(cycle));
+ }
+
+ @Test
+ public void usesExactPrimitiveArrayLayout() {
+ int[] values = new int[37];
+ Assertions.assertEquals(
+ JvmSizeUtils.intArraySize(values.length),
+ ReflectiveObjectSizeEstimator.estimate(values));
+ }
+
+ @Test
+ public void validatesBounds() {
+ Assertions.assertThrows(IllegalArgumentException.class,
+ () -> ReflectiveObjectSizeEstimator.estimate(new Object(), 0, 1));
+ Assertions.assertThrows(IllegalArgumentException.class,
+ () -> ReflectiveObjectSizeEstimator.estimate(new Object(), 1, -1));
+ }
+
+ private static List strings(int size) {
+ List values = new ArrayList<>(size);
+ for (int i = 0; i < size; i++) {
+ values.add(new String("value-" + i));
+ }
+ return values;
+ }
+
+ private static final class Holder {
+ private final List values;
+
+ private Holder(List values) {
+ this.values = values;
+ }
+ }
+
+ private static final class SharedHolder {
+ private final String left;
+ private final String right;
+
+ private SharedHolder(String left, String right) {
+ this.left = left;
+ this.right = right;
+ }
+ }
+
+ private static final class MapHolder {
+ private final Map values;
+
+ private MapHolder(Map values) {
+ this.values = values;
+ }
+ }
+
+ private static final class Cycle {
+ private Cycle next;
+ }
+}
diff --git a/fe/fe-connector/fe-connector-hive/src/main/java/org/apache/doris/connector/hive/HiveCatalogProperties.java b/fe/fe-connector/fe-connector-hive/src/main/java/org/apache/doris/connector/hive/HiveCatalogProperties.java
index d38571a11f2503..c16c7dfb14507e 100644
--- a/fe/fe-connector/fe-connector-hive/src/main/java/org/apache/doris/connector/hive/HiveCatalogProperties.java
+++ b/fe/fe-connector/fe-connector-hive/src/main/java/org/apache/doris/connector/hive/HiveCatalogProperties.java
@@ -187,6 +187,9 @@ public HiveCatalogProperties checkCreateTimeOnlyRules() {
// "The parameter ... is wrong, value is ..." message.
CacheSpec.checkLongProperty(raw.get("file.meta.cache.ttl-second"), 0L, "file.meta.cache.ttl-second");
CacheSpec.checkLongProperty(raw.get("partition.cache.ttl-second"), 0L, "partition.cache.ttl-second");
+ String fileMaxWeightKey = CacheSpec.metaCacheMaxWeightKey(
+ HiveFileListingCache.ENGINE, HiveFileListingCache.ENTRY_FILE);
+ CacheSpec.checkDataSizeProperty(raw.get(fileMaxWeightKey), fileMaxWeightKey);
return this;
}
diff --git a/fe/fe-connector/fe-connector-hive/src/main/java/org/apache/doris/connector/hive/HiveFileListingCache.java b/fe/fe-connector/fe-connector-hive/src/main/java/org/apache/doris/connector/hive/HiveFileListingCache.java
index af7d2ba558acc8..f7814e35e964a0 100644
--- a/fe/fe-connector/fe-connector-hive/src/main/java/org/apache/doris/connector/hive/HiveFileListingCache.java
+++ b/fe/fe-connector/fe-connector-hive/src/main/java/org/apache/doris/connector/hive/HiveFileListingCache.java
@@ -19,6 +19,7 @@
import org.apache.doris.connector.cache.CacheSpec;
import org.apache.doris.connector.cache.MetaCacheEntry;
+import org.apache.doris.connector.cache.MetaCacheEntryStats;
import org.apache.doris.connector.spi.DorisConnectorException;
import org.apache.doris.filesystem.FileEntry;
import org.apache.doris.filesystem.FileIterator;
@@ -104,7 +105,7 @@ interface DirectoryLister {
List list(String location, FileSystem fs);
}
- private final MetaCacheEntry> cache;
+ private final MetaCacheEntry cache;
private final DirectoryLister lister;
public HiveFileListingCache(HiveCatalogProperties properties) {
@@ -138,7 +139,9 @@ private static DirectoryLister defaultLister(HiveCatalogProperties properties) {
CacheSpec.of(true, DEFAULT_TTL_SECOND, DEFAULT_FILE_CAPACITY));
// Contextual-only + manual-miss so the slow listStatus runs on the caller (TCCL-pinned) thread outside
// Caffeine's sync compute lock, deduplicated by a striped lock — mirrors CachingHmsClient's entries.
- this.cache = new MetaCacheEntry<>("hive.file", null, spec, ForkJoinPool.commonPool(), false, true, 0L, true);
+ this.cache = new MetaCacheEntry<>(
+ "hive.file", null, spec, ForkJoinPool.commonPool(), false, true, 0L, true,
+ HiveFileListingSizeEstimator::estimateEntry);
this.lister = Objects.requireNonNull(lister, "lister can not be null");
}
@@ -165,7 +168,7 @@ public List listDataFiles(String dbName, String tableName, Strin
public List listDataFiles(String dbName, String tableName, String location,
List partitionValues, FileSystem fs) {
return cache.get(new FileListingKey(dbName, tableName, location, partitionValues),
- key -> lister.list(key.location, fs));
+ key -> new FileListingValue(lister.list(key.location, fs))).files;
}
/** Drops every cached listing for one table. Backs {@code REFRESH TABLE}. */
@@ -206,6 +209,10 @@ long size() {
return count[0];
}
+ MetaCacheEntryStats stats() {
+ return cache.stats();
+ }
+
/**
* The production {@link DirectoryLister}: a LITERAL listing through the engine-injected Doris
* {@link FileSystem} (a per-catalog {@code SpiSwitchingFileSystem}), filtering out directories and
@@ -334,10 +341,11 @@ private static boolean isSystemicResolutionFailure(Throwable t) {
* size-estimate paths sharing the same entry while making per-partition invalidation possible.
*/
static final class FileListingKey {
- private final String dbName;
- private final String tableName;
- private final String location;
- private final List partitionValues;
+ final String dbName;
+ final String tableName;
+ final String location;
+ final List partitionValues;
+ final long estimatedBytes;
FileListingKey(String dbName, String tableName, String location, List partitionValues) {
this.dbName = dbName;
@@ -346,6 +354,7 @@ static final class FileListingKey {
this.partitionValues = partitionValues == null
? Collections.emptyList()
: Collections.unmodifiableList(new ArrayList<>(partitionValues));
+ this.estimatedBytes = HiveFileListingSizeEstimator.estimateKey(this);
}
boolean matches(String db, String table) {
@@ -376,4 +385,14 @@ public int hashCode() {
return Objects.hash(dbName, tableName, location, partitionValues);
}
}
+
+ static final class FileListingValue {
+ final List files;
+ final long estimatedBytes;
+
+ FileListingValue(List files) {
+ this.files = Collections.unmodifiableList(new ArrayList<>(files));
+ this.estimatedBytes = HiveFileListingSizeEstimator.estimateValue(this);
+ }
+ }
}
diff --git a/fe/fe-connector/fe-connector-hive/src/main/java/org/apache/doris/connector/hive/HiveFileListingSizeEstimator.java b/fe/fe-connector/fe-connector-hive/src/main/java/org/apache/doris/connector/hive/HiveFileListingSizeEstimator.java
new file mode 100644
index 00000000000000..641b3e1957e5d5
--- /dev/null
+++ b/fe/fe-connector/fe-connector-hive/src/main/java/org/apache/doris/connector/hive/HiveFileListingSizeEstimator.java
@@ -0,0 +1,73 @@
+// 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.connector.hive;
+
+import org.apache.doris.connector.cache.JvmSizeUtils;
+import org.apache.doris.connector.hive.HiveFileListingCache.FileListingKey;
+import org.apache.doris.connector.hive.HiveFileListingCache.FileListingValue;
+
+/** Type-specific retained-heap estimator for one Hive directory-listing cache entry. */
+final class HiveFileListingSizeEstimator {
+ private static final long KEY_SHALLOW_BYTES = JvmSizeUtils.instanceSize(FileListingKey.class);
+ private static final long VALUE_SHALLOW_BYTES = JvmSizeUtils.instanceSize(FileListingValue.class);
+ private static final long FILE_STATUS_SHALLOW_BYTES = JvmSizeUtils.instanceSize(HiveFileStatus.class);
+
+ private HiveFileListingSizeEstimator() {
+ }
+
+ /** Caffeine callback: both key and value sizes were computed once during construction. */
+ static long estimateEntry(FileListingKey key, FileListingValue value) {
+ return add(key.estimatedBytes, value.estimatedBytes);
+ }
+
+ static long estimateKey(FileListingKey key) {
+ long bytes = KEY_SHALLOW_BYTES;
+ bytes = add(bytes, JvmSizeUtils.stringSize(key.dbName));
+ bytes = add(bytes, JvmSizeUtils.stringSize(key.tableName));
+ bytes = add(bytes, JvmSizeUtils.stringSize(key.location));
+ bytes = add(bytes, estimateOwnedStringList(key.partitionValues));
+ return bytes;
+ }
+
+ static long estimateValue(FileListingValue value) {
+ long bytes = VALUE_SHALLOW_BYTES;
+ bytes = add(bytes, estimateArrayBackedList(value.files));
+ for (HiveFileStatus file : value.files) {
+ bytes = add(bytes, FILE_STATUS_SHALLOW_BYTES);
+ bytes = add(bytes, JvmSizeUtils.stringSize(file.getPath()));
+ }
+ return bytes;
+ }
+
+ private static long estimateOwnedStringList(java.util.List values) {
+ long bytes = estimateArrayBackedList(values);
+ for (String value : values) {
+ bytes = add(bytes, JvmSizeUtils.stringSize(value));
+ }
+ return bytes;
+ }
+
+ private static long estimateArrayBackedList(java.util.List> values) {
+ long bytes = JvmSizeUtils.instanceSize(values.getClass());
+ return add(bytes, JvmSizeUtils.arrayListSize(values.size()));
+ }
+
+ private static long add(long left, long right) {
+ return JvmSizeUtils.saturatedAdd(left, right);
+ }
+}
diff --git a/fe/fe-connector/fe-connector-hive/src/test/java/org/apache/doris/connector/hive/HiveFileListingCacheTest.java b/fe/fe-connector/fe-connector-hive/src/test/java/org/apache/doris/connector/hive/HiveFileListingCacheTest.java
index 08f567a9d718a8..5206f18b279fc1 100644
--- a/fe/fe-connector/fe-connector-hive/src/test/java/org/apache/doris/connector/hive/HiveFileListingCacheTest.java
+++ b/fe/fe-connector/fe-connector-hive/src/test/java/org/apache/doris/connector/hive/HiveFileListingCacheTest.java
@@ -192,6 +192,29 @@ public void legacyFileMetaCacheTtlZeroBypassesTheCache() {
Assertions.assertEquals(2, lister.totalCalls);
}
+ @Test
+ public void maximumWeightUsesTypedFileListingEstimator() {
+ CountingLister lister = new CountingLister();
+ HiveFileListingCache cache = new HiveFileListingCache(
+ HiveCatalogProperties.of(props("meta.cache.hive.file.max-weight", "1MB")), lister);
+
+ cache.listDataFiles("db", "table", "s3://bucket/table/p=1", Collections.singletonList("1"), FS);
+
+ Assertions.assertTrue(cache.stats().isWeightBounded());
+ Assertions.assertEquals(1024L * 1024, cache.stats().getMaxWeight());
+ Assertions.assertTrue(cache.stats().getEstimatedWeight() > 0L);
+ }
+
+ @Test
+ public void invalidMaximumWeightIsRejectedAtCreateTime() {
+ HiveCatalogProperties properties = HiveCatalogProperties.of(
+ props("meta.cache.hive.file.max-weight", "broken"));
+ IllegalArgumentException error = Assertions.assertThrows(
+ IllegalArgumentException.class, properties::checkCreateTimeOnlyRules);
+ Assertions.assertEquals(
+ "The parameter meta.cache.hive.file.max-weight is wrong, value is broken", error.getMessage());
+ }
+
// ==================== the production lister: filters dirs + hidden files, lists literally ====================
@Test
diff --git a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergCacheSizeEstimator.java b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergCacheSizeEstimator.java
new file mode 100644
index 00000000000000..0ec5079097e6bd
--- /dev/null
+++ b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergCacheSizeEstimator.java
@@ -0,0 +1,619 @@
+// 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.connector.iceberg;
+
+import org.apache.doris.connector.cache.JvmSizeUtils;
+import org.apache.doris.connector.cache.ReflectiveObjectSizeEstimator;
+import org.apache.doris.connector.iceberg.IcebergPartitionCache.CachedPartitions;
+import org.apache.doris.connector.iceberg.IcebergPartitionCache.Key;
+import org.apache.doris.connector.iceberg.IcebergPartitionUtils.IcebergRawPartition;
+import org.apache.doris.connector.iceberg.IcebergTableCache.CachedTable;
+
+import org.apache.iceberg.BlobMetadata;
+import org.apache.iceberg.ContentFile;
+import org.apache.iceberg.DeleteFile;
+import org.apache.iceberg.HasTableOperations;
+import org.apache.iceberg.HistoryEntry;
+import org.apache.iceberg.MetadataUpdate;
+import org.apache.iceberg.PartitionField;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.PartitionStatisticsFile;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.Snapshot;
+import org.apache.iceberg.SnapshotRef;
+import org.apache.iceberg.SortField;
+import org.apache.iceberg.SortOrder;
+import org.apache.iceberg.StatisticsFile;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.TableMetadata;
+import org.apache.iceberg.TableOperations;
+import org.apache.iceberg.UnboundPartitionSpec;
+import org.apache.iceberg.catalog.Namespace;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.types.Type;
+import org.apache.iceberg.types.Types;
+
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.IdentityHashMap;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+/** Type-specific retained-heap estimators for the large Iceberg connector cache values. */
+final class IcebergCacheSizeEstimator {
+ private static final String[] CONTENT_FILE_FIELD_NAMES = {
+ "content", "file_path", "file_format", "partition", "record_count", "file_size_in_bytes",
+ "column_sizes", "value_counts", "null_value_counts", "nan_value_counts", "lower_bounds",
+ "upper_bounds", "key_metadata", "split_offsets", "equality_ids", "sort_order_id", "first_row_id",
+ "referenced_data_file", "content_offset", "content_size_in_bytes"
+ };
+ private static final long TABLE_IDENTIFIER_SHALLOW_BYTES = JvmSizeUtils.instanceSize(TableIdentifier.class);
+ private static final long CACHED_TABLE_SHALLOW_BYTES = JvmSizeUtils.instanceSize(CachedTable.class);
+ private static final long TABLE_METADATA_SHALLOW_BYTES = JvmSizeUtils.instanceSize(TableMetadata.class);
+ private static final long PARTITION_KEY_SHALLOW_BYTES = JvmSizeUtils.instanceSize(Key.class);
+ private static final long CACHED_PARTITIONS_SHALLOW_BYTES = JvmSizeUtils.instanceSize(CachedPartitions.class);
+ private static final long RAW_PARTITION_SHALLOW_BYTES = JvmSizeUtils.instanceSize(IcebergRawPartition.class);
+ private static final long MANIFEST_KEY_SHALLOW_BYTES = JvmSizeUtils.instanceSize(IcebergManifestEntryKey.class);
+ private static final long MANIFEST_VALUE_SHALLOW_BYTES = JvmSizeUtils.instanceSize(ManifestCacheValue.class);
+ private static final long INTEGER_SHALLOW_BYTES = JvmSizeUtils.instanceSize(Integer.class);
+ private static final long LONG_SHALLOW_BYTES = JvmSizeUtils.instanceSize(Long.class);
+ private static final long HASH_MAP_NODE_SHALLOW_BYTES = classSize("java.util.HashMap$Node");
+ private static final long LINKED_HASH_MAP_ENTRY_SHALLOW_BYTES = classSize("java.util.LinkedHashMap$Entry");
+ private static final long CONTENT_FILE_SCHEMA_BYTES = estimateContentFileSchema();
+
+ private IcebergCacheSizeEstimator() {
+ }
+
+ /** Caffeine callback: the expensive table graph was sized once when {@link CachedTable} was constructed. */
+ static long estimateTableEntry(TableIdentifier key, CachedTable value) {
+ return add(estimateTableIdentifier(key), value.estimatedBytes);
+ }
+
+ static long estimateTable(Table table) {
+ long bytes = add(CACHED_TABLE_SHALLOW_BYTES, JvmSizeUtils.instanceSize(table.getClass()));
+ bytes = add(bytes, JvmSizeUtils.stringSize(table.name()));
+ if (!(table instanceof HasTableOperations)) {
+ bytes = add(bytes, JvmSizeUtils.stringSize(table.location()));
+ return add(bytes, estimateStringMap(table.properties()));
+ }
+
+ TableOperations operations = ((HasTableOperations) table).operations();
+ bytes = add(bytes, JvmSizeUtils.instanceSize(operations.getClass()));
+ return add(bytes, estimateTableMetadata(operations.current()));
+ }
+
+ static long estimatePartitionKey(Key key) {
+ return add(PARTITION_KEY_SHALLOW_BYTES, estimateTableIdentifier(key.id));
+ }
+
+ static long estimatePartitions(List partitions) {
+ long bytes = CACHED_PARTITIONS_SHALLOW_BYTES;
+ // CachedPartitions owns an unmodifiable wrapper around an exact-size ArrayList copy.
+ bytes = add(bytes, JvmSizeUtils.instanceSize(partitions.getClass()));
+ bytes = add(bytes, JvmSizeUtils.arrayListSize(partitions.size()));
+ for (IcebergRawPartition partition : partitions) {
+ bytes = add(bytes, RAW_PARTITION_SHALLOW_BYTES);
+ bytes = add(bytes, JvmSizeUtils.stringSize(partition.name));
+ bytes = add(bytes, estimateStringList(partition.columnNames));
+ bytes = add(bytes, estimateStringList(partition.values));
+ bytes = add(bytes, estimateStringList(partition.transforms));
+ }
+ return bytes;
+ }
+
+ /** Caffeine callback: the partition list was sized once when {@link CachedPartitions} was constructed. */
+ static long estimatePartitionEntry(Key key, CachedPartitions value) {
+ return add(key.estimatedBytes, value.estimatedBytes);
+ }
+
+ static long estimateManifestKey(IcebergManifestEntryKey key) {
+ return add(MANIFEST_KEY_SHALLOW_BYTES, JvmSizeUtils.stringSize(key.getManifestPath()));
+ }
+
+ static long estimateManifestValue(ManifestCacheValue value) {
+ long bytes = MANIFEST_VALUE_SHALLOW_BYTES;
+ bytes = add(bytes, estimateContentFileList(value.getDataFiles()));
+ bytes = add(bytes, estimateContentFileList(value.getDeleteFiles()));
+ return Math.max(bytes, ReflectiveObjectSizeEstimator.estimate(value));
+ }
+
+ /** Caffeine callback: key and manifest payload sizes are precomputed during construction. */
+ static long estimateManifestEntry(IcebergManifestEntryKey key, ManifestCacheValue value) {
+ return add(key.getEstimatedBytes(), value.getEstimatedBytes());
+ }
+
+ private static long estimateTableIdentifier(TableIdentifier identifier) {
+ long bytes = TABLE_IDENTIFIER_SHALLOW_BYTES;
+ Namespace namespace = identifier.namespace();
+ bytes = add(bytes, JvmSizeUtils.instanceSize(namespace.getClass()));
+ String[] levels = namespace.levels();
+ bytes = add(bytes, JvmSizeUtils.objectArraySize(levels.length));
+ for (String level : levels) {
+ bytes = add(bytes, JvmSizeUtils.stringSize(level));
+ }
+ return add(bytes, JvmSizeUtils.stringSize(identifier.name()));
+ }
+
+ private static long estimateTableMetadata(TableMetadata metadata) {
+ long bytes = TABLE_METADATA_SHALLOW_BYTES;
+ bytes = add(bytes, JvmSizeUtils.stringSize(metadata.metadataFileLocation()));
+ bytes = add(bytes, JvmSizeUtils.stringSize(metadata.uuid()));
+ bytes = add(bytes, JvmSizeUtils.stringSize(metadata.location()));
+ bytes = add(bytes, estimateStringMap(metadata.properties()));
+
+ List schemas = metadata.schemas();
+ bytes = add(bytes, estimateListStructure(schemas));
+ bytes = add(bytes, estimateIntegerIndexMap(metadata.schemasById()));
+ for (Schema schema : schemas) {
+ bytes = add(bytes, estimateSchema(schema));
+ }
+
+ List specs = metadata.specs();
+ bytes = add(bytes, estimateListStructure(specs));
+ bytes = add(bytes, estimateIntegerIndexMap(metadata.specsById()));
+ for (PartitionSpec spec : specs) {
+ bytes = add(bytes, estimatePartitionSpec(spec));
+ }
+
+ List sortOrders = metadata.sortOrders();
+ bytes = add(bytes, estimateListStructure(sortOrders));
+ bytes = add(bytes, estimateIntegerIndexMap(metadata.sortOrdersById()));
+ for (SortOrder sortOrder : sortOrders) {
+ bytes = add(bytes, estimateSortOrder(sortOrder));
+ }
+
+ List snapshots = metadata.snapshots();
+ bytes = add(bytes, estimateListStructure(snapshots));
+ bytes = add(bytes, estimateLongIndexMap(snapshots));
+ for (Snapshot snapshot : snapshots) {
+ bytes = add(bytes, estimateSnapshot(snapshot));
+ }
+
+ bytes = add(bytes, estimateHistory(metadata.snapshotLog()));
+ bytes = add(bytes, estimateMetadataLog(metadata.previousFiles()));
+ bytes = add(bytes, estimateSnapshotRefs(metadata.refs()));
+ bytes = add(bytes, estimateStatisticsFiles(metadata.statisticsFiles()));
+ bytes = add(bytes, estimatePartitionStatisticsFiles(metadata.partitionStatisticsFiles()));
+ bytes = add(bytes, estimateMetadataUpdates(metadata.changes()));
+ bytes = add(bytes, estimateShallowList(metadata.encryptionKeys()));
+ // TableMetadata retains a serializable snapshot supplier after the immutable snapshot list is loaded.
+ bytes = add(bytes, JvmSizeUtils.objectArraySize(1));
+ return Math.max(bytes, ReflectiveObjectSizeEstimator.estimate(metadata));
+ }
+
+ private static long estimateSchema(Schema schema) {
+ List columns = schema.columns();
+ long bytes = JvmSizeUtils.instanceSize(schema.getClass());
+ bytes = add(bytes, JvmSizeUtils.instanceSize(schema.asStruct().getClass()));
+ bytes = add(bytes, estimateListStructure(columns));
+ for (Types.NestedField field : columns) {
+ bytes = add(bytes, estimateNestedField(field));
+ }
+ bytes = add(bytes, JvmSizeUtils.objectArraySize(schema.identifierFieldIds().size()));
+ bytes = add(bytes, estimateMapStructure(schema.getAliases()));
+ return add(bytes, estimateSchemaIndexes(columns.size()));
+ }
+
+ private static long estimateNestedField(Types.NestedField field) {
+ long bytes = JvmSizeUtils.instanceSize(field.getClass());
+ bytes = add(bytes, JvmSizeUtils.stringSize(field.name()));
+ bytes = add(bytes, JvmSizeUtils.stringSize(field.doc()));
+ return add(bytes, estimateIcebergType(field.type()));
+ }
+
+ private static long estimateIcebergType(Type type) {
+ long bytes = JvmSizeUtils.instanceSize(type.getClass());
+ if (type.isStructType()) {
+ List fields = type.asStructType().fields();
+ bytes = add(bytes, estimateListStructure(fields));
+ for (Types.NestedField field : fields) {
+ bytes = add(bytes, estimateNestedField(field));
+ }
+ } else if (type.isListType()) {
+ bytes = add(bytes, estimateNestedField(type.asListType().fields().get(0)));
+ } else if (type.isMapType()) {
+ for (Types.NestedField field : type.asMapType().fields()) {
+ bytes = add(bytes, estimateNestedField(field));
+ }
+ }
+ return bytes;
+ }
+
+ private static long estimatePartitionSpec(PartitionSpec spec) {
+ List fields = spec.fields();
+ long bytes = add(JvmSizeUtils.instanceSize(spec.getClass()), JvmSizeUtils.objectArraySize(fields.size()));
+ for (PartitionField field : fields) {
+ bytes = add(bytes, JvmSizeUtils.instanceSize(field.getClass()));
+ bytes = add(bytes, JvmSizeUtils.stringSize(field.name()));
+ bytes = add(bytes, JvmSizeUtils.instanceSize(field.transform().getClass()));
+ }
+ return bytes;
+ }
+
+ private static long estimateSortOrder(SortOrder sortOrder) {
+ List fields = sortOrder.fields();
+ long bytes = add(JvmSizeUtils.instanceSize(sortOrder.getClass()), JvmSizeUtils.objectArraySize(fields.size()));
+ for (SortField field : fields) {
+ bytes = add(bytes, JvmSizeUtils.instanceSize(field.getClass()));
+ bytes = add(bytes, JvmSizeUtils.instanceSize(field.transform().getClass()));
+ }
+ return bytes;
+ }
+
+ private static long estimateSnapshot(Snapshot snapshot) {
+ long bytes = JvmSizeUtils.instanceSize(snapshot.getClass());
+ bytes = add(bytes, estimateBoxed(snapshot.parentId(), LONG_SHALLOW_BYTES));
+ bytes = add(bytes, estimateBoxed(snapshot.schemaId(), INTEGER_SHALLOW_BYTES));
+ bytes = add(bytes, estimateBoxed(snapshot.firstRowId(), LONG_SHALLOW_BYTES));
+ bytes = add(bytes, estimateBoxed(snapshot.addedRows(), LONG_SHALLOW_BYTES));
+ bytes = add(bytes, JvmSizeUtils.stringSize(snapshot.operation()));
+ bytes = add(bytes, JvmSizeUtils.stringSize(snapshot.manifestListLocation()));
+ bytes = add(bytes, JvmSizeUtils.stringSize(snapshot.keyId()));
+ return add(bytes, estimateStringMap(snapshot.summary()));
+ }
+
+ private static long estimateHistory(List history) {
+ long bytes = estimateListStructure(history);
+ for (HistoryEntry entry : history) {
+ bytes = add(bytes, JvmSizeUtils.instanceSize(entry.getClass()));
+ }
+ return bytes;
+ }
+
+ private static long estimateMetadataLog(List entries) {
+ long bytes = estimateListStructure(entries);
+ for (TableMetadata.MetadataLogEntry entry : entries) {
+ bytes = add(bytes, JvmSizeUtils.instanceSize(entry.getClass()));
+ bytes = add(bytes, JvmSizeUtils.stringSize(entry.file()));
+ }
+ return bytes;
+ }
+
+ private static long estimateSnapshotRefs(Map refs) {
+ long bytes = estimateMapStructure(refs);
+ for (Map.Entry entry : refs.entrySet()) {
+ bytes = add(bytes, JvmSizeUtils.stringSize(entry.getKey()));
+ SnapshotRef ref = entry.getValue();
+ bytes = add(bytes, JvmSizeUtils.instanceSize(ref.getClass()));
+ bytes = add(bytes, estimateBoxed(ref.minSnapshotsToKeep(), INTEGER_SHALLOW_BYTES));
+ bytes = add(bytes, estimateBoxed(ref.maxSnapshotAgeMs(), LONG_SHALLOW_BYTES));
+ bytes = add(bytes, estimateBoxed(ref.maxRefAgeMs(), LONG_SHALLOW_BYTES));
+ }
+ return bytes;
+ }
+
+ private static long estimateStatisticsFiles(List files) {
+ long bytes = estimateListStructure(files);
+ for (StatisticsFile file : files) {
+ bytes = add(bytes, JvmSizeUtils.instanceSize(file.getClass()));
+ bytes = add(bytes, JvmSizeUtils.stringSize(file.path()));
+ List blobs = file.blobMetadata();
+ bytes = add(bytes, estimateListStructure(blobs));
+ for (BlobMetadata blob : blobs) {
+ bytes = add(bytes, JvmSizeUtils.instanceSize(blob.getClass()));
+ bytes = add(bytes, JvmSizeUtils.stringSize(blob.type()));
+ bytes = add(bytes, estimateBoxedList(blob.fields(), INTEGER_SHALLOW_BYTES));
+ bytes = add(bytes, estimateStringMap(blob.properties()));
+ }
+ }
+ return bytes;
+ }
+
+ private static long estimatePartitionStatisticsFiles(List files) {
+ long bytes = estimateListStructure(files);
+ for (PartitionStatisticsFile file : files) {
+ bytes = add(bytes, JvmSizeUtils.instanceSize(file.getClass()));
+ bytes = add(bytes, JvmSizeUtils.stringSize(file.path()));
+ }
+ return bytes;
+ }
+
+ private static long estimateShallowList(List> values) {
+ long bytes = estimateListStructure(values);
+ for (Object value : values) {
+ bytes = add(bytes, JvmSizeUtils.instanceSize(value.getClass()));
+ }
+ return bytes;
+ }
+
+ private static long estimateMetadataUpdates(List updates) {
+ long bytes = estimateListStructure(updates);
+ for (MetadataUpdate update : updates) {
+ bytes = add(bytes, JvmSizeUtils.instanceSize(update.getClass()));
+ if (update instanceof MetadataUpdate.SetProperties) {
+ bytes = add(bytes, estimateMapStructure(((MetadataUpdate.SetProperties) update).updated()));
+ } else if (update instanceof MetadataUpdate.AddPartitionSpec) {
+ UnboundPartitionSpec spec = ((MetadataUpdate.AddPartitionSpec) update).spec();
+ bytes = add(bytes, JvmSizeUtils.instanceSize(spec.getClass()));
+ bytes = add(bytes, estimateListStructure(spec.fields()));
+ } else if (update instanceof MetadataUpdate.AddSortOrder) {
+ bytes = add(bytes, JvmSizeUtils.instanceSize(
+ ((MetadataUpdate.AddSortOrder) update).sortOrder().getClass()));
+ }
+ }
+ return bytes;
+ }
+
+ private static long estimateContentFileList(List extends ContentFile>> files) {
+ if (files.isEmpty()) {
+ return 0L;
+ }
+ long bytes = add(estimateListStructure(files), CONTENT_FILE_SCHEMA_BYTES);
+ Set ownedObjects = java.util.Collections.newSetFromMap(new IdentityHashMap<>());
+ for (ContentFile> file : files) {
+ bytes = add(bytes, estimateContentFile(file, ownedObjects));
+ }
+ return bytes;
+ }
+
+ private static long estimateContentFileSchema() {
+ long bytes = JvmSizeUtils.instanceSize(Types.StructType.class);
+ bytes = add(bytes, JvmSizeUtils.arrayListSize(CONTENT_FILE_FIELD_NAMES.length));
+ for (String name : CONTENT_FILE_FIELD_NAMES) {
+ bytes = add(bytes, JvmSizeUtils.instanceSize(Types.NestedField.class));
+ bytes = add(bytes, JvmSizeUtils.stringSize(name));
+ }
+ return bytes;
+ }
+
+ private static long estimateContentFile(ContentFile> file, Set ownedObjects) {
+ long bytes = JvmSizeUtils.instanceSize(file.getClass());
+ if (file instanceof StructLike) {
+ bytes = add(bytes, JvmSizeUtils.intArraySize(((StructLike) file).size()));
+ bytes = add(bytes, LONG_SHALLOW_BYTES);
+ }
+ bytes = add(bytes, estimateOwnedCharSequence(file.path(), ownedObjects));
+ bytes = add(bytes, estimateOwnedString(file.manifestLocation(), ownedObjects));
+ bytes = add(bytes, estimatePartition(file.partition(), ownedObjects));
+ bytes = add(bytes, estimateLongMap(file.columnSizes()));
+ bytes = add(bytes, estimateLongMap(file.valueCounts()));
+ bytes = add(bytes, estimateLongMap(file.nullValueCounts()));
+ bytes = add(bytes, estimateLongMap(file.nanValueCounts()));
+ bytes = add(bytes, estimateByteBufferMap(file.lowerBounds()));
+ bytes = add(bytes, estimateByteBufferMap(file.upperBounds()));
+ ByteBuffer keyMetadata = file.keyMetadata();
+ if (keyMetadata != null) {
+ bytes = add(bytes, JvmSizeUtils.byteArraySize(keyMetadata.remaining()));
+ }
+ List splitOffsets = file.splitOffsets();
+ if (splitOffsets != null) {
+ bytes = add(bytes, JvmSizeUtils.longArraySize(splitOffsets.size()));
+ }
+ List equalityFieldIds = file.equalityFieldIds();
+ if (equalityFieldIds != null) {
+ bytes = add(bytes, JvmSizeUtils.intArraySize(equalityFieldIds.size()));
+ }
+ bytes = add(bytes, estimateBoxed(file.pos(), LONG_SHALLOW_BYTES));
+ bytes = add(bytes, estimateBoxed(file.sortOrderId(), INTEGER_SHALLOW_BYTES));
+ bytes = add(bytes, estimateBoxed(file.dataSequenceNumber(), LONG_SHALLOW_BYTES));
+ bytes = add(bytes, estimateBoxed(file.fileSequenceNumber(), LONG_SHALLOW_BYTES));
+ bytes = add(bytes, estimateBoxed(file.firstRowId(), LONG_SHALLOW_BYTES));
+ if (file instanceof DeleteFile) {
+ DeleteFile deleteFile = (DeleteFile) file;
+ bytes = add(bytes, estimateOwnedString(deleteFile.referencedDataFile(), ownedObjects));
+ bytes = add(bytes, estimateBoxed(deleteFile.contentOffset(), LONG_SHALLOW_BYTES));
+ bytes = add(bytes, estimateBoxed(deleteFile.contentSizeInBytes(), LONG_SHALLOW_BYTES));
+ }
+ return bytes;
+ }
+
+ private static long estimatePartition(StructLike partition, Set ownedObjects) {
+ if (partition == null || !ownedObjects.add(partition)) {
+ return 0L;
+ }
+ long bytes = JvmSizeUtils.instanceSize(partition.getClass());
+ bytes = add(bytes, JvmSizeUtils.objectArraySize(partition.size()));
+ for (int i = 0; i < partition.size(); i++) {
+ bytes = add(bytes, estimateOwnedScalar(partition.get(i, Object.class), ownedObjects));
+ }
+ return bytes;
+ }
+
+ private static long estimateOwnedScalar(Object value, Set ownedObjects) {
+ if (value == null || !ownedObjects.add(value)) {
+ return 0L;
+ }
+ if (value instanceof CharSequence) {
+ return estimateCharSequence((CharSequence) value);
+ }
+ if (value instanceof ByteBuffer) {
+ return estimateByteBuffer((ByteBuffer) value);
+ }
+ return JvmSizeUtils.instanceSize(value.getClass());
+ }
+
+ private static long estimateOwnedCharSequence(CharSequence value, Set ownedObjects) {
+ return value == null || !ownedObjects.add(value) ? 0L : estimateCharSequence(value);
+ }
+
+ private static long estimateOwnedString(String value, Set super String> ownedStrings) {
+ return value == null || !ownedStrings.add(value) ? 0L : JvmSizeUtils.stringSize(value);
+ }
+
+ private static long estimateCharSequence(CharSequence value) {
+ if (value instanceof String) {
+ return JvmSizeUtils.stringSize((String) value);
+ }
+ return add(JvmSizeUtils.instanceSize(value.getClass()), JvmSizeUtils.stringSize(value.toString()));
+ }
+
+ private static long estimateLongMap(Map values) {
+ if (values == null || values.isEmpty()) {
+ return 0L;
+ }
+ return add(estimateMapStructure(values), multiply(values.size(), INTEGER_SHALLOW_BYTES + LONG_SHALLOW_BYTES));
+ }
+
+ private static long estimateByteBufferMap(Map values) {
+ if (values == null || values.isEmpty()) {
+ return 0L;
+ }
+ long bytes = add(estimateMapStructure(values), multiply(values.size(), INTEGER_SHALLOW_BYTES));
+ for (ByteBuffer value : values.values()) {
+ bytes = add(bytes, estimateByteBuffer(value));
+ }
+ return bytes;
+ }
+
+ private static long estimateByteBuffer(ByteBuffer value) {
+ if (value == null) {
+ return 0L;
+ }
+ long bytes = JvmSizeUtils.instanceSize(value.getClass());
+ return value.hasArray() ? add(bytes, JvmSizeUtils.byteArraySize(value.capacity())) : bytes;
+ }
+
+ private static long estimateSchemaIndexes(int fieldCount) {
+ if (fieldCount == 0) {
+ return 0L;
+ }
+ long oneIndex = JvmSizeUtils.instanceSize(HashMap.class);
+ oneIndex = add(oneIndex, JvmSizeUtils.objectArraySize(hashCapacity(fieldCount)));
+ oneIndex = add(oneIndex, multiply(fieldCount, HASH_MAP_NODE_SHALLOW_BYTES));
+ long bytes = multiply(3L, oneIndex);
+ return add(bytes, multiply(2L * fieldCount, INTEGER_SHALLOW_BYTES));
+ }
+
+ private static long estimateIntegerIndexMap(Map values) {
+ return add(estimateMapStructure(values), multiply(values.size(), INTEGER_SHALLOW_BYTES));
+ }
+
+ private static long estimateLongIndexMap(List> values) {
+ if (values.isEmpty()) {
+ return JvmSizeUtils.instanceSize(HashMap.class);
+ }
+ long bytes = JvmSizeUtils.instanceSize(HashMap.class);
+ bytes = add(bytes, JvmSizeUtils.objectArraySize(hashCapacity(values.size())));
+ bytes = add(bytes, multiply(values.size(), HASH_MAP_NODE_SHALLOW_BYTES));
+ return add(bytes, multiply(values.size(), LONG_SHALLOW_BYTES));
+ }
+
+ private static long estimateStringMap(Map values) {
+ if (values == null) {
+ return 0L;
+ }
+ long bytes = estimateMapStructure(values);
+ for (Map.Entry entry : values.entrySet()) {
+ bytes = add(bytes, JvmSizeUtils.stringSize(entry.getKey()));
+ bytes = add(bytes, JvmSizeUtils.stringSize(entry.getValue()));
+ }
+ return bytes;
+ }
+
+ private static long estimateStringList(List values) {
+ if (values == null || values.isEmpty()) {
+ return 0L;
+ }
+ long bytes = estimateListStructure(values);
+ for (String value : values) {
+ bytes = add(bytes, JvmSizeUtils.stringSize(value));
+ }
+ return bytes;
+ }
+
+ private static long estimateBoxedList(List> values, long elementBytes) {
+ if (values == null || values.isEmpty()) {
+ return 0L;
+ }
+ return add(estimateListStructure(values), multiply(values.size(), elementBytes));
+ }
+
+ private static long estimateBoxed(Object value, long bytes) {
+ return value == null ? 0L : bytes;
+ }
+
+ private static long estimateListStructure(List> values) {
+ int capacity = values instanceof ArrayList ? arrayListCapacity(values.size()) : values.size();
+ long bytes = JvmSizeUtils.instanceSize(values.getClass());
+ if (values.getClass().getName().equals("java.util.ImmutableCollections$List12")) {
+ return bytes;
+ }
+ return add(bytes, JvmSizeUtils.objectArraySize(capacity));
+ }
+
+ private static long estimateMapStructure(Map, ?> values) {
+ if (values == null) {
+ return 0L;
+ }
+ long bytes = JvmSizeUtils.instanceSize(values.getClass());
+ if (values.isEmpty()) {
+ return bytes;
+ }
+ if (values instanceof HashMap) {
+ int capacity = hashCapacity(values.size());
+ long nodeBytes = values instanceof LinkedHashMap
+ ? LINKED_HASH_MAP_ENTRY_SHALLOW_BYTES
+ : HASH_MAP_NODE_SHALLOW_BYTES;
+ bytes = add(bytes, JvmSizeUtils.objectArraySize(capacity));
+ return add(bytes, multiply(values.size(), nodeBytes));
+ }
+ bytes = add(bytes, JvmSizeUtils.objectArraySize(saturatedDouble(values.size())));
+ return add(bytes, multiply(values.size(), HASH_MAP_NODE_SHALLOW_BYTES));
+ }
+
+ private static int arrayListCapacity(int size) {
+ if (size == 0) {
+ return 0;
+ }
+ int capacity = 10;
+ while (capacity < size) {
+ int grown = capacity + (capacity >> 1);
+ if (grown < 0) {
+ return Integer.MAX_VALUE;
+ }
+ capacity = grown;
+ }
+ return capacity;
+ }
+
+ private static int hashCapacity(int size) {
+ if (size == 0) {
+ return 0;
+ }
+ long needed = (size * 4L + 2L) / 3L;
+ int capacity = 16;
+ while (capacity < needed && capacity < 1 << 30) {
+ capacity <<= 1;
+ }
+ return capacity;
+ }
+
+ private static int saturatedDouble(int value) {
+ return value > Integer.MAX_VALUE / 2 ? Integer.MAX_VALUE : value * 2;
+ }
+
+ private static long multiply(long left, long right) {
+ return JvmSizeUtils.saturatedMultiply(left, right);
+ }
+
+ private static long classSize(String className) {
+ try {
+ return JvmSizeUtils.instanceSize(Class.forName(className));
+ } catch (ClassNotFoundException e) {
+ throw new IllegalStateException("Required JVM collection class is missing: " + className, e);
+ }
+ }
+
+ private static long add(long left, long right) {
+ return JvmSizeUtils.saturatedAdd(left, right);
+ }
+}
diff --git a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergCatalogProperties.java b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergCatalogProperties.java
index 8cf4db88386e8e..987ba22e3a169a 100644
--- a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergCatalogProperties.java
+++ b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergCatalogProperties.java
@@ -158,6 +158,8 @@ private static void checkMetaCacheProperties(Map properties) {
-1L, IcebergConnector.TABLE_CACHE_TTL_SECOND);
CacheSpec.checkLongProperty(properties.get(IcebergConnector.TABLE_CACHE_CAPACITY),
0L, IcebergConnector.TABLE_CACHE_CAPACITY);
+ CacheSpec.checkDataSizeProperty(properties.get(IcebergConnector.TABLE_CACHE_MAX_WEIGHT),
+ IcebergConnector.TABLE_CACHE_MAX_WEIGHT);
CacheSpec.checkBooleanProperty(properties.get(IcebergConnector.MANIFEST_CACHE_ENABLE),
IcebergConnector.MANIFEST_CACHE_ENABLE);
@@ -165,6 +167,8 @@ private static void checkMetaCacheProperties(Map properties) {
-1L, IcebergConnector.MANIFEST_CACHE_TTL_SECOND);
CacheSpec.checkLongProperty(properties.get(IcebergConnector.MANIFEST_CACHE_CAPACITY),
0L, IcebergConnector.MANIFEST_CACHE_CAPACITY);
+ CacheSpec.checkDataSizeProperty(properties.get(IcebergConnector.MANIFEST_CACHE_MAX_WEIGHT),
+ IcebergConnector.MANIFEST_CACHE_MAX_WEIGHT);
}
/** The metastore backend, lower-cased; {@code null} when the catalog does not name one. */
diff --git a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergConnector.java b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergConnector.java
index 47df2d3f9c6f06..79c580705ec858 100644
--- a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergConnector.java
+++ b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergConnector.java
@@ -17,6 +17,7 @@
package org.apache.doris.connector.iceberg;
+import org.apache.doris.connector.cache.CacheSpec;
import org.apache.doris.connector.cache.ConnectorMetadataCache;
import org.apache.doris.connector.metastore.HmsMetaStoreProperties;
import org.apache.doris.connector.metastore.iceberg.jdbc.IcebergJdbcMetaStoreProperties;
@@ -141,9 +142,11 @@ public class IcebergConnector implements Connector {
static final String TABLE_CACHE_ENABLE = "meta.cache.iceberg.table.enable";
static final String TABLE_CACHE_TTL_SECOND = "meta.cache.iceberg.table.ttl-second";
static final String TABLE_CACHE_CAPACITY = "meta.cache.iceberg.table.capacity";
+ static final String TABLE_CACHE_MAX_WEIGHT = "meta.cache.iceberg.table.max-weight";
static final String MANIFEST_CACHE_ENABLE = "meta.cache.iceberg.manifest.enable";
static final String MANIFEST_CACHE_TTL_SECOND = "meta.cache.iceberg.manifest.ttl-second";
static final String MANIFEST_CACHE_CAPACITY = "meta.cache.iceberg.manifest.capacity";
+ static final String MANIFEST_CACHE_MAX_WEIGHT = "meta.cache.iceberg.manifest.max-weight";
static final long DEFAULT_TABLE_CACHE_TTL_SECOND = 86400L;
static final int DEFAULT_TABLE_CACHE_CAPACITY = 1000;
@@ -224,7 +227,7 @@ public class IcebergConnector implements Connector {
listPartitionsViewCache;
// Manifest content cache — pure metadata, default-off (meta.cache.iceberg.manifest.enable), and consumed
// ONLY after a per-user resolveTable(ForRead) -- exempt: no read path without a per-user load.
- private final IcebergManifestCache manifestCache = new IcebergManifestCache();
+ private final IcebergManifestCache manifestCache;
// Lazily-built plugin-side Kerberos authenticator (single-owner auth; see TcclPinningConnectorContext).
// null for a non-Kerberos catalog. Its doAs acts on the PLUGIN's UserGroupInformation copy — the one the
@@ -248,6 +251,8 @@ public IcebergConnector(Map properties, ConnectorContext context
// authenticator never logs in — so without this the DDL/read hits secured HDFS as SIMPLE auth.
this.context = new TcclPinningConnectorContext(context, getClass().getClassLoader(),
this::pluginAuthenticator);
+ CacheSpec tableCacheSpec = resolveTableCacheSpec(this.properties);
+ this.manifestCache = new IcebergManifestCache(this.properties);
// Authorization-sensitive projection (snapshotId/schemaId). Under iceberg.rest.session=user the value is
// per-user AUTHORIZED metadata that a "can-list-cannot-load" principal must not see. beginQuerySnapshot
// reads this cache WITHOUT a preceding per-user loadTable, so a shared (table-keyed, no user dimension)
@@ -270,8 +275,7 @@ public IcebergConnector(Map properties, ConnectorContext context
this.tableCache = (isUserSessionEnabled()
|| IcebergScanPlanProvider.restVendedCredentialsEnabled(this.properties))
? null
- : new IcebergTableCache(
- resolveTableCacheTtlSecond(this.properties), DEFAULT_TABLE_CACHE_CAPACITY);
+ : new IcebergTableCache(tableCacheSpec);
// PERF-02: partition-view cache. Authorization-sensitive projection: a shared (table+snapshot-keyed, no
// user dimension) hit would disclose one user's partition list. Its readers are all downstream of a
// per-user resolveTableForRead today (so a hit cannot precede authz), but that safety rests entirely on
@@ -280,8 +284,7 @@ public IcebergConnector(Map properties, ConnectorContext context
// otherwise (single static identity). Readers already tolerate a null cache (loadRawPartitions).
this.partitionCache = isUserSessionEnabled()
? null
- : new IcebergPartitionCache(
- resolveTableCacheTtlSecond(this.properties), DEFAULT_TABLE_CACHE_CAPACITY);
+ : new IcebergPartitionCache(tableCacheSpec);
// PERF-03: inferred-file-format cache. Same authorization-sensitive treatment as partitionCache (disabled
// under session=user, kept otherwise); readers already tolerate a null cache (resolveFileFormatName).
this.formatCache = isUserSessionEnabled()
@@ -330,6 +333,24 @@ static long resolveTableCacheTtlSecond(Map properties) {
}
}
+ private static CacheSpec resolveTableCacheSpec(Map properties) {
+ CacheSpec parsed = CacheSpec.fromProperties(
+ properties,
+ "iceberg",
+ "table",
+ CacheSpec.of(true, DEFAULT_TABLE_CACHE_TTL_SECOND, DEFAULT_TABLE_CACHE_CAPACITY));
+ if (parsed.getTtlSecond() > 0L) {
+ return parsed;
+ }
+ if (parsed.isWeightBounded()) {
+ return CacheSpec.ofWeight(
+ parsed.isEnable(), CacheSpec.CACHE_TTL_DISABLE_CACHE,
+ parsed.getCapacity(), parsed.getMaxWeight().getAsLong());
+ }
+ return CacheSpec.of(
+ parsed.isEnable(), CacheSpec.CACHE_TTL_DISABLE_CACHE, parsed.getCapacity());
+ }
+
@Override
public ConnectorMetadata getMetadata(ConnectorSession session) {
return new IcebergConnectorMetadata(newCatalogBackedOps(session), catalogProps, context,
diff --git a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergManifestCache.java b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergManifestCache.java
index f3c46732fb2fc1..afe6bedf1ab166 100644
--- a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergManifestCache.java
+++ b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergManifestCache.java
@@ -19,6 +19,7 @@
import org.apache.doris.connector.cache.CacheSpec;
import org.apache.doris.connector.cache.MetaCacheEntry;
+import org.apache.doris.connector.cache.MetaCacheEntryStats;
import org.apache.iceberg.DataFile;
import org.apache.iceberg.DeleteFile;
@@ -100,12 +101,26 @@ private static final class ScanStats {
this(maxSize, DEFAULT_STATS_TTL_SECONDS, System::nanoTime);
}
+ IcebergManifestCache(Map properties) {
+ this(CacheSpec.fromProperties(
+ properties,
+ "iceberg",
+ "manifest",
+ CacheSpec.of(false, CacheSpec.CACHE_NO_TTL, DEFAULT_MANIFEST_CACHE_CAPACITY)),
+ DEFAULT_STATS_TTL_SECONDS,
+ System::nanoTime);
+ }
+
/** Visible for testing: injectable stats TTL + clock so the leak sweep is deterministic without sleeping. */
IcebergManifestCache(int maxSize, long statsTtlSeconds, LongSupplier nanoClock) {
// Always enabled, no expiry, capacity-bounded (CACHE_NO_TTL == -1 means "no expiration", enabled).
- CacheSpec spec = CacheSpec.of(true, CacheSpec.CACHE_NO_TTL, Math.max(1, maxSize));
+ this(CacheSpec.of(true, CacheSpec.CACHE_NO_TTL, Math.max(1, maxSize)), statsTtlSeconds, nanoClock);
+ }
+
+ private IcebergManifestCache(CacheSpec spec, long statsTtlSeconds, LongSupplier nanoClock) {
this.entry = new MetaCacheEntry<>("iceberg-manifest", null, spec,
- ForkJoinPool.commonPool(), false, true, 0L, true);
+ ForkJoinPool.commonPool(), false, true, 0L, true,
+ IcebergCacheSizeEstimator::estimateManifestEntry);
this.statsTtlNanos = TimeUnit.SECONDS.toNanos(Math.max(1L, statsTtlSeconds));
this.nanoClock = nanoClock;
}
@@ -232,4 +247,8 @@ int size() {
entry.forEach((key, value) -> count[0]++);
return count[0];
}
+
+ MetaCacheEntryStats stats() {
+ return entry.stats();
+ }
}
diff --git a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergManifestEntryKey.java b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergManifestEntryKey.java
index 90552bef6d5d3a..c9901d387d0403 100644
--- a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergManifestEntryKey.java
+++ b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergManifestEntryKey.java
@@ -35,10 +35,12 @@
public class IcebergManifestEntryKey {
private final String manifestPath;
private final ManifestContent content;
+ private final long estimatedBytes;
public IcebergManifestEntryKey(String manifestPath, ManifestContent content) {
this.manifestPath = Objects.requireNonNull(manifestPath, "manifestPath can not be null");
this.content = Objects.requireNonNull(content, "content can not be null");
+ this.estimatedBytes = IcebergCacheSizeEstimator.estimateManifestKey(this);
}
public static IcebergManifestEntryKey of(ManifestFile manifest) {
@@ -53,6 +55,10 @@ public ManifestContent getContent() {
return content;
}
+ long getEstimatedBytes() {
+ return estimatedBytes;
+ }
+
@Override
public boolean equals(Object o) {
if (this == o) {
diff --git a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergPartitionCache.java b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergPartitionCache.java
index a85e2187043e7c..30a3834f8721b1 100644
--- a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergPartitionCache.java
+++ b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergPartitionCache.java
@@ -19,11 +19,14 @@
import org.apache.doris.connector.cache.CacheSpec;
import org.apache.doris.connector.cache.MetaCacheEntry;
+import org.apache.doris.connector.cache.MetaCacheEntryStats;
import org.apache.doris.connector.iceberg.IcebergPartitionUtils.IcebergRawPartition;
import org.apache.iceberg.catalog.Namespace;
import org.apache.iceberg.catalog.TableIdentifier;
+import java.util.ArrayList;
+import java.util.Collections;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.ForkJoinPool;
@@ -61,10 +64,12 @@ final class IcebergPartitionCache {
static final class Key {
final TableIdentifier id;
final long snapshotId;
+ final long estimatedBytes;
Key(TableIdentifier id, long snapshotId) {
this.id = id;
this.snapshotId = snapshotId;
+ this.estimatedBytes = IcebergCacheSizeEstimator.estimatePartitionKey(this);
}
@Override
@@ -85,13 +90,27 @@ public int hashCode() {
}
}
- private final MetaCacheEntry> entry;
+ static final class CachedPartitions {
+ final List partitions;
+ final long estimatedBytes;
+
+ CachedPartitions(List partitions) {
+ this.partitions = Collections.unmodifiableList(new ArrayList<>(partitions));
+ this.estimatedBytes = IcebergCacheSizeEstimator.estimatePartitions(this.partitions);
+ }
+ }
+
+ private final MetaCacheEntry entry;
IcebergPartitionCache(long ttlSeconds, int maxSize) {
// "<= 0 disables" connector TTL contract, folded to CacheSpec's disable sentinel (CacheSpec.ofConnectorTtl).
- CacheSpec spec = CacheSpec.ofConnectorTtl(ttlSeconds, maxSize);
+ this(CacheSpec.ofConnectorTtl(ttlSeconds, maxSize));
+ }
+
+ IcebergPartitionCache(CacheSpec spec) {
this.entry = new MetaCacheEntry<>("iceberg-partition", null, spec,
- ForkJoinPool.commonPool(), false, true, 0L, true);
+ ForkJoinPool.commonPool(), false, true, 0L, true,
+ IcebergCacheSizeEstimator::estimatePartitionEntry);
}
/** Caching is on only when the TTL is positive; ttl-second <= 0 means "always scan live". */
@@ -105,7 +124,7 @@ boolean isEnabled() {
* loader runs OUTSIDE Caffeine's compute lock (single-flight per key) and its exception propagates unwrapped.
*/
List getOrLoad(Key key, Supplier> loader) {
- return entry.get(key, ignored -> loader.get());
+ return entry.get(key, ignored -> new CachedPartitions(loader.get())).partitions;
}
/** Drops every cached snapshot entry for one table so the next read scans live (REFRESH TABLE). */
@@ -135,4 +154,8 @@ int size() {
long loadCountForTest() {
return entry.stats().getLoadSuccessCount();
}
+
+ MetaCacheEntryStats stats() {
+ return entry.stats();
+ }
}
diff --git a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergPartitionUtils.java b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergPartitionUtils.java
index a5ff4562e915cd..2307cf7de7d422 100644
--- a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergPartitionUtils.java
+++ b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergPartitionUtils.java
@@ -971,15 +971,15 @@ static long latestSnapshotId(String name, Map> mergeMap,
/** One PARTITIONS-metadata-table row reduced to what the MTMV partition view needs (port of IcebergPartition). */
static final class IcebergRawPartition {
- private final String name;
+ final String name;
// Partition-field SOURCE column names (lowercased), parallel to {@link #values}, so listPartitions can
// build a value map keyed by the generic partition-column remote name (see IcebergConnectorMetadata
// buildTableSchema's "partition_columns" derivation).
- private final List columnNames;
- private final List values;
- private final List transforms;
- private final long lastUpdateTime;
- private final long lastSnapshotId;
+ final List columnNames;
+ final List values;
+ final List transforms;
+ final long lastUpdateTime;
+ final long lastSnapshotId;
IcebergRawPartition(String name, List columnNames, List values, List transforms,
long lastUpdateTime, long lastSnapshotId) {
diff --git a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergTableCache.java b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergTableCache.java
index 426b706cf11f90..e7bac2d6f46fcf 100644
--- a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergTableCache.java
+++ b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergTableCache.java
@@ -19,6 +19,7 @@
import org.apache.doris.connector.cache.CacheSpec;
import org.apache.doris.connector.cache.MetaCacheEntry;
+import org.apache.doris.connector.cache.MetaCacheEntryStats;
import org.apache.iceberg.Table;
import org.apache.iceberg.catalog.Namespace;
@@ -58,13 +59,27 @@
*/
final class IcebergTableCache {
- private final MetaCacheEntry entry;
+ static final class CachedTable {
+ final Table table;
+ final long estimatedBytes;
+
+ CachedTable(Table table) {
+ this.table = table;
+ this.estimatedBytes = IcebergCacheSizeEstimator.estimateTable(table);
+ }
+ }
+
+ private final MetaCacheEntry entry;
IcebergTableCache(long ttlSeconds, int maxSize) {
// "<= 0 disables" connector TTL contract, folded to CacheSpec's disable sentinel (CacheSpec.ofConnectorTtl).
- CacheSpec spec = CacheSpec.ofConnectorTtl(ttlSeconds, maxSize);
+ this(CacheSpec.ofConnectorTtl(ttlSeconds, maxSize));
+ }
+
+ IcebergTableCache(CacheSpec spec) {
this.entry = new MetaCacheEntry<>("iceberg-table", null, spec,
- ForkJoinPool.commonPool(), false, true, 0L, true);
+ ForkJoinPool.commonPool(), false, true, 0L, true,
+ IcebergCacheSizeEstimator::estimateTableEntry);
}
/** Caching is on only when the TTL is positive; ttl-second <= 0 means "always read live". */
@@ -80,7 +95,7 @@ boolean isEnabled() {
* propagates unwrapped.
*/
Table getOrLoad(TableIdentifier identifier, Supplier loader) {
- return entry.get(identifier, ignored -> loader.get());
+ return entry.get(identifier, ignored -> new CachedTable(loader.get())).table;
}
/** Drops the cached entry for one table so the next read goes live (REFRESH TABLE). */
@@ -110,4 +125,8 @@ int size() {
entry.forEach((key, value) -> count[0]++);
return count[0];
}
+
+ MetaCacheEntryStats stats() {
+ return entry.stats();
+ }
}
diff --git a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/ManifestCacheValue.java b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/ManifestCacheValue.java
index b9b9da68dbd3c5..06e68ad769878e 100644
--- a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/ManifestCacheValue.java
+++ b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/ManifestCacheValue.java
@@ -33,10 +33,12 @@
public class ManifestCacheValue {
private final List dataFiles;
private final List deleteFiles;
+ private final long estimatedBytes;
private ManifestCacheValue(List dataFiles, List deleteFiles) {
this.dataFiles = dataFiles == null ? Collections.emptyList() : dataFiles;
this.deleteFiles = deleteFiles == null ? Collections.emptyList() : deleteFiles;
+ this.estimatedBytes = IcebergCacheSizeEstimator.estimateManifestValue(this);
}
public static ManifestCacheValue forDataFiles(List dataFiles) {
@@ -54,4 +56,8 @@ public List getDataFiles() {
public List getDeleteFiles() {
return deleteFiles;
}
+
+ long getEstimatedBytes() {
+ return estimatedBytes;
+ }
}
diff --git a/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonCatalogProperties.java b/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonCatalogProperties.java
index 2a16f1e81f67d0..6909ab7a6a3dca 100644
--- a/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonCatalogProperties.java
+++ b/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonCatalogProperties.java
@@ -165,6 +165,8 @@ private static void checkMetaCacheProperties(Map properties) {
-1L, PaimonConnector.TABLE_CACHE_TTL_SECOND);
CacheSpec.checkLongProperty(properties.get(PaimonConnector.TABLE_CACHE_CAPACITY),
0L, PaimonConnector.TABLE_CACHE_CAPACITY);
+ CacheSpec.checkDataSizeProperty(properties.get(PaimonConnector.PARTITION_VIEW_CACHE_MAX_WEIGHT),
+ PaimonConnector.PARTITION_VIEW_CACHE_MAX_WEIGHT);
}
// R2: warn (do not reject, do not strip) when a CREATE/ALTER CATALOG carries the now-dead paimon
diff --git a/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonConnector.java b/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonConnector.java
index 5a620f1a222448..05a30ed9baae03 100644
--- a/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonConnector.java
+++ b/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonConnector.java
@@ -17,6 +17,7 @@
package org.apache.doris.connector.paimon;
+import org.apache.doris.connector.cache.CacheSpec;
import org.apache.doris.connector.cache.ConnectorMetadataCache;
import org.apache.doris.connector.metastore.HmsMetaStoreProperties;
import org.apache.doris.connector.metastore.paimon.jdbc.PaimonJdbcMetaStoreProperties;
@@ -101,6 +102,8 @@ public class PaimonConnector implements Connector {
// still validated at CREATE/ALTER for legacy parity (reject non-boolean / out-of-range garbage).
static final String TABLE_CACHE_ENABLE = "meta.cache.paimon.table.enable";
static final String TABLE_CACHE_CAPACITY = "meta.cache.paimon.table.capacity";
+ static final String PARTITION_VIEW_CACHE_MAX_WEIGHT =
+ CacheSpec.metaCacheMaxWeightKey("paimon", "partition_view");
// Legacy default = Config.external_cache_expire_time_seconds_after_access (24h); the connector is isolated
// from fe-core Config, so the legacy default is mirrored here (an explicit ttl-second always overrides it).
static final long DEFAULT_TABLE_CACHE_TTL_SECOND = 86400L;
@@ -165,9 +168,10 @@ public PaimonConnector(Map properties, ConnectorContext context)
this::pluginAuthenticator);
this.latestSnapshotCache =
new PaimonLatestSnapshotCache(resolveTableCacheTtlSecond(properties), DEFAULT_TABLE_CACHE_CAPACITY);
- // Reads its own meta.cache.paimon.partition_view.(enable|ttl-second|capacity) from the catalog
- // properties via the framework's CacheSpec (default ON / 24h / 1000).
- this.partitionViewCache = new ConnectorMetadataCache<>("paimon", "partition_view", properties);
+ // Reads its own meta.cache.paimon.partition_view.(enable|ttl-second|capacity|max-weight) from the
+ // catalog properties via the framework's CacheSpec (default ON / 24h / 1000).
+ this.partitionViewCache = new ConnectorMetadataCache<>("paimon", "partition_view", properties,
+ PaimonPartitionViewSizeEstimator::estimateEntry);
}
/**
diff --git a/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonConnectorMetadata.java b/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonConnectorMetadata.java
index 7f124db18c55fb..2e712dd61b7db2 100644
--- a/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonConnectorMetadata.java
+++ b/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonConnectorMetadata.java
@@ -1285,7 +1285,12 @@ private List cachedPartitions(PaimonTableHandle paimonHa
return collectPartitions(paimonHandle);
}
ConnectorTableKey key = partitionViewCacheKey(paimonHandle);
- return partitionViewCache.get(key, () -> collectPartitions(paimonHandle));
+ return partitionViewCache.get(key, () -> {
+ List partitions = collectPartitions(paimonHandle);
+ return partitionViewCache.isWeightBounded()
+ ? new PaimonPartitionView(key, partitions)
+ : partitions;
+ });
}
/**
diff --git a/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonPartitionView.java b/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonPartitionView.java
new file mode 100644
index 00000000000000..3ae03c6b88d77c
--- /dev/null
+++ b/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonPartitionView.java
@@ -0,0 +1,52 @@
+// 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.connector.paimon;
+
+import org.apache.doris.connector.cache.ConnectorTableKey;
+import org.apache.doris.connector.spi.ConnectorPartitionInfo;
+
+import java.util.AbstractList;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.RandomAccess;
+
+/** Immutable Paimon partition-view cache value with its entry weight computed once at construction. */
+final class PaimonPartitionView extends AbstractList implements RandomAccess {
+ private final List partitions;
+ private final long estimatedBytes;
+
+ PaimonPartitionView(ConnectorTableKey key, List partitions) {
+ this.partitions = Collections.unmodifiableList(new ArrayList<>(partitions));
+ this.estimatedBytes = PaimonPartitionViewSizeEstimator.estimateEntryOnConstruction(key, this);
+ }
+
+ @Override
+ public ConnectorPartitionInfo get(int index) {
+ return partitions.get(index);
+ }
+
+ @Override
+ public int size() {
+ return partitions.size();
+ }
+
+ long getEstimatedBytes() {
+ return estimatedBytes;
+ }
+}
diff --git a/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonPartitionViewSizeEstimator.java b/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonPartitionViewSizeEstimator.java
new file mode 100644
index 00000000000000..bc83996fd54199
--- /dev/null
+++ b/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonPartitionViewSizeEstimator.java
@@ -0,0 +1,113 @@
+// 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.connector.paimon;
+
+import org.apache.doris.connector.cache.ConnectorTableKey;
+import org.apache.doris.connector.cache.JvmSizeUtils;
+import org.apache.doris.connector.spi.ConnectorPartitionInfo;
+
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.List;
+
+/** Type-specific retained-heap estimator for the Paimon derived partition-view cache. */
+final class PaimonPartitionViewSizeEstimator {
+ private static final long KEY_SHALLOW_BYTES = JvmSizeUtils.instanceSize(ConnectorTableKey.class);
+ private static final long VIEW_SHALLOW_BYTES = JvmSizeUtils.instanceSize(PaimonPartitionView.class);
+ private static final long PARTITION_SHALLOW_BYTES = JvmSizeUtils.instanceSize(ConnectorPartitionInfo.class);
+ private static final long UNMODIFIABLE_LIST_SHALLOW_BYTES = JvmSizeUtils.instanceSize(
+ Collections.unmodifiableList(Collections.emptyList()).getClass());
+ private static final long UNMODIFIABLE_MAP_SHALLOW_BYTES = JvmSizeUtils.instanceSize(
+ Collections.unmodifiableMap(Collections.emptyMap()).getClass());
+ private static final long LINKED_HASH_MAP_SHALLOW_BYTES = JvmSizeUtils.instanceSize(LinkedHashMap.class);
+ private static final long LINKED_HASH_MAP_ENTRY_SHALLOW_BYTES = classSize("java.util.LinkedHashMap$Entry");
+
+ private PaimonPartitionViewSizeEstimator() {
+ }
+
+ /** Caffeine callback: the complete key/value weight was computed when the immutable view was built. */
+ static long estimateEntry(ConnectorTableKey key, List value) {
+ return ((PaimonPartitionView) value).getEstimatedBytes();
+ }
+
+ static long estimateEntryOnConstruction(ConnectorTableKey key, PaimonPartitionView value) {
+ long bytes = KEY_SHALLOW_BYTES;
+ bytes = add(bytes, JvmSizeUtils.stringSize(key.getDb()));
+ bytes = add(bytes, JvmSizeUtils.stringSize(key.getTable()));
+ bytes = add(bytes, VIEW_SHALLOW_BYTES);
+ bytes = add(bytes, UNMODIFIABLE_LIST_SHALLOW_BYTES);
+ bytes = add(bytes, JvmSizeUtils.arrayListSize(value.size()));
+ for (ConnectorPartitionInfo partition : value) {
+ bytes = add(bytes, estimatePartition(partition));
+ }
+ return bytes;
+ }
+
+ private static long estimatePartition(ConnectorPartitionInfo partition) {
+ long bytes = PARTITION_SHALLOW_BYTES;
+ bytes = add(bytes, JvmSizeUtils.stringSize(partition.getPartitionName()));
+
+ int valueCount = partition.getPartitionValues().size();
+ if (valueCount > 0) {
+ bytes = add(bytes, UNMODIFIABLE_MAP_SHALLOW_BYTES);
+ bytes = add(bytes, LINKED_HASH_MAP_SHALLOW_BYTES);
+ bytes = add(bytes, JvmSizeUtils.objectArraySize(hashCapacity(valueCount)));
+ bytes = add(bytes, multiply(valueCount, LINKED_HASH_MAP_ENTRY_SHALLOW_BYTES));
+ }
+
+ List orderedValues = partition.getOrderedPartitionValues();
+ bytes = add(bytes, estimateCopiedList(orderedValues));
+ for (String value : orderedValues) {
+ // The same rendered String is retained by both partitionValues and orderedPartitionValues.
+ bytes = add(bytes, JvmSizeUtils.stringSize(value));
+ }
+ return add(bytes, estimateCopiedList(partition.getPartitionValueNullFlags()));
+ }
+
+ private static long estimateCopiedList(List> values) {
+ if (values.isEmpty()) {
+ return 0L;
+ }
+ return add(JvmSizeUtils.instanceSize(values.getClass()), JvmSizeUtils.arrayListSize(values.size()));
+ }
+
+ private static int hashCapacity(int size) {
+ long needed = (size * 4L + 2L) / 3L;
+ int capacity = 16;
+ while (capacity < needed && capacity < 1 << 30) {
+ capacity <<= 1;
+ }
+ return capacity;
+ }
+
+ private static long classSize(String className) {
+ try {
+ return JvmSizeUtils.instanceSize(Class.forName(className));
+ } catch (ClassNotFoundException e) {
+ throw new IllegalStateException("Required JVM collection class is missing: " + className, e);
+ }
+ }
+
+ private static long multiply(long left, long right) {
+ return JvmSizeUtils.saturatedMultiply(left, right);
+ }
+
+ private static long add(long left, long right) {
+ return JvmSizeUtils.saturatedAdd(left, right);
+ }
+}
diff --git a/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonConnectorValidatePropertiesTest.java b/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonConnectorValidatePropertiesTest.java
index 516a027870b883..1c167c47b5f4ef 100644
--- a/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonConnectorValidatePropertiesTest.java
+++ b/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonConnectorValidatePropertiesTest.java
@@ -114,6 +114,16 @@ public void acceptsValidMetaCacheKnobs() {
"meta.cache.paimon.table.ttl-second", "0")));
}
+ @Test
+ public void validatesPartitionViewMaxWeight() {
+ Assertions.assertDoesNotThrow(() -> validate(props(
+ "paimon.catalog.type", "filesystem", "warehouse", "/wh",
+ PaimonConnector.PARTITION_VIEW_CACHE_MAX_WEIGHT, "256MB")));
+ Assertions.assertThrows(IllegalArgumentException.class, () -> validate(props(
+ "paimon.catalog.type", "filesystem", "warehouse", "/wh",
+ PaimonConnector.PARTITION_VIEW_CACHE_MAX_WEIGHT, "not-a-size")));
+ }
+
@Test
public void requiresWarehouseForRest() {
// Legacy parity: AbstractPaimonProperties requires warehouse and PaimonRestMetaStoreProperties
diff --git a/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonPartitionViewSizeEstimatorTest.java b/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonPartitionViewSizeEstimatorTest.java
new file mode 100644
index 00000000000000..75ddcecc1bd575
--- /dev/null
+++ b/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonPartitionViewSizeEstimatorTest.java
@@ -0,0 +1,65 @@
+// 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.connector.paimon;
+
+import org.apache.doris.connector.cache.ConnectorTableKey;
+import org.apache.doris.connector.spi.ConnectorPartitionInfo;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+
+public class PaimonPartitionViewSizeEstimatorTest {
+
+ @Test
+ public void weightIsPrecomputedAndGrowsWithOwnedPartitions() {
+ ConnectorTableKey key = new ConnectorTableKey("db", "table", 10L, -1L);
+ PaimonPartitionView one = new PaimonPartitionView(key,
+ Collections.singletonList(partition(1)));
+ PaimonPartitionView two = new PaimonPartitionView(key,
+ Arrays.asList(partition(1), partition(2)));
+
+ Assertions.assertEquals(one.getEstimatedBytes(),
+ PaimonPartitionViewSizeEstimator.estimateEntry(key, one));
+ Assertions.assertTrue(two.getEstimatedBytes() > one.getEstimatedBytes());
+ Assertions.assertThrows(UnsupportedOperationException.class, () -> one.add(partition(3)));
+ }
+
+ private static ConnectorPartitionInfo partition(int bucket) {
+ Map values = new LinkedHashMap<>();
+ values.put("dt", "2026-08-07");
+ values.put("bucket", Integer.toString(bucket));
+ List orderedValues = new ArrayList<>(values.values());
+ return new ConnectorPartitionInfo(
+ "dt=2026-08-07/bucket=" + bucket,
+ values,
+ Collections.emptyMap(),
+ 10_000L + bucket,
+ 128L * 1024L * 1024L,
+ 1_786_048_000_000L + bucket,
+ 4L,
+ orderedValues,
+ Arrays.asList(false, false));
+ }
+}
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 2b3abfce4e9a76..eab8103d575daf 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;
@@ -70,22 +71,49 @@ public CacheFactory(
// Build a loading cache, without executor, it will use fork-join pool for refresh
public LoadingCache buildCache(CacheLoader cacheLoader) {
- Caffeine builder = buildWithParams();
+ Caffeine builder = buildSizeBoundedWithParams();
return builder.build(cacheLoader);
}
// Build a loading cache, with executor, it will use given executor for refresh
public LoadingCache buildCache(CacheLoader cacheLoader,
ExecutorService executor) {
- Caffeine builder = buildWithParams();
+ Caffeine builder = buildSizeBoundedWithParams();
builder.executor(executor);
return builder.build(cacheLoader);
}
+ /**
+ * Build a loading cache bounded by weight instead of entry count.
+ */
+ public LoadingCache buildCacheWithWeight(CacheLoader cacheLoader,
+ ExecutorService executor, long maxWeight, Weigher weigher) {
+ Caffeine builder = buildWithParams()
+ .maximumWeight(maxWeight)
+ .weigher(weigher);
+ builder.executor(executor);
+ return builder.build(cacheLoader);
+ }
+
+ /**
+ * Build a loading cache bounded by weight with a synchronous removal listener.
+ */
+ public LoadingCache buildCacheWithWeightAndSyncRemovalListener(CacheLoader cacheLoader,
+ long maxWeight, Weigher weigher, RemovalListener removalListener) {
+ Caffeine builder = buildWithParams()
+ .maximumWeight(maxWeight)
+ .weigher(weigher);
+ if (removalListener != null) {
+ builder.removalListener(removalListener);
+ }
+ builder.executor(Runnable::run); // Sync execution to avoid thread pool deadlock
+ return builder.build(cacheLoader);
+ }
+
// Build cache with sync removal listener to prevent deadlock when listener calls invalidateAll()
public LoadingCache buildCacheWithSyncRemovalListener(CacheLoader cacheLoader,
RemovalListener removalListener) {
- Caffeine builder = buildWithParams();
+ Caffeine builder = buildSizeBoundedWithParams();
if (removalListener != null) {
builder.removalListener(removalListener);
}
@@ -96,7 +124,7 @@ public LoadingCache buildCacheWithSyncRemovalListener(CacheLoader AsyncLoadingCache buildAsyncCache(AsyncCacheLoader cacheLoader,
ExecutorService executor) {
- Caffeine builder = buildWithParams();
+ Caffeine builder = buildSizeBoundedWithParams();
builder.executor(executor);
return builder.buildAsync(cacheLoader);
}
@@ -104,8 +132,6 @@ public AsyncLoadingCache buildAsyncCache(AsyncCacheLoader cac
@NotNull
private Caffeine buildWithParams() {
Caffeine builder = Caffeine.newBuilder();
- builder.maximumSize(maxSize);
-
if (expireAfterAccessSec.isPresent()) {
builder.expireAfterAccess(Duration.ofSeconds(expireAfterAccessSec.getAsLong()));
}
@@ -122,4 +148,9 @@ private Caffeine buildWithParams() {
}
return builder;
}
+
+ @NotNull
+ private Caffeine buildSizeBoundedWithParams() {
+ return buildWithParams().maximumSize(maxSize);
+ }
}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/common/util/ParseUtil.java b/fe/fe-core/src/main/java/org/apache/doris/common/util/ParseUtil.java
index 135a8eeac0a2df..fa3494f099a4c9 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/common/util/ParseUtil.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/common/util/ParseUtil.java
@@ -70,6 +70,36 @@ public static long analyzeDataVolume(String dataVolumnStr) throws AnalysisExcept
return dataVolumn;
}
+ /**
+ * Parse a data volume while allowing zero and rejecting arithmetic overflow.
+ */
+ public static long analyzeDataVolumeAllowZero(String dataVolumnStr) throws AnalysisException {
+ long dataVolumn;
+ Matcher m = dataVolumnPattern.matcher(dataVolumnStr);
+ if (!m.matches()) {
+ throw new AnalysisException("invalid data volume expression:" + dataVolumnStr);
+ }
+ try {
+ dataVolumn = Long.parseLong(m.group(1));
+ } catch (NumberFormatException nfe) {
+ throw new AnalysisException("invalid data volume:" + m.group(1));
+ }
+
+ String unit = "B";
+ String tmpUnit = m.group(2);
+ if (!Strings.isNullOrEmpty(tmpUnit)) {
+ unit = tmpUnit.toUpperCase();
+ }
+ if (!validDataVolumnUnitMultiplier.containsKey(unit)) {
+ throw new AnalysisException("invalid unit:" + tmpUnit);
+ }
+ try {
+ return Math.multiplyExact(dataVolumn, validDataVolumnUnitMultiplier.get(unit));
+ } catch (ArithmeticException e) {
+ throw new AnalysisException("data volume is too large:" + dataVolumnStr);
+ }
+ }
+
public static long analyzeReplicaNumber(String replicaNumberStr) throws AnalysisException {
long replicaNumber = 0;
try {
diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/metacache/AbstractExternalMetaCache.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/metacache/AbstractExternalMetaCache.java
index eca9816fd5f982..25b758f55f7a46 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/datasource/metacache/AbstractExternalMetaCache.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/metacache/AbstractExternalMetaCache.java
@@ -300,7 +300,7 @@ private MetaCacheEntry newMetaCacheEntry(
wrapSchemaValidator(entryDef.getLoader(), entryDef.getValueType()),
cacheSpec,
refreshExecutor, entryDef.isAutoRefresh(), entryDef.isContextualOnly(),
- MetaCacheEntry.defaultObjectStripeCount());
+ MetaCacheEntry.defaultObjectStripeCount(), entryDef.getSizeEstimator());
}
private Function wrapSchemaValidator(Function loader, Class valueType) {
diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/metacache/CacheSpec.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/metacache/CacheSpec.java
index 0bb640ad0d753c..cbab3458ef9ed1 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/datasource/metacache/CacheSpec.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/metacache/CacheSpec.java
@@ -17,7 +17,9 @@
package org.apache.doris.datasource.metacache;
+import org.apache.doris.common.AnalysisException;
import org.apache.doris.common.DdlException;
+import org.apache.doris.common.util.ParseUtil;
import org.apache.commons.lang3.math.NumberUtils;
@@ -33,7 +35,9 @@
*
* enable=false disables cache
* ttlSecond=0 disables cache, ttlSecond=-1 means no expiration
- * capacity=0 disables cache; capacity is count-based
+ * capacity=0 disables a count-bounded cache
+ * when maxWeight is present, it replaces capacity as the effective bound
+ * maxWeight accepts an optional binary unit such as KB, MB, or GB; a bare number means bytes
*
*/
public final class CacheSpec {
@@ -43,19 +47,26 @@ public final class CacheSpec {
private static final String KEY_ENABLE = ".enable";
private static final String KEY_TTL_SECOND = ".ttl-second";
private static final String KEY_CAPACITY = ".capacity";
+ private static final String KEY_MAX_WEIGHT = ".max-weight";
private final boolean enable;
private final long ttlSecond;
private final long capacity;
+ private final OptionalLong maxWeight;
- private CacheSpec(boolean enable, long ttlSecond, long capacity) {
+ private CacheSpec(boolean enable, long ttlSecond, long capacity, OptionalLong maxWeight) {
this.enable = enable;
this.ttlSecond = ttlSecond;
this.capacity = capacity;
+ this.maxWeight = Objects.requireNonNull(maxWeight, "maxWeight is required");
}
public static CacheSpec of(boolean enable, long ttlSecond, long capacity) {
- return new CacheSpec(enable, ttlSecond, capacity);
+ return new CacheSpec(enable, ttlSecond, capacity, OptionalLong.empty());
+ }
+
+ public static CacheSpec ofWeight(boolean enable, long ttlSecond, long capacity, long maxWeight) {
+ return new CacheSpec(enable, ttlSecond, capacity, OptionalLong.of(maxWeight));
}
public static PropertySpec.Builder propertySpecBuilder() {
@@ -77,12 +88,14 @@ public static CacheSpec fromProperties(Map properties, PropertyS
boolean enable = getBooleanProperty(properties, propertySpec.getEnableKey(), propertySpec.isDefaultEnable());
long ttlSecond = getLongProperty(properties, propertySpec.getTtlKey(), propertySpec.getDefaultTtlSecond());
long capacity = getLongProperty(properties, propertySpec.getCapacityKey(), propertySpec.getDefaultCapacity());
- return of(enable, ttlSecond, capacity);
+ OptionalLong maxWeight = getOptionalDataSizeProperty(
+ properties, propertySpec.getMaxWeightKey(), propertySpec.getDefaultMaxWeight());
+ return new CacheSpec(enable, ttlSecond, capacity, maxWeight);
}
/**
* Build a cache spec from catalog properties by standard external meta cache key pattern:
- * meta.cache.<engine>.<entry>.(enable|ttl-second|capacity)
+ * meta.cache.<engine>.<entry>.(enable|ttl-second|capacity|max-weight)
*/
public static CacheSpec fromProperties(Map properties,
String engine, String entryName, CacheSpec defaultSpec) {
@@ -95,6 +108,7 @@ public static PropertySpec metaCachePropertySpec(String engine, String entryName
.enable(cacheKeyPrefix + KEY_ENABLE, defaultSpec.isEnable())
.ttl(cacheKeyPrefix + KEY_TTL_SECOND, defaultSpec.getTtlSecond())
.capacity(cacheKeyPrefix + KEY_CAPACITY, defaultSpec.getCapacity())
+ .maxWeight(cacheKeyPrefix + KEY_MAX_WEIGHT, defaultSpec.getMaxWeight())
.build();
}
@@ -193,6 +207,26 @@ private static long getLongProperty(Map properties, String key,
return NumberUtils.toLong(value, defaultValue);
}
+ private static OptionalLong getOptionalDataSizeProperty(
+ Map properties, String key, OptionalLong defaultValue) {
+ if (key == null) {
+ return defaultValue;
+ }
+ String value = properties.get(key);
+ if (value == null) {
+ return defaultValue;
+ }
+ try {
+ return OptionalLong.of(ParseUtil.analyzeDataVolumeAllowZero(value));
+ } catch (AnalysisException e) {
+ throw invalidDataSizeProperty(key, value);
+ }
+ }
+
+ private static IllegalArgumentException invalidDataSizeProperty(String key, String value) {
+ return new IllegalArgumentException("The parameter " + key + " is wrong, value is " + value);
+ }
+
public boolean isEnable() {
return enable;
}
@@ -205,6 +239,20 @@ public long getCapacity() {
return capacity;
}
+ public OptionalLong getMaxWeight() {
+ return maxWeight;
+ }
+
+ public boolean isWeightBounded() {
+ return maxWeight.isPresent();
+ }
+
+ public boolean isCacheEnabled() {
+ return enable
+ && ttlSecond != CACHE_TTL_DISABLE_CACHE
+ && maxWeight.orElse(capacity) != 0L;
+ }
+
public static final class PropertySpec {
private final String enableKey;
private final boolean defaultEnable;
@@ -212,15 +260,20 @@ public static final class PropertySpec {
private final long defaultTtlSecond;
private final String capacityKey;
private final long defaultCapacity;
+ private final String maxWeightKey;
+ private final OptionalLong defaultMaxWeight;
private PropertySpec(String enableKey, boolean defaultEnable, String ttlKey,
- long defaultTtlSecond, String capacityKey, long defaultCapacity) {
+ long defaultTtlSecond, String capacityKey, long defaultCapacity,
+ String maxWeightKey, OptionalLong defaultMaxWeight) {
this.enableKey = enableKey;
this.defaultEnable = defaultEnable;
this.ttlKey = ttlKey;
this.defaultTtlSecond = defaultTtlSecond;
this.capacityKey = capacityKey;
this.defaultCapacity = defaultCapacity;
+ this.maxWeightKey = maxWeightKey;
+ this.defaultMaxWeight = defaultMaxWeight;
}
public String getEnableKey() {
@@ -247,6 +300,14 @@ public long getDefaultCapacity() {
return defaultCapacity;
}
+ public String getMaxWeightKey() {
+ return maxWeightKey;
+ }
+
+ public OptionalLong getDefaultMaxWeight() {
+ return defaultMaxWeight;
+ }
+
public static final class Builder {
private String enableKey;
private boolean defaultEnable;
@@ -254,6 +315,8 @@ public static final class Builder {
private long defaultTtlSecond;
private String capacityKey;
private long defaultCapacity;
+ private String maxWeightKey;
+ private OptionalLong defaultMaxWeight = OptionalLong.empty();
public Builder enable(String key, boolean defaultValue) {
this.enableKey = key;
@@ -273,6 +336,12 @@ public Builder capacity(String key, long defaultValue) {
return this;
}
+ public Builder maxWeight(String key, OptionalLong defaultValue) {
+ this.maxWeightKey = Objects.requireNonNull(key, "maxWeightKey is required");
+ this.defaultMaxWeight = Objects.requireNonNull(defaultValue, "defaultMaxWeight is required");
+ return this;
+ }
+
public PropertySpec build() {
return new PropertySpec(
Objects.requireNonNull(enableKey, "enableKey is required"),
@@ -280,7 +349,9 @@ public PropertySpec build() {
Objects.requireNonNull(ttlKey, "ttlKey is required"),
defaultTtlSecond,
Objects.requireNonNull(capacityKey, "capacityKey is required"),
- defaultCapacity);
+ defaultCapacity,
+ maxWeightKey,
+ defaultMaxWeight);
}
}
}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/metacache/MetaCacheEntry.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/metacache/MetaCacheEntry.java
index d3590790afed87..15b97f5185281f 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/datasource/metacache/MetaCacheEntry.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/metacache/MetaCacheEntry.java
@@ -23,6 +23,7 @@
import com.github.benmanes.caffeine.cache.Cache;
import com.github.benmanes.caffeine.cache.CacheLoader;
import com.github.benmanes.caffeine.cache.LoadingCache;
+import com.github.benmanes.caffeine.cache.Policy;
import com.github.benmanes.caffeine.cache.RemovalListener;
import com.github.benmanes.caffeine.cache.stats.CacheStats;
import com.google.common.base.Preconditions;
@@ -116,6 +117,13 @@ public MetaCacheEntry(String name, @Nullable Function loader, CacheSpec ca
defaultObjectStripeCount(), null, false);
}
+ public MetaCacheEntry(String name, @Nullable Function loader, CacheSpec cacheSpec,
+ ExecutorService refreshExecutor, boolean autoRefresh, boolean contextualOnly,
+ @Nullable MetaCacheSizeEstimator sizeEstimator) {
+ this(name, loader, cacheSpec, refreshExecutor, autoRefresh, contextualOnly,
+ defaultObjectStripeCount(), sizeEstimator, null, false);
+ }
+
public MetaCacheEntry(String name, Function loader, CacheSpec cacheSpec, ExecutorService refreshExecutor,
boolean autoRefresh, int stripeCount) {
this(name, loader, cacheSpec, refreshExecutor, autoRefresh, false, stripeCount, null, false);
@@ -126,6 +134,13 @@ public MetaCacheEntry(String name, @Nullable Function loader, CacheSpec ca
this(name, loader, cacheSpec, refreshExecutor, autoRefresh, contextualOnly, stripeCount, null, false);
}
+ public MetaCacheEntry(String name, @Nullable Function loader, CacheSpec cacheSpec,
+ ExecutorService refreshExecutor, boolean autoRefresh, boolean contextualOnly, int stripeCount,
+ @Nullable MetaCacheSizeEstimator sizeEstimator) {
+ this(name, loader, cacheSpec, refreshExecutor, autoRefresh, contextualOnly,
+ stripeCount, sizeEstimator, null, false);
+ }
+
public static MetaCacheEntry withSyncRemovalListener(String name, Function loader,
CacheSpec cacheSpec, ExecutorService refreshExecutor, RemovalListener removalListener) {
return withSyncRemovalListener(name, loader, cacheSpec, refreshExecutor,
@@ -147,9 +162,40 @@ public static MetaCacheEntry withSyncRemovalListener(String name, F
true);
}
+ public static MetaCacheEntry withSyncRemovalListener(String name, Function loader,
+ CacheSpec cacheSpec, ExecutorService refreshExecutor, MetaCacheSizeEstimator