github-actions[bot] commented on code in PR #67417:
URL: https://github.com/apache/doris/pull/67417#discussion_r3978531406
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/CatalogMgr.java:
##########
@@ -815,16 +815,17 @@ public void registerExternalTableFromEvent(String dbName,
String tableName,
if (!(catalog instanceof ExternalCatalog)) {
throw new DdlException("Only support create ExternalCatalog
Tables");
}
+ HMSExternalCatalog hmsCatalog = (HMSExternalCatalog) catalog;
DatabaseIf db = catalog.getDbNullable(dbName);
if (db == null) {
if (!ignoreIfExists) {
throw new DdlException("Database " + dbName + " does not exist
in catalog " + catalog.getName());
}
return;
}
+ long metadataLoadEpoch = ((HMSExternalDatabase)
db).acquireTableMetadataLoadEpoch();
Review Comment:
[P1] Bind event admission to the cached database instance
The database is resolved before its table epoch is acquired. If a routine
parent-cache removal (`onRefreshCache()` or size/expiry eviction) removes that
entry in the gap, its listener resets the detached DB but the catalog epoch
does not change. This call then returns the detached object's new local epoch,
the later validator accepts it, and the event processor can advance after
publishing into an object no longer reachable from the catalog cache. This is
distinct from the existing catalog-reset thread: these removals change no
catalog lifecycle epoch, so that fence cannot catch them. Please bind token
acquisition to the resolved cache identity (or re-resolve/identity-check before
publication) and add an eviction-between-lookup-and-token test.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/metacache/MetaCache.java:
##########
@@ -126,27 +472,41 @@ public Optional<T> getMetaObjById(long id) {
}
public void updateCache(String remoteName, String localName, T obj, long
id) {
- metaObjCache.put(localName, Optional.of(obj));
- namesCache.asMap().compute("", (k, v) -> {
- if (v == null) {
- return Lists.newArrayList(Pair.of(remoteName, localName));
- } else {
- v.add(Pair.of(remoteName, localName));
- return v;
+ updateCache(remoteName, localName, obj, id,
namesLoadEpochSupplier.getAsLong());
+ }
+
+ public boolean updateCache(String remoteName, String localName, T obj,
long id, long expectedEpoch) {
+ // Object loaders already use this monitor and can re-enter
listNames(). Keep the same lock
+ // order here, and never enter Caffeine's same-key mutation while
holding namesMutationLock.
+ synchronized (metaObjCache) {
+ synchronized (namesMutationLock) {
+ if (!namesLoadEpochValidator.test(expectedEpoch)) {
+ return false;
+ }
+ long generation = advanceNamesGeneration();
+ NamesCacheValue current = namesCache.getIfPresent("");
+ Map<String, Pair<String, String>> names = current == null
+ ? Maps.newLinkedHashMap() : current.names;
+ names.put(localName, Pair.of(remoteName, localName));
+ namesCache.put("", new NamesCacheValue(generation, names,
current != null && current.complete));
+ nameUpdateAction.accept(remoteName, localName);
+ idToName.put(id, localName);
}
- });
- idToName.put(id, localName);
+ metaObjCache.put(localName, Optional.of(obj));
+ }
+ return true;
}
public void invalidate(String localName, long id) {
- namesCache.asMap().compute("", (k, v) -> {
- if (v == null) {
- return Lists.newArrayList();
- } else {
- v.removeIf(pair -> pair.value().equals(localName));
- return v;
+ synchronized (namesMutationLock) {
Review Comment:
[P1] Serialize per-name invalidation with event publication
`updateCache()` now holds the outer `metaObjCache` monitor while it
publishes names/id and later puts the object, but this invalidation path does
not. An update can publish its name and pause before line 495; this method then
removes the name/routing, invalidates the still-absent object, and removes the
id; the update finally puts the object. The result is a dropped object still
reachable by direct case-sensitive `getMetaObj()` lookup, and the opposite
schedule can leave a published name with its object removed. This is distinct
from the existing deadlock threads: it is cross-cache per-name ordering
introduced by moving the put outside `namesMutationLock`. Please put update and
invalidate under the same outer monitor/order and add a same-key race test.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/metacache/MetaCache.java:
##########
@@ -74,24 +160,284 @@ public MetaCache(String name,
maxSize,
true,
null);
- namesCache = namesCacheFactory.buildCache(namesCacheLoader, executor);
+ namesCache = namesCacheFactory.buildCache();
// Use sync removal listener to prevent deadlock (removal listener
calls invalidateAll)
// NOTE: This cache should NOT use refreshAfterWrite, as it would
become synchronous
metaObjCache =
objCacheFactory.buildCacheWithSyncRemovalListener(metaObjCacheLoader,
removalListener);
}
public List<String> listNames() {
- return
Objects.requireNonNull(namesCache.get("")).stream().map(Pair::value).collect(Collectors.toList());
+ return
getNames(false).stream().map(Pair::value).collect(Collectors.toList());
+ }
+
+ public List<String> refreshNames() {
+ throwIfInterrupted();
+ NamesLoad loadInProgress;
+ synchronized (namesMutationLock) {
+ loadInProgress = activeNamesLoad;
+ }
+ if (loadInProgress != null) {
+ try {
+ awaitNamesLoad(loadInProgress);
+ } catch (RuntimeException e) {
+ if (Thread.currentThread().isInterrupted()) {
+ throw e;
+ }
+ // This load started before the forced refresh and must not
decide its result.
+ }
+ }
+ throwIfInterrupted();
+ return
getNames(true).stream().map(Pair::value).collect(Collectors.toList());
+ }
+
+ private void throwIfInterrupted() {
+ if (Thread.currentThread().isInterrupted()) {
+ throw new CompletionException(new InterruptedException());
+ }
+ }
+
+ private List<Pair<String, String>> getNames(boolean forceRefresh) {
+ for (int attempt = 0; attempt < MAX_NAMES_LOAD_ATTEMPTS; attempt++) {
+ if (forceRefresh) {
+ throwIfInterrupted();
+ }
+ NamesCacheValue value = forceRefresh ? null :
namesCache.getIfPresent("");
+ List<Pair<String, String>> currentNames = null;
+ synchronized (namesMutationLock) {
+ if (value != null && value.complete && value.generation ==
namesGeneration.get()) {
+ currentNames = value.snapshot();
+ }
+ }
+ if (currentNames != null) {
+ scheduleNamesRefresh(value);
+ return currentNames;
+ }
+ value = loadNames(forceRefresh, true, null);
+ synchronized (namesMutationLock) {
+ if (value != null && value.complete && value.generation ==
namesGeneration.get()) {
+ return value.snapshot();
+ }
+ }
+ }
+ throw new IllegalStateException("Failed to load names for " + name
+ + " because metadata kept changing");
+ }
+
+ private NamesCacheValue loadNames(boolean forceRefresh, boolean
awaitActiveLoad, Long expectedGeneration) {
+ NamesLoad namesLoad = null;
+ boolean loadOwner = false;
+ long requestedGeneration;
+ synchronized (namesMutationLock) {
+ if (expectedGeneration != null && expectedGeneration !=
namesGeneration.get()) {
+ return null;
+ }
+ NamesCacheValue cached = namesCache.getIfPresent("");
+ if (!forceRefresh && cached != null && cached.complete
+ && cached.generation == namesGeneration.get()) {
+ return cached;
+ }
+ long loadGeneration = namesGeneration.get();
+ if (activeNamesLoad != null && activeNamesLoad.generation ==
loadGeneration) {
+ if (!awaitActiveLoad) {
+ return null;
+ }
+ namesLoad = activeNamesLoad;
+ }
+ requestedGeneration = loadGeneration;
+ }
+
+ if (namesLoad != null) {
+ return awaitNamesLoad(namesLoad);
+ }
+
+ // Lifecycle admission may acquire the catalog monitor. Keep it
outside the names
+ // mutation lock because catalog reset advances the names generation
under that monitor.
+ long loadEpoch = namesLoadEpochSupplier.getAsLong();
+ synchronized (namesMutationLock) {
+ if (requestedGeneration != namesGeneration.get()
+ || expectedGeneration != null && expectedGeneration !=
namesGeneration.get()) {
+ return null;
+ }
+ NamesCacheValue cached = namesCache.getIfPresent("");
+ if (!forceRefresh && cached != null && cached.complete
+ && cached.generation == requestedGeneration) {
+ return cached;
+ }
+ if (activeNamesLoad != null && activeNamesLoad.generation ==
requestedGeneration) {
+ if (!awaitActiveLoad) {
+ return null;
+ }
+ namesLoad = activeNamesLoad;
+ } else {
+ if (physicalNamesLoads.size() >= MAX_PHYSICAL_NAMES_LOADS) {
+ return null;
+ }
+ Map<String, Pair<String, String>> incompleteNames = cached !=
null && !cached.complete
+ ? Maps.newLinkedHashMap(cached.names) :
Maps.newLinkedHashMap();
+ namesLoad = new NamesLoad(requestedGeneration, loadEpoch,
incompleteNames);
+ activeNamesLoad = namesLoad;
+ physicalNamesLoads.add(namesLoad);
+ loadOwner = true;
+ }
+ }
+
+ if (!loadOwner) {
+ return awaitNamesLoad(namesLoad);
+ }
+
+ try {
+ List<Pair<String, String>> loadedNames =
Objects.requireNonNull(namesCacheLoader.load(""));
+ NamesCacheValue value = null;
+ synchronized (namesMutationLock) {
+ if (namesLoad.generation == namesGeneration.get()
+ && namesLoad.generation >= minimumLoadGeneration
+ && namesLoadEpochValidator.test(namesLoad.loadEpoch)) {
+ Map<String, Pair<String, String>> names =
toNamesMap(loadedNames);
+ names.putAll(namesLoad.incompleteNames);
+ value = new NamesCacheValue(namesGeneration.get(), names,
true);
+ namesCache.put("", value);
+ publishNames(value);
+ }
+ }
+ namesLoad.result.complete(value);
+ return value;
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ CompletionException failure = new CompletionException(e);
+ namesLoad.result.completeExceptionally(failure);
+ throw failure;
+ } catch (RuntimeException e) {
+ if (!namesLoad.result.completeExceptionally(e)) {
+ return null;
+ }
+ throw e;
+ } catch (Error e) {
+ if (!namesLoad.result.completeExceptionally(e)) {
Review Comment:
[P1] Do not convert a retired load's Error into a retry
If a generation advance has already completed this owner's future with
`null`, `completeExceptionally(e)` returns false and this branch returns
`null`. The owning `getNames()` call then retries, so fatal failures such as
`OutOfMemoryError` or linkage/initialization errors from the physical loader
are swallowed and can trigger more loading work. The existing retired-failure
handling is appropriate for recoverable connector exceptions, but `Error`
should always propagate from the owner. Please rethrow it regardless of
future-completion state and cover a retired-load `Error` case.
--
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]
