This is an automated email from the ASF dual-hosted git repository.
morningman pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new be63e5f0cc7 [improvement](fe) Estimate wide metadata-cache values one
element at a time (#69558)
be63e5f0cc7 is described below
commit be63e5f0cc7a6976697d42c97810aa34c9a28ef9
Author: Mingyu Chen (Rayner) <[email protected]>
AuthorDate: Sun Oct 25 08:05:29 2026 +0000
[improvement](fe) Estimate wide metadata-cache values one element at a time
(#69558)
### What problem does this PR solve?
Issue Number: None
Related PR: #66717, #67726
Problem Summary:
**In short.** With a metadata-cache weight limit set, the cached
metadata of a wide table (thousands of columns), a long name list or a
long partition list was never cached: its size estimate gave up and the
value was reloaded on every statement. This PR weighs such values one
element at a time, so they are admitted and charged for their size,
while the safety bound of the estimate stays in place.
**Background.** Master (#67726) lets the metadata caches of external
catalogs be bounded by weight: the FE-global
`external_meta_cache_max_weight`, a catalog's `meta.cache.max-weight`,
or an entry's `meta.cache.<engine>.<entry>.max-weight`. Once any of them
applies, each value is weighed before it is admitted. Several entries
use the complete walk of `ReflectiveObjectSizeEstimator`, which visits
every object of the value and gives up after 10,000 of them, so that one
pathological value cannot cost unbounded CPU and temporary memory. An
estimate that gives up is incomplete, and an incomplete value is
returned to the query but not cached. branch-4.x (#66717) governs only a
few entries and weighs them with per-element formulas, so it has no such
limit.
**The problem, and what it cost.** The 10,000-visit budget is reached by
values that are large only because of one list, long before they are
large in bytes. Measured on master, the first value that never enters a
weighted cache:
| Entry | Value | Never cached from |
|---|---|---|
| `hms-table` | Hive table | 995 INT / 382 `struct<a:int,b:string>`
columns |
| `paimon-table`, `paimon-schema-at` | Paimon schema | 2,497 INT / 908
`ROW<a INT, b STRING>` columns |
| fe-core `schema` (`default`, `doris` engines) | schema of any other
external catalog | 3,332 INT / 832 struct columns |
| `adbc-table-schema` | Arrow schema | 2,499 INT / 1,110 struct fields |
| `adbc-table-names`, `adbc-namespaces` | name list | 9,999 names |
| `max-compute-partition` | partition list | 1,249 three-level
partitions |
For such a table every statement goes back to the source (an HMS call, a
Paimon schema file read, an ODPS partition listing, an ADBC metadata
call), and `weight_reject_count` in
`information_schema.catalog_meta_cache_statistics` keeps growing. In a
unit test with a 128 MiB global limit, three reads of a 2,000-column HMS
table loaded it three times and cached nothing.
**How this PR fixes it.**
- `ReflectiveObjectSizeEstimator.estimateCompletePerElement(root,
collections...)` walks `root` like the complete estimate, except that
each listed collection it reaches is counted as its own storage plus
every element estimated separately, each with its own 10,000-visit
budget. An element that appears in more than one of the listed
collections is counted once.
`MetaCacheSizeEstimators.reflectivePerElement(...)` and
`reflectiveEachElement()` wrap it as cache estimators.
- The affected entries use it: the HMS table (columns, partition keys),
the Paimon table schema and the schema-at snapshot (fields), the fe-core
schema cache (`SchemaCacheValue.getColumnLists()`: schema, and for
plugin tables also partition and display columns), the ADBC table schema
(fields) and name lists, and the MaxCompute partition list.
- The visit budget still bounds every element and the rest of the value,
so a value that reaches a huge shared structure still fails fast, and
the temporary identity set holds one element's graph at a time instead
of the whole value's. The whole-value walk and its budget are unchanged
for every other entry.
- Objects that elements share (for example a scalar type singleton) are
counted once per element, so the weight is somewhat above the retained
size (measured 1.1x to 1.6x) and never below it. A plugin table's
display columns reuse its column objects and are not counted twice.
**Results.**
| Value, under a weight limit | Before | After |
|---|---|---|
| 3,000-column HMS table (INT or struct columns) | estimate incomplete,
reloaded on every read | complete, loaded once, `weight_reject_count =
0` |
| 4,000-column Paimon table, INT or ROW columns; 4,000-field schema-at
snapshot | incomplete | complete, loaded once |
| fe-core schema with 4,000 INT + 1,000 struct columns | incomplete |
complete, loaded once |
| 3,000-field ADBC schema; 20,000 table names | incomplete | complete,
loaded once |
| 10,000 MaxCompute partitions | incomplete | complete, loaded once |
| single element larger than the visit budget | incomplete | incomplete
(unchanged) |
The new tests also check that the estimate is never below an unbounded
walk of the value and at most twice it, and that it doubles when the
column count doubles. With the per-element walk disabled, all nine new
wide-value tests fail; without the element de-duplication, a plugin
schema with reused display columns weighs twice as much (2,052,208 to
4,104,168 bytes) and its test fails.
---
.../doris/connector/adbc/AdbcMetadataCache.java | 10 +-
.../connector/adbc/AdbcMetadataCacheTest.java | 31 ++++++
.../connector/cache/MetaCacheSizeEstimators.java | 22 +++++
.../cache/ReflectiveObjectSizeEstimator.java | 49 ++++++++-
.../cache/ReflectiveObjectSizeEstimatorTest.java | 84 ++++++++++++++++
.../doris/connector/hms/HmsCacheSizeEstimator.java | 7 +-
.../connector/hms/HmsCacheSizeEstimatorTest.java | 110 +++++++++++++++++++++
.../maxcompute/MaxComputePartitionCache.java | 3 +-
.../maxcompute/MaxComputePartitionCacheTest.java | 32 ++++++
.../connector/paimon/PaimonCacheSizeEstimator.java | 8 +-
.../doris/connector/paimon/PaimonSchemaAtMemo.java | 5 +-
.../paimon/PaimonCacheSizeEstimatorTest.java | 23 +++++
.../connector/paimon/PaimonSchemaAtMemoTest.java | 34 +++++++
.../apache/doris/datasource/SchemaCacheValue.java | 9 ++
.../metacache/ExternalCatalogMetaCache.java | 12 ++-
.../plugin/PluginDrivenSchemaCacheValue.java | 6 ++
.../metacache/ExternalCatalogMetaCacheTest.java | 102 +++++++++++++++++++
17 files changed, 538 insertions(+), 9 deletions(-)
diff --git
a/fe/fe-connector/fe-connector-adbc/src/main/java/org/apache/doris/connector/adbc/AdbcMetadataCache.java
b/fe/fe-connector/fe-connector-adbc/src/main/java/org/apache/doris/connector/adbc/AdbcMetadataCache.java
index 27a7a3b2a4f..443137178d8 100644
---
a/fe/fe-connector/fe-connector-adbc/src/main/java/org/apache/doris/connector/adbc/AdbcMetadataCache.java
+++
b/fe/fe-connector/fe-connector-adbc/src/main/java/org/apache/doris/connector/adbc/AdbcMetadataCache.java
@@ -26,6 +26,7 @@ import org.apache.doris.connector.cache.ScopePath;
import org.apache.arrow.vector.types.pojo.Schema;
+import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Objects;
@@ -104,22 +105,25 @@ public final class AdbcMetadataCache {
AdbcMetadataCache(CatalogMetaCache owner, Map<String, String> properties) {
this.owner = Objects.requireNonNull(owner, "owner can not be null");
CacheSpec spec = cacheSpec(properties);
+ // Long name lists and wide schemas would exhaust the visit budget of
a whole-value walk, so these values
+ // are estimated one namespace, name or field at a time.
this.namespaces = owner.create(MetaCacheDefinition
.<String, List<AdbcNamespace>>builder("adbc-namespaces", spec,
ignored -> ScopePath.catalog())
.budgetGroup(ENTRY)
- .sizeEstimator(MetaCacheSizeEstimators.reflective())
+ .sizeEstimator(MetaCacheSizeEstimators.reflectiveEachElement())
.build());
this.tableNames = owner.create(MetaCacheDefinition
.<TableNameKey, List<String>>builder("adbc-table-names", spec,
key -> ScopePath.database(key.dorisDbName))
.budgetGroup(ENTRY)
- .sizeEstimator(MetaCacheSizeEstimators.reflective())
+ .sizeEstimator(MetaCacheSizeEstimators.reflectiveEachElement())
.build());
this.tableSchemas = owner.create(MetaCacheDefinition
.<TableKey, Schema>builder("adbc-table-schema", spec,
key -> ScopePath.table(key.dorisDbName, key.table))
.budgetGroup(ENTRY)
- .sizeEstimator(MetaCacheSizeEstimators.reflective())
+ .sizeEstimator(MetaCacheSizeEstimators.reflectivePerElement(
+ schema ->
Collections.singletonList(schema.getFields())))
.build());
}
diff --git
a/fe/fe-connector/fe-connector-adbc/src/test/java/org/apache/doris/connector/adbc/AdbcMetadataCacheTest.java
b/fe/fe-connector/fe-connector-adbc/src/test/java/org/apache/doris/connector/adbc/AdbcMetadataCacheTest.java
index 5bc22477858..bb446ce030a 100644
---
a/fe/fe-connector/fe-connector-adbc/src/test/java/org/apache/doris/connector/adbc/AdbcMetadataCacheTest.java
+++
b/fe/fe-connector/fe-connector-adbc/src/test/java/org/apache/doris/connector/adbc/AdbcMetadataCacheTest.java
@@ -27,6 +27,7 @@ import org.apache.arrow.vector.types.pojo.Schema;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
+import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Map;
@@ -72,6 +73,36 @@ class AdbcMetadataCacheTest {
}
}
+ @Test
+ void wideSchemasAndLongNameListsAreCachedUnderAWeightLimit() {
+ // A walk of the whole value ran out of the 10,000-visit budget at
about 2,500 Arrow fields or 10,000 names,
+ // so such a value was asked from the source on every statement once a
weight limit was set. MUTATION:
+ // estimating the value as one graph -> two calls and a rejected
admission -> red.
+ Map<String, String> properties = Map.of("meta.cache.max-weight",
"64MB");
+ try (CatalogMetaCache owner = CatalogMetaCache.managed(91090L, "adbc",
properties)) {
+ AdbcMetadataCache cache = new AdbcMetadataCache(owner, properties);
+ List<Field> fields = new ArrayList<>();
+ for (int i = 0; i < 3_000; i++) {
+ fields.add(Field.nullable("col_" + i, new ArrowType.Int(32,
true)));
+ }
+ List<String> tableNames = new ArrayList<>();
+ for (int i = 0; i < 20_000; i++) {
+ tableNames.add("table_" + i);
+ }
+ Counting<Schema> schema = new Counting<>(new Schema(fields));
+ Counting<List<String>> names = new Counting<>(tableNames);
+ for (int i = 0; i < 2; i++) {
+ cache.tableSchema(MAIN_T1, schema);
+ cache.tableNames(MAIN, names);
+ }
+
+ Assertions.assertEquals(1, schema.calls);
+ Assertions.assertEquals(1, names.calls);
+ Assertions.assertEquals(0L,
owner.entries().get("adbc-table-schema").metrics().getWeightRejectCount());
+ Assertions.assertEquals(0L,
owner.entries().get("adbc-table-names").metrics().getWeightRejectCount());
+ }
+ }
+
@Test
void rejectedSchemaIsReturnedWithoutBeingCached() {
Map<String, String> properties =
Map.of("meta.cache.adbc.metadata.max-weight", "1");
diff --git
a/fe/fe-connector/fe-connector-cache/src/main/java/org/apache/doris/connector/cache/MetaCacheSizeEstimators.java
b/fe/fe-connector/fe-connector-cache/src/main/java/org/apache/doris/connector/cache/MetaCacheSizeEstimators.java
index 8ec78638fa0..95757f9aaae 100644
---
a/fe/fe-connector/fe-connector-cache/src/main/java/org/apache/doris/connector/cache/MetaCacheSizeEstimators.java
+++
b/fe/fe-connector/fe-connector-cache/src/main/java/org/apache/doris/connector/cache/MetaCacheSizeEstimators.java
@@ -17,6 +17,10 @@
package org.apache.doris.connector.cache;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.function.Function;
+
/** Shared estimators for small or structurally unknown metadata-cache
entries. */
public final class MetaCacheSizeEstimators {
private MetaCacheSizeEstimators() {
@@ -32,4 +36,22 @@ public final class MetaCacheSizeEstimators {
ReflectiveObjectSizeEstimator.estimateComplete(key),
ReflectiveObjectSizeEstimator.estimateComplete(value)));
}
+
+ /**
+ * Returns a {@link #reflective()} estimator that walks the long lists
{@code perElementCollections} picks from
+ * a value element by element (see {@link
ReflectiveObjectSizeEstimator#estimateCompletePerElement}), for values
+ * that grow with one list, such as a wide table's schema.
+ */
+ public static <K, V> MetaCacheSizeEstimator<K, V> reflectivePerElement(
+ Function<? super V, ? extends Collection<? extends Collection<?>>>
perElementCollections) {
+ return (key, value) ->
MetaCacheSizeEstimate.complete(JvmSizeUtils.saturatedAdd(
+ ReflectiveObjectSizeEstimator.estimateComplete(key),
+ ReflectiveObjectSizeEstimator.estimateCompletePerElement(value,
+ perElementCollections.apply(value).toArray(new
Collection<?>[0]))));
+ }
+
+ /** {@link #reflectivePerElement} for a value that is itself the long
list, such as a partition list. */
+ public static <K, V extends Collection<?>> MetaCacheSizeEstimator<K, V>
reflectiveEachElement() {
+ return reflectivePerElement(value -> Collections.singletonList(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
index 5f1d0c56e38..1d85a5d8c66 100644
---
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
@@ -23,6 +23,7 @@ import java.lang.reflect.Modifier;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Collection;
+import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.IdentityHashMap;
@@ -48,7 +49,9 @@ import java.util.concurrent.ConcurrentHashMap;
* The diagnostic API samples collections and object arrays. The admission API
instead traverses every element
* up to a fixed visit budget: this catches skewed tail values while bounding
CPU and temporary identity-set
* memory. Strongly encapsulated reference fields, depth truncation, and
visit-budget exhaustion fail the
- * estimate so admission can reject the value without turning an incomplete
graph into a false low weight.
+ * estimate so admission can reject the value without turning an incomplete
graph into a false low weight. A value
+ * that is large only because of a few long lists, such as a wide table's
columns, is estimated with those lists
+ * walked element by element instead, so the budget bounds one element rather
than the whole value.
*
* <p>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
@@ -141,14 +144,36 @@ public final class ReflectiveObjectSizeEstimator {
return new Walker(Integer.MAX_VALUE, visitBudget).estimate(root,
maxDepth);
}
+ /**
+ * Completely estimates {@code root} like {@link
#estimateComplete(Object)}, except that each of
+ * {@code perElementCollections} that the walk reaches is estimated
element by element, every element with its
+ * own visit budget. A value that is large only because of such lists,
such as a wide table's columns or a long
+ * partition list, stays admissible, while the budget still bounds each
element and the rest of the graph, and
+ * the temporary identity set holds one element's graph at a time. Objects
that elements share are counted
+ * once per element, which can only over-estimate; an element in several
of the collections is counted once.
+ */
+ public static long estimateCompletePerElement(Object root,
Collection<?>... perElementCollections) {
+ Set<Object> collections = Collections.newSetFromMap(new
IdentityHashMap<>());
+ Collections.addAll(collections, perElementCollections);
+ return new Walker(Integer.MAX_VALUE, DEFAULT_COMPLETE_VISIT_BUDGET,
collections)
+ .estimate(root, DEFAULT_MAX_DEPTH);
+ }
+
private static final class Walker {
private final int sampleSize;
private final int visitBudget;
private final Map<Object, Boolean> visited = new IdentityHashMap<>();
+ private final Set<Object> perElementCollections;
+ private final Set<Object> estimatedElements =
Collections.newSetFromMap(new IdentityHashMap<>());
private Walker(int sampleSize, int visitBudget) {
+ this(sampleSize, visitBudget, Collections.emptySet());
+ }
+
+ private Walker(int sampleSize, int visitBudget, Set<Object>
perElementCollections) {
this.sampleSize = sampleSize;
this.visitBudget = visitBudget;
+ this.perElementCollections = perElementCollections;
}
private long estimate(Object value, int depth) {
@@ -159,6 +184,9 @@ public final class ReflectiveObjectSizeEstimator {
throw new IllegalStateException(
"Complete estimate exceeded visit budget " +
visitBudget);
}
+ if (perElementCollections.contains(value)) {
+ return estimateEachElement((Collection<?>) value, depth);
+ }
Class<?> type = value.getClass();
if (type.isEnum() || value instanceof Class<?>) {
@@ -230,6 +258,25 @@ public final class ReflectiveObjectSizeEstimator {
return add(bytes, scale(sampledBytes, samples, length));
}
+ private long estimateEachElement(Collection<?> values, int depth) {
+ int size = values.size();
+ long bytes = add(
+ JvmSizeUtils.instanceSize(values.getClass()),
+ collectionStorageBytes(values, size));
+ if (size == 0) {
+ return bytes;
+ }
+ if (depth == 0) {
+ throw incomplete(values.getClass());
+ }
+ for (Object element : values) {
+ if (estimatedElements.add(element)) {
+ bytes = add(bytes, new Walker(Integer.MAX_VALUE,
visitBudget).estimate(element, depth - 1));
+ }
+ }
+ return bytes;
+ }
+
private long estimateByteBuffer(ByteBuffer value, int depth) {
long bytes = JvmSizeUtils.instanceSize(value.getClass());
byte[] backing = JvmSizeUtils.byteBufferBackingArray(value);
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
index edb3d3f843e..ccef17bec58 100644
---
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
@@ -23,7 +23,9 @@ import org.junit.jupiter.api.Test;
import java.net.URI;
import java.nio.ByteBuffer;
import java.util.AbstractList;
+import java.util.ArrayList;
import java.util.LinkedHashMap;
+import java.util.List;
import java.util.Map;
import java.util.RandomAccess;
import java.util.concurrent.atomic.AtomicInteger;
@@ -129,6 +131,88 @@ class ReflectiveObjectSizeEstimatorTest {
Assertions.assertTrue(estimate.getIncompleteReason().startsWith("depth_limit:"));
}
+ @Test
+ void perElementEstimateCompletesALongListThatTheWholeValueWalkRejects() {
+ // A value that is large only because of one long list, such as a wide
table's columns, has the list walked
+ // element by element, each element under its own visit budget.
MUTATION: walking the list as part of the
+ // value's graph -> the visit budget runs out -> red.
+ Holder small = holderOf(1_000);
+ Holder large = holderOf(30_000);
+ MetaCacheSizeEstimate whole =
MetaCacheSizeEstimator.estimateSafely("whole_value",
+ () ->
MetaCacheSizeEstimate.complete(ReflectiveObjectSizeEstimator.estimateComplete(large)));
+ Assertions.assertFalse(whole.isComplete());
+
+ long smallBytes =
ReflectiveObjectSizeEstimator.estimateCompletePerElement(small, small.values);
+ long largeBytes =
ReflectiveObjectSizeEstimator.estimateCompletePerElement(large, large.values);
+
+ // The elements share no object, so the estimate is exactly an
unbounded walk of the value.
+
Assertions.assertEquals(ReflectiveObjectSizeEstimator.estimateComplete(large,
Integer.MAX_VALUE, 20),
+ largeBytes);
+ Assertions.assertTrue(largeBytes > smallBytes * 25, "the estimate must
grow with the list");
+ }
+
+ @Test
+ void perElementEstimateCountsAnElementInTwoOfTheListsOnce() {
+ // A schema's display columns reuse most of its column objects.
MUTATION: estimating every element of every
+ // listed collection -> the shared elements count twice -> red.
+ Holder holder = holderOf(20_000);
+ holder.view = new ArrayList<>(holder.values);
+
+
Assertions.assertEquals(ReflectiveObjectSizeEstimator.estimateComplete(holder,
Integer.MAX_VALUE, 20),
+
ReflectiveObjectSizeEstimator.estimateCompletePerElement(holder, holder.values,
holder.view));
+ }
+
+ @Test
+ void perElementEstimateStillBoundsEachElement() {
+ AtomicInteger reads = new AtomicInteger();
+ Holder holder = holderOf(10);
+ holder.values.add(new CountingList(reads));
+
+ MetaCacheSizeEstimate estimate =
MetaCacheSizeEstimator.estimateSafely("bounded_element",
+ () -> MetaCacheSizeEstimate.complete(
+
ReflectiveObjectSizeEstimator.estimateCompletePerElement(holder,
holder.values)));
+
+ Assertions.assertFalse(estimate.isComplete());
+ Assertions.assertTrue(reads.get() <= 10_001, "one element's walk must
stop at the visit budget");
+ }
+
+ @Test
+ void eachElementEstimatorWeighsAValueThatIsTheLongList() {
+ List<Object> partitions = holderOf(30_000).values;
+
+ MetaCacheSizeEstimate estimate = MetaCacheSizeEstimators
+ .<String, List<Object>>reflectiveEachElement().estimate("key",
partitions);
+
+ Assertions.assertTrue(estimate.isComplete(),
estimate.getIncompleteReason());
+
Assertions.assertEquals(ReflectiveObjectSizeEstimator.estimateComplete("key")
+ + ReflectiveObjectSizeEstimator.estimateComplete(partitions,
Integer.MAX_VALUE, 20),
+ estimate.getBytes());
+ }
+
+ private static Holder holderOf(int elements) {
+ Holder holder = new Holder();
+ for (int i = 0; i < elements; i++) {
+ holder.values.add(new Element("element-" + i, 1_000L + i));
+ }
+ return holder;
+ }
+
+ private static final class Holder {
+ private final String name = "holder";
+ private final List<Object> values = new ArrayList<>();
+ private List<Object> view;
+ }
+
+ private static final class Element {
+ private final String name;
+ private final Long id;
+
+ private Element(String name, Long id) {
+ this.name = name;
+ this.id = id;
+ }
+ }
+
private static final class CountingList extends AbstractList<String>
implements RandomAccess {
private final AtomicInteger reads;
diff --git
a/fe/fe-connector/fe-connector-hms/src/main/java/org/apache/doris/connector/hms/HmsCacheSizeEstimator.java
b/fe/fe-connector/fe-connector-hms/src/main/java/org/apache/doris/connector/hms/HmsCacheSizeEstimator.java
index 59dfd6596e2..10ea2908a32 100644
---
a/fe/fe-connector/fe-connector-hms/src/main/java/org/apache/doris/connector/hms/HmsCacheSizeEstimator.java
+++
b/fe/fe-connector/fe-connector-hms/src/main/java/org/apache/doris/connector/hms/HmsCacheSizeEstimator.java
@@ -32,7 +32,12 @@ final class HmsCacheSizeEstimator {
}
static MetaCacheSizeEstimate estimateTable(Object key, HmsTableInfo value)
{
- return complete(key, value);
+ // A whole-value walk of a wide table would exhaust the visit budget
and keep the table out of a weighted
+ // cache, so each column is estimated on its own.
+ return MetaCacheSizeEstimate.complete(add(
+ ReflectiveObjectSizeEstimator.estimateComplete(key),
+ ReflectiveObjectSizeEstimator.estimateCompletePerElement(
+ value, value.getColumns(), value.getPartitionKeys())));
}
static MetaCacheSizeEstimate estimatePartition(Object key,
HmsPartitionInfo value) {
diff --git
a/fe/fe-connector/fe-connector-hms/src/test/java/org/apache/doris/connector/hms/HmsCacheSizeEstimatorTest.java
b/fe/fe-connector/fe-connector-hms/src/test/java/org/apache/doris/connector/hms/HmsCacheSizeEstimatorTest.java
new file mode 100644
index 00000000000..824f442352a
--- /dev/null
+++
b/fe/fe-connector/fe-connector-hms/src/test/java/org/apache/doris/connector/hms/HmsCacheSizeEstimatorTest.java
@@ -0,0 +1,110 @@
+// 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.hms;
+
+import org.apache.doris.connector.cache.CacheSpec;
+import org.apache.doris.connector.cache.CatalogMetaCache;
+import org.apache.doris.connector.cache.MetaCache;
+import org.apache.doris.connector.cache.MetaCacheBudgetManager;
+import org.apache.doris.connector.cache.MetaCacheDefinition;
+import org.apache.doris.connector.cache.MetaCacheSizeEstimate;
+import org.apache.doris.connector.cache.ReflectiveObjectSizeEstimator;
+import org.apache.doris.connector.cache.ScopePath;
+import org.apache.doris.connector.spi.ConnectorColumn;
+import org.apache.doris.connector.spi.ConnectorType;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.OptionalLong;
+import java.util.concurrent.atomic.AtomicInteger;
+
+class HmsCacheSizeEstimatorTest {
+
+ @Test
+ void wideTablesAreWeighedCompletelyAndLinearly() {
+ // A walk of the whole table ran out of the 10,000-visit budget at 995
INT or 382 struct columns, so a wider
+ // table never entered a weighted cache. MUTATION: estimating the
table as one graph -> incomplete -> red.
+ for (String hiveType : List.of("int", "struct<a:int,b:string>")) {
+ MetaCacheSizeEstimate half =
HmsCacheSizeEstimator.estimateTable("db.wide", table(1_500, hiveType));
+ HmsTableInfo table = table(3_000, hiveType);
+ MetaCacheSizeEstimate full =
HmsCacheSizeEstimator.estimateTable("db.wide", table);
+
+ Assertions.assertTrue(full.isComplete(), hiveType + ": " +
full.getIncompleteReason());
+ // Never below an unbounded walk of the value; above it only by
the objects that columns share.
+ long walked =
ReflectiveObjectSizeEstimator.estimateComplete("db.wide", Integer.MAX_VALUE, 20)
+ + ReflectiveObjectSizeEstimator.estimateComplete(table,
Integer.MAX_VALUE, 20);
+ Assertions.assertTrue(full.getBytes() >= walked, hiveType + ": " +
full.getBytes() + " < " + walked);
+ Assertions.assertTrue(full.getBytes() <= walked * 2,
+ hiveType + ": " + full.getBytes() + " > 2 * " + walked);
+ double growth = (double) full.getBytes() / half.getBytes();
+ Assertions.assertTrue(growth > 1.8 && growth < 2.2, hiveType + ":
doubling the columns grew it " + growth);
+ }
+ }
+
+ @Test
+ void wideTableIsCachedUnderAWeightLimit() {
+ MetaCacheBudgetManager manager = new
MetaCacheBudgetManager(OptionalLong.of(128L << 20));
+ try (CatalogMetaCache owner = new CatalogMetaCache(manager, 1L,
"hive", Collections.emptyMap())) {
+ MetaCache<String, HmsTableInfo> entry =
owner.create(MetaCacheDefinition
+ .<String, HmsTableInfo>builder("hms-table",
CacheSpec.of(true, -1L, 100L),
+ ignored -> ScopePath.catalog())
+ .sizeEstimator(HmsCacheSizeEstimator::estimateTable)
+ .build());
+ HmsTableInfo wide = table(3_000, "struct<a:int,b:string>");
+ AtomicInteger loads = new AtomicInteger();
+ for (int i = 0; i < 3; i++) {
+ Assertions.assertSame(wide, entry.get("db.wide", ignored -> {
+ loads.incrementAndGet();
+ return wide;
+ }));
+ }
+
+ Assertions.assertEquals(1, loads.get());
+ Assertions.assertSame(wide, entry.getIfPresent("db.wide"));
+ Assertions.assertEquals(0L,
entry.metrics().getWeightRejectCount());
+ Assertions.assertTrue(entry.metrics().getEstimatedWeight() > 0L);
+ }
+ }
+
+ private static HmsTableInfo table(int columns, String hiveType) {
+ List<ConnectorColumn> dataColumns = new ArrayList<>(columns);
+ for (int i = 0; i < columns; i++) {
+ dataColumns.add(new ConnectorColumn("col_" + i,
HmsTypeMapping.toConnectorType(hiveType), "comment " + i,
+ true, null, true));
+ }
+ Map<String, String> parameters = new HashMap<>();
+ for (int i = 0; i < 10; i++) {
+ parameters.put("param_" + i, "value_" + i);
+ }
+ return
HmsTableInfo.builder().dbName("db").tableName("wide").tableType("EXTERNAL_TABLE")
+ .parameters(parameters).columns(dataColumns)
+ .partitionKeys(Collections.singletonList(
+ new ConnectorColumn("dt", ConnectorType.of("STRING"),
null, true, null, true)))
+ .location("hdfs://nn/warehouse/db/wide")
+
.inputFormat("org.apache.hadoop.hive.ql.io.parquet.MapredParquetInputFormat")
+
.outputFormat("org.apache.hadoop.hive.ql.io.parquet.MapredParquetOutputFormat")
+
.serializationLib("org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe")
+ .build();
+ }
+}
diff --git
a/fe/fe-connector/fe-connector-maxcompute/src/main/java/org/apache/doris/connector/maxcompute/MaxComputePartitionCache.java
b/fe/fe-connector/fe-connector-maxcompute/src/main/java/org/apache/doris/connector/maxcompute/MaxComputePartitionCache.java
index 0f1aaeddccd..34a22b5dfd8 100644
---
a/fe/fe-connector/fe-connector-maxcompute/src/main/java/org/apache/doris/connector/maxcompute/MaxComputePartitionCache.java
+++
b/fe/fe-connector/fe-connector-maxcompute/src/main/java/org/apache/doris/connector/maxcompute/MaxComputePartitionCache.java
@@ -104,7 +104,8 @@ public class MaxComputePartitionCache {
.<PartitionKey,
List<PartitionSpec>>builder("max-compute-partition", spec,
key -> ScopePath.table(key.dbName, key.tableName))
.budgetGroup(ENTRY_PARTITION)
- .sizeEstimator(MetaCacheSizeEstimators.reflective())
+ // A long partition list would exhaust the visit budget of a
whole-value walk.
+ .sizeEstimator(MetaCacheSizeEstimators.reflectiveEachElement())
.build());
this.lister = Objects.requireNonNull(lister, "lister can not be null");
}
diff --git
a/fe/fe-connector/fe-connector-maxcompute/src/test/java/org/apache/doris/connector/maxcompute/MaxComputePartitionCacheTest.java
b/fe/fe-connector/fe-connector-maxcompute/src/test/java/org/apache/doris/connector/maxcompute/MaxComputePartitionCacheTest.java
index 61b8d2acaf7..ad85d2a09fa 100644
---
a/fe/fe-connector/fe-connector-maxcompute/src/test/java/org/apache/doris/connector/maxcompute/MaxComputePartitionCacheTest.java
+++
b/fe/fe-connector/fe-connector-maxcompute/src/test/java/org/apache/doris/connector/maxcompute/MaxComputePartitionCacheTest.java
@@ -142,6 +142,38 @@ public class MaxComputePartitionCacheTest {
}
}
+ @Test
+ public void longPartitionListIsCachedUnderAWeightLimit() throws Exception {
+ // A walk of the whole list ran out of the 10,000-visit budget at
about 1,250 three-level partitions, so a
+ // longer listing was fetched from ODPS on every statement once a
weight limit was set. MUTATION: estimating
+ // the list as one graph -> two loads and a rejected admission -> red.
+ Odps odps = new Odps(new AliyunAccount("ak", "sk"));
+ Constructor<Partition> constructor =
Partition.class.getDeclaredConstructor(
+ PartitionSpec.class, String.class, String.class, String.class,
Odps.class);
+ constructor.setAccessible(true);
+ List<Partition> partitions = new ArrayList<>();
+ for (int i = 0; i < 10_000; i++) {
+ PartitionSpec spec = new PartitionSpec(
+ "ds='2026-09-" + (i % 30) + "',hr='" + (i % 24) + "',id='"
+ i + "'");
+ partitions.add(constructor.newInstance(spec, "project", null,
"table", odps));
+ }
+ Map<String, String> props = Map.of("meta.cache.max-weight", "64MB");
+ AtomicInteger loads = new AtomicInteger();
+ try (CatalogMetaCache owner = CatalogMetaCache.managed(91091L,
"max_compute", props)) {
+ MaxComputePartitionCache cache = new
MaxComputePartitionCache(owner, props, (db, table) -> {
+ loads.incrementAndGet();
+ return partitions;
+ });
+ for (int i = 0; i < 2; i++) {
+ Assertions.assertEquals(10_000, cache.getPartitions("db",
"t").size());
+ }
+
+ Assertions.assertEquals(1, loads.get());
+ Assertions.assertEquals(0L,
+
owner.entries().get("max-compute-partition").metrics().getWeightRejectCount());
+ }
+ }
+
// ==================== caching: hit / miss keyed by (db, table)
====================
@Test
diff --git
a/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonCacheSizeEstimator.java
b/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonCacheSizeEstimator.java
index d5260ed8549..64ee96eb7ea 100644
---
a/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonCacheSizeEstimator.java
+++
b/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonCacheSizeEstimator.java
@@ -135,7 +135,7 @@ final class PaimonCacheSizeEstimator {
return add(bytes, estimateFileStoreTable(
((DelegatedFileStoreTable) table).wrapped(), visited));
}
- bytes = add(bytes, estimateCompleteOnce(table.schema(), visited));
+ bytes = add(bytes, estimateSchemaOnce(table.schema(), visited));
bytes = add(bytes, estimateStoreGrowth(table.schema()));
bytes = add(bytes, estimateTableFileIO(table.fileIO(), visited));
bytes = add(bytes, estimatePath(table.location(), visited));
@@ -200,6 +200,12 @@ final class PaimonCacheSizeEstimator {
return add(JvmSizeUtils.instanceSize(fileIO.getClass()),
FILE_IO_BYTES);
}
+ /** A whole-value walk of a wide schema would exhaust the visit budget, so
each field is estimated on its own. */
+ private static long estimateSchemaOnce(TableSchema schema, Set<Object>
visited) {
+ return !visited.add(schema)
+ ? 0L :
ReflectiveObjectSizeEstimator.estimateCompletePerElement(schema,
schema.fields());
+ }
+
private static long estimateCompleteOnce(Object value, Set<Object>
visited) {
return value == null || !visited.add(value)
? 0L : ReflectiveObjectSizeEstimator.estimateComplete(value);
diff --git
a/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonSchemaAtMemo.java
b/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonSchemaAtMemo.java
index d1cc19beff9..3f9336c631e 100644
---
a/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonSchemaAtMemo.java
+++
b/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonSchemaAtMemo.java
@@ -24,6 +24,7 @@ import org.apache.doris.connector.cache.MetaCacheDefinition;
import org.apache.doris.connector.cache.MetaCacheSizeEstimators;
import org.apache.doris.connector.cache.ScopePath;
+import java.util.Collections;
import java.util.Objects;
import java.util.function.Supplier;
@@ -70,7 +71,9 @@ final class PaimonSchemaAtMemo {
.<MemoKey, PaimonCatalogOps.PaimonSchemaSnapshot>builder(
"paimon-schema-at", spec,
key -> ScopePath.table(key.databaseName,
key.tableName))
- .sizeEstimator(MetaCacheSizeEstimators.reflective())
+ // A wide schema's fields would exhaust the visit budget of a
whole-value walk.
+ .sizeEstimator(MetaCacheSizeEstimators.reflectivePerElement(
+ snapshot ->
Collections.singletonList(snapshot.fields())))
.build());
}
diff --git
a/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonCacheSizeEstimatorTest.java
b/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonCacheSizeEstimatorTest.java
index a7399c41a76..84e1ff99709 100644
---
a/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonCacheSizeEstimatorTest.java
+++
b/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonCacheSizeEstimatorTest.java
@@ -109,6 +109,16 @@ class PaimonCacheSizeEstimatorTest {
table("nested_many", 0, 0, 0, 200, null));
}
+ @Test
+ void wideSchemasAreWeighedCompletely() throws Exception {
+ // A walk of the whole schema ran out of the 10,000-visit budget at
about 2,500 INT or 900 ROW columns, so a
+ // wider table never entered a weighted cache. MUTATION: estimating
the schema as one graph -> incomplete
+ // -> red.
+ assertGrownDelta("wide fields", table("wide_fields_2000", 2_000, 0, 0,
0, null),
+ table("wide_fields_4000", 4_000, 0, 0, 0, null));
+ weight(rowColumnsTable("wide_rows_4000", 4_000));
+ }
+
@Test
void everyTableCarriesAnAllowanceForItsFileIo() throws Exception {
// A REST catalog gives each table its own RESTTokenFileIO or
ResolvingFileIO and a Hive catalog a FileIO
@@ -289,6 +299,19 @@ class PaimonCacheSizeEstimatorTest {
return FileStoreTableFactory.create(fileIO, path);
}
+ private FileStoreTable rowColumnsTable(String name, int columns) throws
Exception {
+ Schema.Builder builder = Schema.newBuilder().column("part",
DataTypes.INT());
+ for (int i = 0; i < columns; i++) {
+ builder.column("row_" + i, DataTypes.ROW(DataTypes.FIELD(0, "a",
DataTypes.INT()),
+ DataTypes.FIELD(1, "b", DataTypes.STRING())));
+ }
+ builder.partitionKeys("part");
+ LocalFileIO fileIO = LocalFileIO.create();
+ Path path = new Path(warehouse.resolve(name).toUri());
+ new FileSystemSchemaManager(fileIO, path).createTable(builder.build());
+ return FileStoreTableFactory.create(fileIO, path);
+ }
+
private static RowType rowOf(int fieldCount) {
List<DataField> fields = new ArrayList<>();
for (int i = 0; i < fieldCount; i++) {
diff --git
a/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonSchemaAtMemoTest.java
b/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonSchemaAtMemoTest.java
index 5b3aecd623c..cfde3692623 100644
---
a/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonSchemaAtMemoTest.java
+++
b/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonSchemaAtMemoTest.java
@@ -17,10 +17,17 @@
package org.apache.doris.connector.paimon;
+import org.apache.doris.connector.cache.CatalogMetaCache;
+
+import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.DataTypes;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
+import java.util.ArrayList;
import java.util.Collections;
+import java.util.List;
+import java.util.Map;
import java.util.concurrent.atomic.AtomicInteger;
/**
@@ -40,6 +47,33 @@ public class PaimonSchemaAtMemoTest {
Collections.emptyList(), Collections.emptyList(),
Collections.emptyList());
}
+ @Test
+ public void wideSchemaIsCachedUnderAWeightLimit() {
+ // A walk of the whole snapshot ran out of the 10,000-visit budget at
about 2,500 fields, so a wider schema
+ // was read again on every time-travel query once a weight limit was
set. MUTATION: estimating the snapshot
+ // as one graph -> two loads and a rejected admission -> red.
+ Map<String, String> properties = Map.of("meta.cache.max-weight",
"64MB");
+ try (CatalogMetaCache owner = CatalogMetaCache.managed(91201L,
"paimon", properties)) {
+ PaimonSchemaAtMemo memo = new PaimonSchemaAtMemo(owner, 100);
+ List<DataField> fields = new ArrayList<>();
+ for (int i = 0; i < 4_000; i++) {
+ fields.add(new DataField(i, "col_" + i, DataTypes.INT(),
"comment " + i));
+ }
+ PaimonCatalogOps.PaimonSchemaSnapshot wide = new
PaimonCatalogOps.PaimonSchemaSnapshot(
+ fields, Collections.emptyList(), Collections.emptyList());
+ AtomicInteger loads = new AtomicInteger();
+ for (int i = 0; i < 2; i++) {
+ memo.getOrLoad(handle("db", "t"), 5L, () -> {
+ loads.incrementAndGet();
+ return wide;
+ });
+ }
+
+ Assertions.assertEquals(1, loads.get());
+ Assertions.assertEquals(0L,
owner.entries().get("paimon-schema-at").metrics().getWeightRejectCount());
+ }
+ }
+
@Test
public void sameKeyLoadsOnce() {
PaimonSchemaAtMemo memo = new PaimonSchemaAtMemo(100);
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/datasource/SchemaCacheValue.java
b/fe/fe-core/src/main/java/org/apache/doris/datasource/SchemaCacheValue.java
index ac7c55105af..5b641ff6df5 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/datasource/SchemaCacheValue.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/SchemaCacheValue.java
@@ -19,6 +19,7 @@ package org.apache.doris.datasource;
import org.apache.doris.catalog.Column;
+import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
@@ -40,6 +41,14 @@ public class SchemaCacheValue {
return schema;
}
+ /**
+ * The column lists that grow with the table's width. The cache's size
estimator walks them column by column, so
+ * a wide table stays admissible to a weighted schema cache.
+ */
+ public List<List<Column>> getColumnLists() {
+ return Collections.singletonList(schema);
+ }
+
public void validateSchema() throws IllegalArgumentException {
Set<String> columnNames = new HashSet<>();
for (Column column : schema) {
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/datasource/metacache/ExternalCatalogMetaCache.java
b/fe/fe-core/src/main/java/org/apache/doris/datasource/metacache/ExternalCatalogMetaCache.java
index b7f9809e264..4a639d8b202 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/datasource/metacache/ExternalCatalogMetaCache.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/datasource/metacache/ExternalCatalogMetaCache.java
@@ -22,6 +22,7 @@ import org.apache.doris.common.Config;
import org.apache.doris.connector.cache.CacheSpec;
import org.apache.doris.connector.cache.MetaCache;
import org.apache.doris.connector.cache.MetaCacheDefinition;
+import org.apache.doris.connector.cache.MetaCacheSizeEstimator;
import org.apache.doris.connector.cache.MetaCacheSizeEstimators;
import org.apache.doris.datasource.CacheException;
import org.apache.doris.datasource.CatalogIf;
@@ -30,6 +31,7 @@ import org.apache.doris.datasource.ExternalTable;
import org.apache.doris.datasource.NameMapping;
import org.apache.doris.datasource.SchemaCacheValue;
+import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.Maps;
import java.time.Duration;
@@ -290,7 +292,7 @@ public abstract class ExternalCatalogMetaCache implements
ExternalMetaCache {
Function<K, V> loader = wrapSchemaValidator(entryDef.getLoader(),
entryDef.getValueType());
MetaCacheDefinition.Builder<K, V> builder = MetaCacheDefinition.<K,
V>builder(
entryDef.getName(), cacheSpec,
entryDef.getInvalidation()::scope)
- .sizeEstimator(MetaCacheSizeEstimators.reflective());
+ .sizeEstimator(sizeEstimator(entryDef.getValueType()));
if (loader != null) {
builder.loader(loader);
}
@@ -301,6 +303,14 @@ public abstract class ExternalCatalogMetaCache implements
ExternalMetaCache {
return group.owner().create(builder.build());
}
+ /** A whole-value walk of a wide table's schema would exhaust the visit
budget, so it goes column by column. */
+ @VisibleForTesting
+ static <K, V> MetaCacheSizeEstimator<K, V> sizeEstimator(Class<V>
valueType) {
+ return SchemaCacheValue.class.isAssignableFrom(valueType)
+ ? MetaCacheSizeEstimators.reflectivePerElement(value ->
((SchemaCacheValue) value).getColumnLists())
+ : MetaCacheSizeEstimators.reflective();
+ }
+
private <K, V> Function<K, V> wrapSchemaValidator(Function<K, V> loader,
Class<V> valueType) {
if (loader == null) {
return null;
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/datasource/plugin/PluginDrivenSchemaCacheValue.java
b/fe/fe-core/src/main/java/org/apache/doris/datasource/plugin/PluginDrivenSchemaCacheValue.java
index 4e7650aa2a8..71e1b646b63 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/datasource/plugin/PluginDrivenSchemaCacheValue.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/datasource/plugin/PluginDrivenSchemaCacheValue.java
@@ -21,6 +21,7 @@ import org.apache.doris.catalog.Column;
import org.apache.doris.connector.spi.ConnectorCapability;
import org.apache.doris.datasource.SchemaCacheValue;
+import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.Map;
@@ -141,4 +142,9 @@ public class PluginDrivenSchemaCacheValue extends
SchemaCacheValue {
public List<Column> getDisplayColumns() {
return displayColumns;
}
+
+ @Override
+ public List<List<Column>> getColumnLists() {
+ return Arrays.asList(schema, partitionColumns, displayColumns);
+ }
}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/datasource/metacache/ExternalCatalogMetaCacheTest.java
b/fe/fe-core/src/test/java/org/apache/doris/datasource/metacache/ExternalCatalogMetaCacheTest.java
index 34b989380f3..7524288f2cb 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/datasource/metacache/ExternalCatalogMetaCacheTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/datasource/metacache/ExternalCatalogMetaCacheTest.java
@@ -19,20 +19,32 @@ package org.apache.doris.datasource.metacache;
import org.apache.doris.catalog.Column;
import org.apache.doris.catalog.PrimitiveType;
+import org.apache.doris.catalog.ScalarType;
+import org.apache.doris.catalog.StructField;
+import org.apache.doris.catalog.StructType;
import org.apache.doris.connector.cache.CacheSpec;
import org.apache.doris.connector.cache.MetaCache;
+import org.apache.doris.connector.cache.MetaCacheSizeEstimate;
+import org.apache.doris.connector.cache.MetaCacheSizeEstimator;
+import org.apache.doris.connector.cache.ReflectiveObjectSizeEstimator;
import org.apache.doris.datasource.NameMapping;
import org.apache.doris.datasource.SchemaCacheKey;
import org.apache.doris.datasource.SchemaCacheValue;
+import org.apache.doris.datasource.plugin.PluginDrivenSchemaCacheValue;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Supplier;
public class ExternalCatalogMetaCacheTest {
@@ -145,6 +157,96 @@ public class ExternalCatalogMetaCacheTest {
}
}
+ @Test
+ public void testWideSchemaIsCachedUnderAWeightLimit() {
+ // A walk of the whole value ran out of the 10,000-visit budget at
about 3,300 INT or 830 struct columns, so a
+ // wider table's schema was loaded again on every statement once a
weight limit was set. MUTATION: estimating
+ // the schema as one graph -> two loads and a rejected admission ->
red.
+ ExecutorService refreshExecutor = Executors.newSingleThreadExecutor();
+ try {
+ SchemaCacheValue wide = new SchemaCacheValue(wideColumns(4_000,
1_000));
+ AtomicInteger loads = new AtomicInteger();
+ WideSchemaMetaCache cache = new
WideSchemaMetaCache(refreshExecutor, () -> {
+ loads.incrementAndGet();
+ return wide;
+ });
+ Map<String, String> properties = Maps.newHashMap();
+ properties.put("meta.cache.max-weight", "64MB");
+ cache.initCatalog(2L, properties);
+ MetaCache<SchemaCacheKey, SchemaCacheValue> schemaEntry =
cache.entry(
+ 2L, "schema", SchemaCacheKey.class,
SchemaCacheValue.class);
+ SchemaCacheKey key = new
SchemaCacheKey(NameMapping.createForTest(2L, "db1", "wide"));
+ for (int i = 0; i < 2; i++) {
+ Assertions.assertSame(wide, schemaEntry.get(key));
+ }
+
+ Assertions.assertEquals(1, loads.get());
+ Assertions.assertEquals(0L,
schemaEntry.metrics().getWeightRejectCount());
+ cache.invalidateCatalog(2L);
+ } finally {
+ refreshExecutor.shutdownNow();
+ }
+ }
+
+ @Test
+ public void
testSchemaEstimateIsCompleteLinearAndCountsReusedDisplayColumnsOnce() {
+ MetaCacheSizeEstimator<SchemaCacheKey, SchemaCacheValue> estimator =
+ ExternalCatalogMetaCache.sizeEstimator(SchemaCacheValue.class);
+ SchemaCacheKey key = new SchemaCacheKey(NameMapping.createForTest(3L,
"db1", "wide"));
+ MetaCacheSizeEstimate half = estimator.estimate(key, new
SchemaCacheValue(wideColumns(2_000, 500)));
+ List<Column> columns = wideColumns(4_000, 1_000);
+ SchemaCacheValue value = new SchemaCacheValue(columns);
+ MetaCacheSizeEstimate full = estimator.estimate(key, value);
+
+ Assertions.assertTrue(full.isComplete(), full.getIncompleteReason());
+ // Never below an unbounded walk of the value; above it only by the
objects that columns share.
+ long walked = ReflectiveObjectSizeEstimator.estimateComplete(key,
Integer.MAX_VALUE, 20)
+ + ReflectiveObjectSizeEstimator.estimateComplete(value,
Integer.MAX_VALUE, 20);
+ Assertions.assertTrue(full.getBytes() >= walked, full.getBytes() + " <
" + walked);
+ Assertions.assertTrue(full.getBytes() <= walked * 2, full.getBytes() +
" > 2 * " + walked);
+ double growth = (double) full.getBytes() / half.getBytes();
+ Assertions.assertTrue(growth > 1.8 && growth < 2.2, "doubling the
columns grew the estimate " + growth);
+
+ // A plugin table's display columns reuse its column objects; they
must not be weighed twice. MUTATION:
+ // estimating every element of every column list -> about twice the
weight -> red.
+ MetaCacheSizeEstimator<SchemaCacheKey, PluginDrivenSchemaCacheValue>
pluginEstimator =
+
ExternalCatalogMetaCache.sizeEstimator(PluginDrivenSchemaCacheValue.class);
+ long sameList = pluginEstimator.estimate(key, new
PluginDrivenSchemaCacheValue(columns,
+ Collections.emptyList(), Collections.emptyList(), null, null,
null, columns)).getBytes();
+ long copiedList = pluginEstimator.estimate(key, new
PluginDrivenSchemaCacheValue(columns,
+ Collections.emptyList(), Collections.emptyList(), null, null,
null, new ArrayList<>(columns)))
+ .getBytes();
+ Assertions.assertTrue(copiedList < sameList * 1.1, sameList + " -> " +
copiedList);
+ }
+
+ private static List<Column> wideColumns(int intColumns, int structColumns)
{
+ List<Column> columns = new ArrayList<>(intColumns + structColumns);
+ for (int i = 0; i < intColumns; i++) {
+ Column column = new Column("int_" + i, PrimitiveType.INT);
+ column.setComment("comment " + i);
+ columns.add(column);
+ }
+ for (int i = 0; i < structColumns; i++) {
+ columns.add(new Column("struct_" + i, new StructType(
+ new StructField("a",
ScalarType.createType(PrimitiveType.INT)),
+ new StructField("b", ScalarType.createStringType()))));
+ }
+ return columns;
+ }
+
+ private static final class WideSchemaMetaCache extends
ExternalCatalogMetaCache {
+ private WideSchemaMetaCache(ExecutorService refreshExecutor,
Supplier<SchemaCacheValue> loader) {
+ super("test_engine", refreshExecutor);
+ registerEntry(MetaCacheEntryDef.of(
+ "schema",
+ SchemaCacheKey.class,
+ SchemaCacheValue.class,
+ key -> loader.get(),
+ CacheSpec.of(true, CacheSpec.CACHE_NO_TTL, 10L),
+
MetaCacheEntryInvalidation.forNameMapping(SchemaCacheKey::getNameMapping)));
+ }
+ }
+
private static final class TestExternalMetaCache extends
ExternalCatalogMetaCache {
private TestExternalMetaCache(ExecutorService refreshExecutor) {
super("test_engine", refreshExecutor);
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
