github-actions[bot] commented on code in PR #66717:
URL: https://github.com/apache/doris/pull/66717#discussion_r3836886774
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/ExternalMetaCacheMgr.java:
##########
@@ -227,16 +390,84 @@ public void invalidateCatalogByEngine(long catalogId,
String engine) {
() -> cache.invalidateCatalogEntries(catalogId)));
}
+ /**
+ * Run an action under the same per-catalog lifecycle fence that guards
lazy initialization
+ * and group retirement. The stripe lock is reentrant, so fenced actions
may call back into
+ * removeCatalog/rollbackCatalogProperties for the same catalog.
+ */
+ public <T> T withCatalogLifecycleLock(long catalogId,
java.util.function.Supplier<T> action) {
+ Lock lifecycleLock = catalogLifecycleLocks.get(catalogId);
+ lifecycleLock.lock();
+ try {
+ return action.get();
+ } finally {
+ lifecycleLock.unlock();
+ }
+ }
+
public void removeCatalog(long catalogId) {
- routeCatalogEngines(catalogId, cache -> safeInvalidate(
- cache, catalogId, "removeCatalog",
- () -> cache.invalidateCatalog(catalogId)));
+ Lock lifecycleLock = catalogLifecycleLocks.get(catalogId);
+ lifecycleLock.lock();
+ try {
+ routeCatalogEngines(catalogId, cache -> safeInvalidate(
+ cache, catalogId, "removeCatalog",
+ () -> cache.invalidateCatalog(catalogId)));
+ } finally {
+ lifecycleLock.unlock();
+ }
+ }
+
+ /**
+ * DROP CATALOG (or its replay): the id is never reused. Retires the cache
groups like
+ * {@link #removeCatalog} and additionally releases engine side state that
must survive
+ * same-id policy rebuilds; the hook reaches every engine even when its
entry group is
+ * already retired.
+ */
+ public void removeCatalogPermanently(long catalogId) {
+ Lock lifecycleLock = catalogLifecycleLocks.get(catalogId);
+ lifecycleLock.lock();
+ try {
+ routeCatalogEngines(catalogId, cache -> safeInvalidate(
+ cache, catalogId, "removeCatalogPermanently",
+ () -> cache.invalidateCatalog(catalogId)));
+ for (ExternalMetaCache cache : cacheRegistry.allCaches()) {
Review Comment:
[P2] Publish the permanent-drop state before closing routed groups.
`CatalogMgr` has already removed the catalog, and `invalidateCatalog()`
detaches an engine group before synchronously closing it. While that close (or
another engine's close) is blocked, a retained lookup sees no group and no
catalog but still has no tombstone, so it sleeps/retries for the full
two-second handoff window. This is the ordering gap left in the terminal-retry
fix. Please mark DROP terminal before detachment/close, while keeping any
post-close engine cleanup separately fenced, and latch-test a lookup during a
blocked close.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergExternalMetaCache.java:
##########
@@ -108,13 +136,141 @@ public Table getIcebergTable(ExternalTable dorisTable) {
return
tableEntry.get(nameMapping.getCtlId()).get(nameMapping).getIcebergTable();
}
+ public Table getWritableIcebergTable(ExternalTable dorisTable) {
+ return getWritableIcebergTable(dorisTable, null);
+ }
+
+ /**
+ * Acquire a writable table for a caller that retains the {@code
IcebergMetadataOps} of the
+ * generation its operation was dispatched on (a DDL method of that ops
instance, a DML
+ * transaction). The acquisition fails when the catalog has moved to a
different generation,
+ * so the caller's later updates - executed through its retained ops and
authenticator -
+ * can never operate a newer generation's handle.
+ */
+ public Table getWritableIcebergTable(ExternalTable dorisTable, @Nullable
IcebergMetadataOps expectedOps) {
+ NameMapping nameMapping = dorisTable.getOrBuildNameMapping();
+ CatalogIf catalog = getCatalog(nameMapping.getCtlId());
+ if (catalog == null) {
+ throw new RuntimeException("Cannot find catalog " +
nameMapping.getCtlId()
+ + " when loading a writable Iceberg table");
+ }
+ // DDL/actions must start from the live catalog generation. DML that
was planned against a
+ // retained read generation wraps this live table separately in
IcebergTransaction. The
+ // authenticator, ops and loaded handle must all come from that one
generation, so the
+ // acquisition is re-validated afterwards: a concurrent
property/credential ALTER can reset
+ // and reinitialize the catalog mid-flight without retiring this
engine's cache group.
+ ExecutionAuthenticator authenticator =
requireExecutionAuthenticator(catalog);
+ IcebergMetadataOps ops = resolveMetadataOps(catalog);
+ if (expectedOps != null && ops != expectedOps) {
+ throw catalogGenerationMoved(nameMapping);
+ }
+ Table table = execute(authenticator, () -> ops.loadTable(
+ nameMapping.getRemoteDbName(),
nameMapping.getRemoteTblName()));
+ ensureCatalogGenerationStable(catalog, ops, authenticator,
nameMapping);
+ return table;
+ }
+
+ Table getQueryScopedIcebergTable(ExternalTable dorisTable) {
+ NameMapping nameMapping = dorisTable.getOrBuildNameMapping();
+ MetaCacheEntry<NameMapping, IcebergTableCacheValue> entry =
+ tableEntry.get(nameMapping.getCtlId());
+ IcebergTableCacheValue tableValue =
+ entry.get(nameMapping);
+ return createQueryTable(nameMapping, tableValue);
+ }
+
+ /** Resolve the current table generation, exposing the handle and its
captured context together. */
+ IcebergTableCacheValue getTableCacheValue(ExternalTable dorisTable) {
+ NameMapping nameMapping = dorisTable.getOrBuildNameMapping();
+ return tableEntry.get(nameMapping.getCtlId()).get(nameMapping);
+ }
+
+ /** Query-scoped view of an already-resolved generation; see {@link
#getTableCacheValue}. */
+ Table createQueryScopedTable(ExternalTable dorisTable,
IcebergTableCacheValue tableValue) {
+ return createQueryTable(dorisTable.getOrBuildNameMapping(),
tableValue);
+ }
+
+ private Table createQueryTable(
+ NameMapping nameMapping, IcebergTableCacheValue tableValue) {
+ boolean isolateForQueries = tableValue.isQueryIsolationPrepared()
+ ||
snapshotEntry.get(nameMapping.getCtlId()).isWeightAccounting();
+ if (!isolateForQueries) {
+ return tableValue.getIcebergTable();
+ }
+ Table queryTable = tableValue.newQueryScopedTable();
+ IcebergSnapshotCacheValue.loadQueryMetadataForStatement(queryTable);
+ return queryTable;
+ }
+
public IcebergSnapshotCacheValue getSnapshotCache(ExternalTable
dorisTable) {
NameMapping nameMapping = dorisTable.getOrBuildNameMapping();
- return
tableEntry.get(nameMapping.getCtlId()).get(nameMapping).getLatestSnapshotCacheValue();
+ IcebergTableCacheValue tableValue =
+ tableEntry.get(nameMapping.getCtlId()).get(nameMapping);
Review Comment:
[P1] Retire the cached base generation when an operational catalog property
changes. Every property update resets the catalog authenticator and closes its
SDK resources, but credential/storage keys do not remove the Iceberg or Paimon
cache group. A brand-new Iceberg statement can therefore get the admitted A1
table here, accept its matching A1 projection, and then fail the A2 planning
fence; retries keep hitting A1 until managed refresh is due. Paimon's memoized
latest path can keep returning the old closed generation without a live-context
check. Please retire the affected engine group (or synchronously
validate/reload its base) on execution-context changes, and test an immediate
post-ALTER statement without manually publishing A2 first.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/PaimonExternalMetaCache.java:
##########
@@ -86,41 +114,315 @@ 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);
+ if (tables.isEffectivelyEnabled()) {
+ // Serve the memoized latest projection of this table generation
while it is still
+ // published: the latest read is as stale-until-TTL/refresh as the
cached table
+ // handle itself and costs no snapshot IO, preserving the
pre-existing external
+ // metadata cache contract. The fence is re-observed only when no
projection of this
+ // generation is reachable anymore (first read, expiry, weight
eviction, explicit
+ // invalidation), which is also when rollback ordering below
matters.
+ ObservedFence observed = latestObservedFences.get(
+ new LatestFenceOwner(nameMapping,
tableValue.getGeneration()));
+ if (observed != null) {
+ PaimonSnapshotCacheValue memoized =
+
snapshotEntry.get(nameMapping.getCtlId()).peekIfPresent(observed.key);
+ if (memoized != null) {
+ return memoized;
+ }
+ }
+ }
+ PaimonSnapshot fence = loadLatestSnapshotFence(nameMapping,
tableValue).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 executeForGeneration(tableValue, nameMapping,
+ () -> latestSnapshotProjectionLoader.loadAtFence(
+ nameMapping, fence, tableValue.getGeneration()))
+ .bindCapturedAuthenticator(tableValue.getAuthenticator());
+ }
+ // 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();
Review Comment:
[P2] Assign this ordering at the fence-capture boundary. A can capture fence
8 at line 136 and pause before this increment; B can then capture fence 9, get
observation N, and publish it; when A resumes it gets N+1, replaces B in
`latestObservedFences`, and retires the fence-9 projection. Subsequent memoized
latest reads then return the older fence 8. The reversed-completion test pauses
during projection enumeration, after this number is assigned, so it misses this
window. Please serialize capture plus sequence assignment per owner and add the
capture-before-counter interleaving.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/metacache/MetaCacheEntry.java:
##########
@@ -198,62 +552,738 @@ public MetaCacheEntryStats stats() {
failureCount,
totalLoadTime,
totalLoadCount == 0 ? 0D : (double) totalLoadTime /
totalLoadCount,
- cacheStats.evictionCount(),
+ MetaCacheWeightUtils.saturatedAdd(
+ cacheStats.evictionCount(), localEvictionCount.get()),
invalidateCount.get(),
lastLoadSuccessTimeMs.get(),
lastLoadFailureTimeMs.get(),
- lastError.get());
+ lastError.get(),
+ weightBounded,
+ weightBounded ? cacheSpec.getMaxWeight().getAsLong() : -1L,
+ weightBounded ? entryBudget.getUsedWeight() : -1L,
+ weightBounded ? MetaCacheWeightUtils.saturatedAdd(
+ automaticEvictionWeight.get(),
localEvictionWeight.get()) : -1L,
+ weightBounded ? weightAdmissionRejectedCount.get() : -1L,
+ weightBounded ? entryBudget.getCatalogMaxWeight() : -1L,
+ weightBounded ? entryBudget.getCatalogUsedWeight() : -1L,
+ weightBounded ? entryBudget.getGlobalMaxWeight() : -1L,
+ weightBounded ? entryBudget.getGlobalUsedWeight() : -1L,
+ weightBounded ? lastWeightRejectReason.get() : "");
+ }
+
+ public boolean isWeightBounded() {
+ return weightBounded;
+ }
+
+ /** True when this entry stores values at all (enabled with a positive
capacity or weight). */
+ public boolean isEffectivelyEnabled() {
+ return effectiveEnabled;
+ }
+
+ /**
+ * True when publication sizing can lead to a weighted admission: an entry
may be configured
+ * with a weight bound yet be ineffective (max-weight 0, disabled, zero
TTL or capacity), in
+ * which case preparing values for publication is pure waste.
+ */
+ public boolean isWeightAccounting() {
+ return weightBounded && effectiveEnabled;
+ }
+
+ private AdmissionResult admitWeightedValue(
+ K key, V value, @Nullable V expectedCurrent, boolean
requireExpected,
+ @Nullable KeyMutationToken expectedMutation, long
expectedReservationGeneration,
+ boolean advanceMutationOnAdmission) {
+ if (closed.get()) {
+ return AdmissionResult.DISABLED;
+ }
+ MetaCacheSizeEstimate estimate;
+ try {
+ estimate = Objects.requireNonNull(sizeEstimator.estimate(key,
value), "size estimate");
+ } catch (IllegalArgumentException e) {
+ throw e;
+ } catch (RuntimeException e) {
+ rejectWeight("invalid_estimate");
+ return AdmissionResult.REJECTED;
+ }
+ if (!estimate.isComplete()) {
+ rejectWeight(estimate.getIncompleteReason());
+ return AdmissionResult.REJECTED;
+ }
+
+ long estimatedPayloadBytes = estimate.getBytes();
+ // A retained non-null key/value plus Caffeine node can never consume
zero bytes. Treat a
+ // complete zero as an estimator contract violation so an omitted
formula cannot bypass
+ // every quota and admit an unbounded number of zero-weight entries.
+ if (estimatedPayloadBytes == 0L) {
+ rejectWeight("invalid_estimate");
+ return AdmissionResult.REJECTED;
+ }
+ long newWeight = MetaCacheWeightUtils.saturatedAdd(
+ estimatedPayloadBytes, FIXED_ENTRY_ACCOUNTING_OVERHEAD_BYTES);
+ synchronized (admissionLock) {
+ if (closed.get()) {
+ return AdmissionResult.DISABLED;
+ }
+ if (expectedMutation != null && !isKeyMutationCurrent(key,
expectedMutation)) {
+ return AdmissionResult.NOT_CURRENT;
+ }
+ V oldValue = data.asMap().get(key);
+ ReservationRecord record = reservations.get(key);
+ if (expectedReservationGeneration >= 0L
+ && (record == null || record.generation !=
expectedReservationGeneration)) {
+ return AdmissionResult.NOT_CURRENT;
+ }
+ if (requireExpected && oldValue != expectedCurrent) {
+ return AdmissionResult.NOT_CURRENT;
+ }
+ if (oldValue == null && record != null) {
+ // The previous generation was already removed by Caffeine; if
that removal was
+ // an eviction whose asynchronous cleanup has not run yet,
account it here.
+ if (pendingEvictionGenerations.remove(key, record.generation))
{
+ automaticEvictionWeight.accumulateAndGet(
+ record.weight, MetaCacheWeightUtils::saturatedAdd);
+ }
+ reservations.remove(key, record);
+ record.reservation.release();
+ record = null;
+ }
+ if (oldValue != null && (record == null || !record.published)) {
+ rejectWeight("missing_reservation");
+ return AdmissionResult.REJECTED;
+ }
+
+ if (record == null) {
+ Optional<AdmissionReservation> reservation =
reserveWithLocalEviction(key, newWeight);
+ if (!reservation.isPresent()) {
+ rejectWeight("budget_exceeded");
+ return AdmissionResult.REJECTED;
+ }
+ ReservationRecord newRecord = new ReservationRecord(
+ newWeight, reservation.get(),
nextReservationGeneration());
+ if (advanceMutationOnAdmission) {
+ advanceKeyMutation(key);
+ }
+ reservations.put(key, newRecord);
+ try {
+ beforeWeightedCachePutForTest(key, value);
+ data.put(key, value);
+ if (reservations.get(key) == newRecord &&
data.asMap().get(key) == value) {
+ newRecord.published = true;
+ notifyReplacement(key, null, value);
+ }
+ return AdmissionResult.ADMITTED;
+ } catch (RuntimeException | Error e) {
+ reservations.remove(key, newRecord);
+ newRecord.reservation.release();
+ throw e;
+ }
+ }
+
+ ReservationRecord previousRecord = record;
+ long reservedWeight = Math.max(previousRecord.weight, newWeight);
+ if (!resizeWithLocalEviction(key, previousRecord.reservation,
reservedWeight)) {
+ rejectWeight("budget_exceeded");
+ return AdmissionResult.REJECTED;
+ }
+ if (advanceMutationOnAdmission) {
+ advanceKeyMutation(key);
+ }
+ ReservationRecord newRecord = new ReservationRecord(
+ newWeight, previousRecord.reservation,
nextReservationGeneration());
+ reservations.put(key, newRecord);
+ try {
+ beforeWeightedCachePutForTest(key, value);
+ data.put(key, value);
+ boolean retained = reservations.get(key) == newRecord &&
data.asMap().get(key) == value;
+ if (retained) {
+ newRecord.published = true;
+ }
+ if (retained && reservedWeight != newWeight &&
!newRecord.reservation.tryResize(newWeight)) {
+ throw new IllegalStateException("failed to release cache
replacement reservation delta");
+ }
+ if (retained) {
+ notifyReplacement(key, oldValue, value);
+ }
+ return AdmissionResult.ADMITTED;
+ } catch (RuntimeException | Error e) {
+ if (reservations.replace(key, newRecord, previousRecord)) {
+ if (data.asMap().get(key) == null) {
+ reservations.remove(key, previousRecord);
+ previousRecord.reservation.release();
+ } else if
(!previousRecord.reservation.tryResize(previousRecord.weight)) {
+ throw new IllegalStateException("failed to roll back
cache replacement reservation", e);
+ }
+ }
+ throw e;
+ }
+ }
+ }
+
+ private Optional<AdmissionReservation> reserveWithLocalEviction(K
incomingKey, long bytes) {
+ if (bytes > entryBudget.getEffectiveMaxWeight()) {
+ return Optional.empty();
+ }
+ Optional<AdmissionReservation> reservation =
entryBudget.tryReserve(bytes);
+ while (!reservation.isPresent()) {
+ AtomicReference<Optional<AdmissionReservation>> retried =
+ new AtomicReference<>(Optional.empty());
+ int evicted = evictLocalColdest(incomingKey,
LOCAL_EVICTION_BATCH_SIZE, () -> {
+ Optional<AdmissionReservation> attempt =
entryBudget.tryReserve(bytes);
+ retried.set(attempt);
+ return attempt.isPresent();
+ });
+ reservation = retried.get();
+ if (reservation.isPresent()) {
+ break;
+ }
+ if (evicted == 0) {
+ entryBudget.requestPeerReclaim(bytes);
+ break;
+ }
+ }
+ return reservation;
+ }
+
+ private boolean resizeWithLocalEviction(K incomingKey,
AdmissionReservation reservation, long newBytes) {
+ if (newBytes > entryBudget.getEffectiveMaxWeight()) {
+ return false;
+ }
+ if (reservation.tryResize(newBytes)) {
+ return true;
+ }
+ while (true) {
+ AtomicBoolean resized = new AtomicBoolean();
+ int evicted = evictLocalColdest(incomingKey,
LOCAL_EVICTION_BATCH_SIZE, () -> {
+ if (reservation.tryResize(newBytes)) {
+ resized.set(true);
+ return true;
+ }
+ return false;
+ });
+ if (resized.get()) {
+ return true;
+ }
+ if (evicted == 0) {
+ entryBudget.requestPeerReclaim(Math.max(0L, newBytes -
reservation.getBytes()));
+ return false;
+ }
+ }
+ }
+
+ /**
+ * Evict up to {@code limit} coldest values, retrying the caller's goal
after every single
+ * eviction: a skewed cache where one large cold value creates all needed
headroom must not
+ * discard the rest of the selected batch.
+ */
+ private int evictLocalColdest(K incomingKey, int limit, BooleanSupplier
stopAfterEviction) {
+ if (!data.policy().eviction().isPresent()) {
+ return 0;
+ }
+ Map<K, V> coldest = data.policy().eviction().get().coldest(limit);
+ int evicted = 0;
+ for (Map.Entry<K, V> candidate : coldest.entrySet()) {
+ if (Objects.equals(candidate.getKey(), incomingKey)) {
+ continue;
+ }
+ V current = data.asMap().get(candidate.getKey());
+ ReservationRecord record = reservations.get(candidate.getKey());
+ long evictedWeight = record != null && record.published && current
!= null ? record.weight : 0L;
+ if (current == candidate.getValue() &&
data.asMap().remove(candidate.getKey(), current)) {
+ if (record != null) {
+ releaseReservation(candidate.getKey(), record.generation);
+ }
+ localEvictionCount.incrementAndGet();
+ localEvictionWeight.accumulateAndGet(evictedWeight,
MetaCacheWeightUtils::saturatedAdd);
+ evicted++;
+ if (stopAfterEviction.getAsBoolean()) {
+ return evicted;
+ }
+ }
+ }
+ return evicted;
+ }
+
+ private long reclaimForPeer(long targetBytes) {
+ if (targetBytes <= 0L || closed.get()) {
+ return 0L;
+ }
+ synchronized (admissionLock) {
+ long before = entryBudget.getUsedWeight();
+ long reclaimed = 0L;
+ while (reclaimed < targetBytes
+ && evictLocalColdest(null, LOCAL_EVICTION_BATCH_SIZE, () ->
+ Math.max(0L, before - entryBudget.getUsedWeight())
>= targetBytes) > 0) {
+ reclaimed = Math.max(0L, before - entryBudget.getUsedWeight());
+ }
+ return reclaimed;
+ }
+ }
+
+ private int weigh(K key, V value) {
+ ReservationRecord record = reservations.get(key);
+ // Every supported write path installs the reservation record before
calling data.put.
+ // Missing ownership is an invariant violation, so fail closed without
invoking an O(n)
+ // estimator from Caffeine's hot weigher callback.
+ long weight = record == null ? Integer.MAX_VALUE : record.weight;
+ return weight >= Integer.MAX_VALUE ? Integer.MAX_VALUE : (int) weight;
+ }
+
+ private void onRemoval(@Nullable K key, @Nullable V value, RemovalCause
cause) {
+ if (key == null) {
+ return;
+ }
+ if (!weightBounded && !generationFencedRefresh && removalListener ==
null) {
+ return;
+ }
+ if (closed.get()) {
+ return;
+ }
+ // Replacement transfers the existing reservation to the newly
published generation. A
+ // soft-value collection instead reports a null value with COLLECTED
and must release it.
+ if (cause == RemovalCause.REPLACED) {
+ return;
+ }
+ if (removalListener != null) {
+ pendingRemovalNotifications.add(new RemovedToken<>(key,
removalToken(value)));
+ scheduleRemovalCleanup();
+ }
+ if (Thread.holdsLock(admissionLock)) {
+ // Other removals have already removed the Caffeine mapping and
can release their owner
+ // inline. A stale callback cannot release a replacement while its
mapping is visible.
+ if (data.asMap().get(key) == null) {
+ if (weightBounded) {
+ ReservationRecord record = reservations.get(key);
+ if (record != null) {
+ if (cause.wasEvicted()) {
+ automaticEvictionWeight.accumulateAndGet(
+ record.weight,
MetaCacheWeightUtils::saturatedAdd);
+ }
+ releaseReservation(key, record.generation);
+ }
+ } else {
+ RefreshRecord record = refreshRecords.get(key);
+ if (record != null) {
+ releaseRefreshRecord(key, record.generation);
+ }
+ }
+ }
+ return;
+ }
+ beforeRemovalOwnerSnapshotForTest(key);
+ long ownerGeneration = currentOwnerGeneration(key);
+ if (ownerGeneration >= 0L) {
+ beforeRemovalReleaseForTest(key);
+ if (closed.get()) {
+ return;
+ }
+ if (cause.wasEvicted()) {
+ pendingEvictionGenerations.merge(key, ownerGeneration,
Math::max);
+ }
+ pendingRemovalGenerations.merge(key, ownerGeneration, Math::max);
+ if (closed.get()) {
+ pendingRemovalGenerations.remove(key, ownerGeneration);
+ pendingEvictionGenerations.remove(key, ownerGeneration);
+ return;
+ }
+ scheduleRemovalCleanup();
+ }
+ }
+
+ private void scheduleRemovalCleanup() {
+ if (closed.get()) {
+ return;
+ }
+ if (removalCleanupScheduled.compareAndSet(false, true)) {
+ try {
+ REMOVAL_CLEANUP_EXECUTOR.execute(this::drainRemovalCleanups);
+ } catch (RejectedExecutionException e) {
+ removalCleanupScheduled.set(false);
+ LOG.warn("Failed to schedule removal cleanup for external
metadata cache entry {}", name, e);
+ }
+ }
+ }
+
+ private void drainRemovalCleanups() {
+ try {
+ drainRemovalNotifications();
Review Comment:
[P2] Release dead reservations before running dependency retirement. Every
entry shares this single cleanup executor, and `drainRemovalNotifications()`
runs first. A Paimon table notification scans and invalidates all
snapshot/schema children, so one large callback can hold this worker while
automatic removals in this and unrelated catalogs remain charged in
`pendingRemovalGenerations`. During that interval new admission can be rejected
or live peers reclaimed even though the parent values are already gone. Please
let reservation cleanup progress independently (or at least before potentially
unbounded listeners), and cover a blocked listener while an unrelated automatic
eviction releases its global/catalog quota.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/metacache/AbstractExternalMetaCache.java:
##########
@@ -222,16 +337,59 @@ protected final ExternalTable
findExternalTable(NameMapping nameMapping, String
nameMapping.getLocalTblName(), engineNameForError));
}
+ // A contended cache-policy handoff resolves within this window; see
requireCatalogEntryGroup.
+ private static final long PREPARE_RETRY_WINDOW_NANOS = 2_000_000_000L;
+ private static final long PREPARE_RETRY_SLEEP_MS = 50L;
+
private CatalogEntryGroup requireCatalogEntryGroup(long catalogId) {
CatalogEntryGroup group = catalogEntries.get(catalogId);
+ if (group == null && catalogPreparer != null) {
+ // The caller prepared the catalog before capturing this engine,
but a cache-policy
+ // ALTER retired the group in between. Re-prepare under the
lifecycle fence so the
+ // lookup observes the new policy instead of failing a valid
catalog. The preparer
+ // never blocks on the fence (a nested default loader may hold a
Caffeine bin lock
+ // that retirement itself needs), so a contended handoff is
absorbed with a bounded
+ // sleep-and-retry: the ALTER finishes within the window, or the
lookup fails as
+ // before without any deadlock.
+ long deadlineNanos = System.nanoTime() +
PREPARE_RETRY_WINDOW_NANOS;
+ while (!permanentlyRemovedCatalogs.contains(catalogId)) {
+ catalogPreparer.accept(catalogId);
+ group = catalogEntries.get(catalogId);
+ if (group != null || System.nanoTime() >= deadlineNanos
+ || permanentlyRemovedCatalogs.contains(catalogId)) {
+ break;
+ }
+ try {
+ Thread.sleep(PREPARE_RETRY_SLEEP_MS);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ break;
+ }
+ }
+ }
if (group == null) {
+ if (permanentlyRemovedCatalogs.contains(catalogId)) {
+ throw new IllegalStateException(String.format(
+ "Catalog %d was dropped; engine '%s' serves no
metadata for it.",
+ catalogId, engine));
+ }
throw new IllegalStateException(String.format(
"Catalog %d is not initialized for engine '%s'.",
catalogId, engine));
}
return group;
}
+ @Override
+ public void bindCatalogPreparer(LongConsumer catalogPreparer) {
+ this.catalogPreparer = catalogPreparer;
+ }
+
+ @Override
+ public void onCatalogPermanentlyRemoved(long catalogId) {
+ permanentlyRemovedCatalogs.add(catalogId);
Review Comment:
[P2] Avoid retaining every dropped catalog ID forever in every engine.
`removeCatalogPermanently()` calls this hook for all registered caches, even
engines that never had a group, while the only removal is `initCatalog()` and
the DROP contract says IDs are never reused. Catalog churn therefore grows a
boxed ID/map node per engine for the FE lifetime outside the new budgets.
Please return a terminal outcome from the manager/preparer (or use bounded
in-progress DROP state) instead of permanent per-engine membership, and add a
churn/cardinality regression.
--
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]
