github-actions[bot] commented on code in PR #66717:
URL: https://github.com/apache/doris/pull/66717#discussion_r3800214550
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/PaimonExternalMetaCache.java:
##########
@@ -86,41 +111,164 @@ public Table getPaimonTable(NameMapping nameMapping) {
public PaimonSnapshotCacheValue getSnapshotCache(ExternalTable dorisTable)
{
NameMapping nameMapping = dorisTable.getOrBuildNameMapping();
- return
tableEntry.get(nameMapping.getCtlId()).get(nameMapping).getLatestSnapshotCacheValue();
+ MetaCacheEntry<NameMapping, PaimonTableCacheValue> tables =
tableEntry.get(nameMapping.getCtlId());
+ PaimonTableCacheValue tableValue = tables.get(nameMapping);
+ PaimonSnapshot fence = loadLatestSnapshotFence(nameMapping,
tableValue.getPaimonTable()).getSnapshot();
+ if (!tables.isEffectivelyEnabled()) {
+ // Projections are keyed by the synthetic generation of a
published table handle. An
+ // ineffective table entry publishes nothing, so nothing keyed by
this load could ever
+ // be looked up again: serve it directly instead of churning the
snapshot entry.
+ return executeAuthenticated(nameMapping,
+ () -> latestSnapshotProjectionLoader.loadAtFence(
+ nameMapping, fence, tableValue.getGeneration()));
+ }
+ // Order fence observations, not snapshot ids: a rollback moves the
latest snapshot
+ // backwards, and a concurrent call may finish after a later
observation (reversed
+ // completion). Either way the most recently observed fence is the one
future lookups read.
+ long observation = fenceObservations.incrementAndGet();
+ PaimonSnapshotEntryKey key = PaimonSnapshotEntryKey.of(
+ nameMapping, fence, tableValue.getGeneration());
+ MetaCacheEntry<PaimonSnapshotEntryKey, PaimonSnapshotCacheValue> entry
=
+ snapshotEntry.get(nameMapping.getCtlId());
+ AtomicBoolean loaded = new AtomicBoolean();
+ PaimonSnapshotCacheValue snapshotValue = entry.get(key,
+ ignored -> executeAuthenticated(nameMapping, () -> {
+ loaded.set(true);
+ return latestSnapshotProjectionLoader.loadAtFence(
+ nameMapping, fence, tableValue.getGeneration());
+ }));
+ LatestFenceOwner owner = new LatestFenceOwner(nameMapping,
tableValue.getGeneration());
+ ObservedFence latest = latestObservedFences.compute(owner, (ignored,
current) ->
Review Comment:
[P2] Remove owners for generations that were never published
`tables.get()` still returns the loaded table when weighted admission
rejects it, and every such load has a fresh synthetic generation. This compute
therefore retains a distinct owner on each lookup; the later current-generation
check only invalidates the snapshot child, while a rejected generation gets
neither a replacement nor a removal callback that could retire the owner. The
same ordering can resurrect an old owner when a blocked load resumes after
replacement or catalog invalidation already performed cleanup. Persistently
oversized or unsupported tables can thus grow `latestObservedFences` without
any cache/budget bound. Please conditionally remove the published `(owner,
latest)` when the generation is not current, and cover both repeated rejection
and a delayed old-generation load with an owner-cardinality assertion.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/metacache/AbstractExternalMetaCache.java:
##########
@@ -124,10 +228,10 @@ public <K, V> MetaCacheEntry<K, V> entry(long catalogId,
String entryName, Class
}
@Override
- public void invalidateCatalog(long catalogId) {
+ public synchronized void invalidateCatalog(long catalogId) {
Review Comment:
[P2] Do not close a catalog while holding shared lifecycle locks
`removed.close()` synchronously invalidates every entry and releases every
reservation in the catalog, but it runs under both the engine-wide
`AbstractExternalMetaCache` monitor and the manager's lifecycle stripe. There
are only 64 stripes, so dropping or changing one large catalog blocks first
use/reinitialization of every catalog for that engine and also unrelated,
possibly different-engine catalogs whose IDs collide on the stripe. Please
replace the shared stripe with genuinely per-catalog lifecycle state, detach
under the engine monitor, then close after releasing the engine-global lock
while retaining only that catalog's fence; alternatively make budget ownership
generation-aware before allowing old/new groups to coexist. A test can block
one close and verify progress for a second same-engine ID and a
stripe-colliding ID.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/PaimonUtil.java:
##########
@@ -186,13 +188,19 @@ public static PaimonPartitionInfo
generatePartitionInfo(Table table, List<Column
List<String> partitionValues =
Lists.newArrayListWithExpectedSize(partitionColumns.size());
LinkedHashMap<String, String> orderedTypedSpec = new
LinkedHashMap<>();
+ retainedPayloadBytes =
MetaCacheWeightUtils.saturatedAdd(retainedPayloadBytes,
+
PaimonPartitionInfo.partitionColumnBytes(partitionColumns.size()));
for (Column partitionColumn : partitionColumns) {
String partitionColumnName = partitionColumn.getName();
Preconditions.checkState(typedSpec.containsKey(partitionColumnName),
"Partition column not found in Paimon typed spec: " +
partitionColumnName);
String partitionValue = typedSpec.get(partitionColumnName);
partitionValues.add(partitionValue);
orderedTypedSpec.put(partitionColumnName, partitionValue);
+ retainedPayloadBytes =
PaimonPartitionInfo.addRetainedStringPayload(
Review Comment:
[P2] Count shared partition-column names once
`partitionColumnName` is the stable string held by the schema `Column`, and
that same reference is inserted into every partition's `orderedTypedSpec`.
Accumulating its full string payload inside the outer partition loop therefore
charges N copies even though the retained graph has one string plus N map
references (whose structural cost is already charged separately). Large
partition sets can exceed an entry/catalog limit solely because of this
overcount and remain permanently cold. The current JOL/benchmark fixtures hide
the mismatch by constructing fresh field-name strings per partition. Please
charge these schema names once outside the loop (while retaining per-partition
entry costs) and calibrate with shared production-style name identities.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/metacache/ExternalMetaCacheBudgetManager.java:
##########
@@ -0,0 +1,587 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.datasource.metacache;
+
+import org.apache.doris.common.Config;
+
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.OptionalLong;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.function.LongUnaryOperator;
+import java.util.stream.Collectors;
+
+/**
+ * FE-wide admission accounting for managed external metadata caches.
+ *
+ * <p>All changes are serialized by one short critical section. Cache loads and
+ * estimators run outside it, so the lock only protects a few arithmetic and
map
+ * operations while making global/catalog/entry reservation atomic.
+ */
+public final class ExternalMetaCacheBudgetManager {
+ private static final Logger LOG =
LogManager.getLogger(ExternalMetaCacheBudgetManager.class);
+ private static final ExecutorService PEER_RECLAIM_EXECUTOR =
Executors.newSingleThreadExecutor(runnable -> {
+ Thread thread = new Thread(runnable,
"external-meta-cache-peer-reclaim");
+ thread.setDaemon(true);
+ return thread;
+ });
+
+ public static final String CATALOG_MAX_WEIGHT_PROPERTY =
"meta.cache.max-weight";
+
+ private final Object lock = new Object();
+ private final OptionalLong globalMaxWeight;
+ private final Map<Long, Bucket> catalogBuckets = new HashMap<>();
+ private final Map<EntryScope, Bucket> entryBuckets = new HashMap<>();
+ private final Map<EntryScope, EntryBudget> entryBudgets = new HashMap<>();
+ private long globalUsedWeight;
+ private final AtomicLong globalRejectedCount = new AtomicLong();
+
+ public ExternalMetaCacheBudgetManager(OptionalLong globalMaxWeight) {
+ this.globalMaxWeight = Objects.requireNonNull(globalMaxWeight,
"globalMaxWeight");
+ if (globalMaxWeight.isPresent() && globalMaxWeight.getAsLong() <= 0) {
+ throw new IllegalArgumentException("global max weight must be
positive when enabled");
+ }
+ }
+
+ public static ExternalMetaCacheBudgetManager fromConfig() {
+ String configured = Config.external_meta_cache_max_weight;
+ long parsed = CacheSpec.parseWeight(
+ configured,
+ "external_meta_cache_max_weight",
+ true,
+ Runtime.getRuntime().maxMemory());
+ if (configured.trim().endsWith("%") && parsed == 0L) {
+ throw new IllegalArgumentException(
+ "external_meta_cache_max_weight percentage must be greater
than 0%");
+ }
+ return new ExternalMetaCacheBudgetManager(parsed == 0L ?
OptionalLong.empty() : OptionalLong.of(parsed));
+ }
+
+ public OptionalLong parseCatalogMaxWeight(Map<String, String>
catalogProperties) {
+ String configured = catalogProperties.get(CATALOG_MAX_WEIGHT_PROPERTY);
+ if (configured == null) {
+ return OptionalLong.empty();
+ }
+ long parsed = CacheSpec.parseWeight(configured,
CATALOG_MAX_WEIGHT_PROPERTY, false, 0L);
+ if (parsed <= 0) {
+ throw new IllegalArgumentException(CATALOG_MAX_WEIGHT_PROPERTY + "
must be positive");
+ }
+ return OptionalLong.of(parsed);
+ }
+
+ /** Validate a catalog limit at DDL time against this FE's configured
global bound. */
+ public OptionalLong validateCatalogMaxWeight(Map<String, String>
catalogProperties) {
+ OptionalLong catalogMaxWeight =
parseCatalogMaxWeight(catalogProperties);
+ validateHierarchy(catalogMaxWeight, OptionalLong.empty());
+ return catalogMaxWeight;
+ }
+
+ /**
+ * Create the budget handle used by one physical per-catalog cache entry.
+ */
+ public EntryBudget createEntryBudget(long catalogId, String engine, String
entryName,
+ OptionalLong catalogMaxWeight, OptionalLong entryMaxWeight) {
+ Objects.requireNonNull(engine, "engine");
+ Objects.requireNonNull(entryName, "entryName");
+ Objects.requireNonNull(catalogMaxWeight, "catalogMaxWeight");
+ Objects.requireNonNull(entryMaxWeight, "entryMaxWeight");
+ validateCatalogEntryHierarchy(catalogMaxWeight, entryMaxWeight);
+
+ OptionalLong effectiveMax = minimumPresent(globalMaxWeight,
catalogMaxWeight, entryMaxWeight);
+ if (!effectiveMax.isPresent()) {
+ throw new IllegalArgumentException("entry budget requires at least
one configured weight bound");
+ }
+
+ EntryScope scope = new EntryScope(catalogId, engine, entryName);
+ synchronized (lock) {
+ Bucket catalogBucket = catalogBuckets.get(catalogId);
+ long catalogLimit = minimumLimit(globalMaxWeight,
catalogMaxWeight);
+ if (catalogBucket == null) {
+ catalogBucket = new Bucket(catalogLimit);
+ catalogBuckets.put(catalogId, catalogBucket);
+ } else if (catalogBucket.maxWeight != catalogLimit) {
+ throw new IllegalStateException("Conflicting catalog cache max
weight for catalog " + catalogId);
+ }
+
+ if (entryBuckets.containsKey(scope)) {
+ throw new IllegalStateException("Duplicated external meta
cache budget: " + scope);
+ }
+ Bucket entryBucket = new Bucket(effectiveMax.getAsLong());
+ EntryBudget entryBudget = new EntryBudget(
+ this, scope, catalogBucket, entryBucket,
effectiveMax.getAsLong());
+ entryBuckets.put(scope, entryBucket);
+ entryBudgets.put(scope, entryBudget);
+ return entryBudget;
+ }
+ }
+
+ public OptionalLong getGlobalMaxWeight() {
+ return globalMaxWeight;
+ }
+
+ public long getGlobalUsedWeight() {
+ synchronized (lock) {
+ return globalUsedWeight;
+ }
+ }
+
+ public long getGlobalRejectedCount() {
+ return globalRejectedCount.get();
+ }
+
+ public void validateHierarchy(OptionalLong catalogMaxWeight, OptionalLong
entryMaxWeight) {
+ if (globalMaxWeight.isPresent() && catalogMaxWeight.isPresent()
+ && catalogMaxWeight.getAsLong() > globalMaxWeight.getAsLong())
{
+ throw new IllegalArgumentException(CATALOG_MAX_WEIGHT_PROPERTY + "
can not exceed FE global max weight");
+ }
+ OptionalLong parent = catalogMaxWeight.isPresent() ? catalogMaxWeight
: globalMaxWeight;
+ if (parent.isPresent() && entryMaxWeight.isPresent()
+ && entryMaxWeight.getAsLong() > parent.getAsLong()) {
+ throw new IllegalArgumentException("entry max weight can not
exceed its parent max weight");
+ }
+ }
+
+ /**
+ * Validate persisted catalog-to-entry hierarchy without comparing it with
this FE's local
+ * global bound. Catalog properties are validated on the master, while the
global percentage
+ * is resolved independently from each FE's heap. Runtime admission
therefore clamps to the
+ * local global limit instead of rejecting a catalog accepted on a larger
master.
+ */
+ public void validateCatalogEntryHierarchy(OptionalLong catalogMaxWeight,
OptionalLong entryMaxWeight) {
+ OptionalLong parent = catalogMaxWeight;
+ if (parent.isPresent() && entryMaxWeight.isPresent()
+ && entryMaxWeight.getAsLong() > parent.getAsLong()) {
+ throw new IllegalArgumentException("entry max weight can not
exceed its parent max weight");
+ }
+ }
+
+ private Optional<AdmissionReservation> tryReserve(EntryBudget entryBudget,
long bytes) {
+ checkWeight(bytes);
+ synchronized (lock) {
+ if (entryBudget.closed) {
+ return Optional.empty();
+ }
+ if (!fits(limitOf(globalMaxWeight), globalUsedWeight, bytes)
+ || !fits(entryBudget.catalogBucket.maxWeight,
entryBudget.catalogBucket.usedWeight, bytes)
+ || !fits(entryBudget.entryBucket.maxWeight,
entryBudget.entryBucket.usedWeight, bytes)) {
+ entryBudget.rejectedCount.incrementAndGet();
+ globalRejectedCount.incrementAndGet();
+ return Optional.empty();
+ }
+ addUsed(entryBudget, bytes);
+ return Optional.of(new AdmissionReservation(this, entryBudget,
bytes));
+ }
+ }
+
+ private boolean resize(AdmissionReservation reservation, long newBytes) {
+ checkWeight(newBytes);
+ synchronized (lock) {
+ if (!reservation.active || reservation.entryBudget.closed) {
+ return false;
+ }
+ long delta = newBytes - reservation.bytes;
+ if (delta > 0 && (!fits(limitOf(globalMaxWeight),
globalUsedWeight, delta)
+ || !fits(reservation.entryBudget.catalogBucket.maxWeight,
+ reservation.entryBudget.catalogBucket.usedWeight,
delta)
+ || !fits(reservation.entryBudget.entryBucket.maxWeight,
+ reservation.entryBudget.entryBucket.usedWeight,
delta))) {
+ reservation.entryBudget.rejectedCount.incrementAndGet();
+ globalRejectedCount.incrementAndGet();
+ return false;
+ }
+ if (delta >= 0) {
+ addUsed(reservation.entryBudget, delta);
+ } else {
+ subtractUsed(reservation.entryBudget, -delta);
+ }
+ reservation.bytes = newBytes;
+ return true;
+ }
+ }
+
+ private void release(AdmissionReservation reservation) {
+ synchronized (lock) {
+ if (!reservation.active) {
+ return;
+ }
+ if (reservation.entryBudget.closed) {
+ reservation.bytes = 0L;
+ reservation.active = false;
+ return;
+ }
+ subtractUsed(reservation.entryBudget, reservation.bytes);
+ reservation.bytes = 0L;
+ reservation.active = false;
+ }
+ }
+
+ private void close(EntryBudget entryBudget) {
+ synchronized (lock) {
+ if (entryBudget.closed) {
+ return;
+ }
+ if (entryBudget.entryBucket.usedWeight != 0L) {
+ long leakedWeight = entryBudget.entryBucket.usedWeight;
+ LOG.error("Force-closing external metadata cache budget {}
with {} bytes still reserved",
+ entryBudget.scope, leakedWeight);
+ if (leakedWeight <= globalUsedWeight
+ && leakedWeight <=
entryBudget.catalogBucket.usedWeight) {
+ globalUsedWeight -= leakedWeight;
+ entryBudget.catalogBucket.usedWeight -= leakedWeight;
+ entryBudget.entryBucket.usedWeight = 0L;
+ } else {
+ LOG.error("External metadata cache accounting is
inconsistent while closing {}; "
+ + "globalUsed={}, catalogUsed={},
entryUsed={}",
+ entryBudget.scope, globalUsedWeight,
+ entryBudget.catalogBucket.usedWeight,
leakedWeight);
+ globalUsedWeight = Math.max(0L, globalUsedWeight -
leakedWeight);
+ entryBudget.catalogBucket.usedWeight = Math.max(
+ 0L, entryBudget.catalogBucket.usedWeight -
leakedWeight);
+ entryBudget.entryBucket.usedWeight = 0L;
+ }
+ }
+ entryBudget.closed = true;
+ entryBudget.reclaimer = null;
+ entryBuckets.remove(entryBudget.scope, entryBudget.entryBucket);
+ entryBudgets.remove(entryBudget.scope, entryBudget);
+ Bucket catalogBucket = entryBudget.catalogBucket;
+ boolean catalogStillReferenced = entryBuckets.keySet().stream()
+ .anyMatch(scope -> scope.catalogId ==
entryBudget.scope.catalogId);
+ if (!catalogStillReferenced && catalogBucket.usedWeight == 0L) {
+ catalogBuckets.remove(entryBudget.scope.catalogId,
catalogBucket);
+ }
+ }
+ }
+
+ private void addUsed(EntryBudget entryBudget, long bytes) {
+ globalUsedWeight += bytes;
+ entryBudget.catalogBucket.usedWeight += bytes;
+ entryBudget.entryBucket.usedWeight += bytes;
+ }
+
+ private void subtractUsed(EntryBudget entryBudget, long bytes) {
+ if (bytes > globalUsedWeight
+ || bytes > entryBudget.catalogBucket.usedWeight
+ || bytes > entryBudget.entryBucket.usedWeight) {
+ throw new IllegalStateException("external meta cache budget
accounting underflow");
+ }
+ globalUsedWeight -= bytes;
+ entryBudget.catalogBucket.usedWeight -= bytes;
+ entryBudget.entryBucket.usedWeight -= bytes;
+ }
+
+ private void requestPeerReclaim(EntryBudget requester, long
additionalBytes) {
+ if (additionalBytes <= 0L || requester.closed) {
+ return;
+ }
+ long reclaimBytes;
+ synchronized (lock) {
+ if (requester.closed) {
+ return;
+ }
+ long globalDeficit = deficit(limitOf(globalMaxWeight),
globalUsedWeight, additionalBytes);
+ long catalogDeficit = deficit(
+ requester.catalogBucket.maxWeight,
requester.catalogBucket.usedWeight, additionalBytes);
+ reclaimBytes = Math.max(globalDeficit, catalogDeficit);
+ }
+ if (reclaimBytes <= 0L) {
+ return;
+ }
+ // Rejected values are returned uncached; there is no queue of pending
admissions to fund.
+ // Coalesce concurrent misses to the largest single admission instead
of summing identical
+ // deficits and evicting an entire peer cache during a miss burst.
+ requester.requestedAdmissionBytes.accumulateAndGet(additionalBytes,
Math::max);
+ schedulePeerReclaim(requester);
+ }
+
+ private void schedulePeerReclaim(EntryBudget requester) {
+ if (!requester.reclaimScheduled.compareAndSet(false, true)) {
+ return;
+ }
+ try {
+ PEER_RECLAIM_EXECUTOR.execute(() -> drainPeerReclaim(requester));
+ } catch (RejectedExecutionException e) {
+ requester.reclaimScheduled.set(false);
+ LOG.warn("Failed to schedule peer reclamation for external
metadata cache budget {}",
+ requester.scope, e);
+ }
+ }
+
+ private void drainPeerReclaim(EntryBudget requester) {
+ try {
+ long requestedAdmissionBytes =
requester.requestedAdmissionBytes.getAndSet(0L);
+ if (requestedAdmissionBytes <= 0L || requester.closed) {
+ return;
+ }
+ List<EntryBudget> candidates;
+ synchronized (lock) {
Review Comment:
[P2] Sort reclaim candidates outside the global budget lock
`drainPeerReclaim` holds the one FE-wide accounting lock while filtering and
sorting every active `EntryBudget`. That same lock gates every reserve, resize,
release, budget create/close, and stats operation, so with many initialized
catalogs a pressure-triggered reclaim becomes an O(E log E) global pause for
unrelated cache traffic. Please snapshot the candidate/order fields under the
lock, sort outside it, revalidate current/closed state, and recompute the
current deficit before invoking each reclaimer. A many-budget concurrency test
should pause candidate ordering and prove an unrelated reserve/release can
still complete.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
