github-actions[bot] commented on code in PR #67417:
URL: https://github.com/apache/doris/pull/67417#discussion_r4012910481
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/metacache/MetaCache.java:
##########
@@ -74,40 +162,300 @@ 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();
+ // Retire any active load so the forced refresh is not blocked behind
a stuck
+ // background refresh. Advance the generation so the old loader's
publication
+ // predicate fails and it cannot overwrite the forced load's result.
+ synchronized (namesMutationLock) {
+ advanceNamesGeneration();
Review Comment:
[P1] Preserve the warm snapshot when a forced refresh fails
This advances the generation before the forced connector call, but a failure
neither republishes nor restores the existing complete value. The generation-G
entry remains in Caffeine, yet every later ordinary listNames call at G+1
rejects it and performs fresh connector I/O, so one mode-2 miss during an HMS
outage makes otherwise valid cached database/table listings fail until a remote
load succeeds. Please use a separate retirement/publication token or carry the
complete warm value into the new generation on failure, and cover warm ->
failed force -> ordinary read without another physical load.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/metacache/MetaCache.java:
##########
@@ -74,40 +162,300 @@ 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();
+ // Retire any active load so the forced refresh is not blocked behind
a stuck
+ // background refresh. Advance the generation so the old loader's
publication
+ // predicate fails and it cannot overwrite the forced load's result.
+ synchronized (namesMutationLock) {
+ advanceNamesGeneration();
+ physicalNamesLoads.clear();
+ }
+ 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);
+ }
+ }
+ finishNamesLoad(namesLoad);
+ namesLoad.result.complete(value);
+ return value;
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ CompletionException failure = new CompletionException(e);
+ finishNamesLoad(namesLoad);
+ namesLoad.result.completeExceptionally(failure);
+ throw failure;
+ } catch (RuntimeException e) {
+ finishNamesLoad(namesLoad);
+ if (!namesLoad.result.completeExceptionally(e)) {
+ return null;
+ }
+ throw e;
+ } catch (Error e) {
+ finishNamesLoad(namesLoad);
+ namesLoad.result.completeExceptionally(e);
+ throw e;
+ } catch (Exception e) {
+ CompletionException failure = new CompletionException(e);
+ finishNamesLoad(namesLoad);
+ if (!namesLoad.result.completeExceptionally(failure)) {
+ return null;
+ }
+ throw failure;
+ } finally {
+ finishNamesLoad(namesLoad);
+ }
+ }
+
+ private void finishNamesLoad(NamesLoad namesLoad) {
+ synchronized (namesMutationLock) {
+ if (activeNamesLoad == namesLoad) {
+ activeNamesLoad = null;
+ }
+ physicalNamesLoads.remove(namesLoad);
+ }
+ }
+
+ private NamesCacheValue awaitNamesLoad(NamesLoad namesLoad) {
+ try {
+ return namesLoad.result.get();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new CompletionException(e);
+ } catch (ExecutionException e) {
+ Throwable cause = e.getCause();
+ if (cause instanceof RuntimeException) {
+ throw (RuntimeException) cause;
+ }
+ if (cause instanceof Error) {
+ throw (Error) cause;
+ }
+ throw new CompletionException(cause);
+ }
+ }
+
+ private Map<String, Pair<String, String>> toNamesMap(List<Pair<String,
String>> names) {
+ Map<String, Pair<String, String>> namesMap = Maps.newLinkedHashMap();
+ for (Pair<String, String> pair : names) {
+ namesMap.put(pair.value(), pair);
+ }
+ return namesMap;
+ }
+
+ private void scheduleNamesRefresh(NamesCacheValue value) {
+ if (System.nanoTime() - value.writeNanos <
namesRefreshAfterWriteNanos) {
+ return;
+ }
+ scheduleNamesRefresh(value.generation);
+ }
+
+ private void scheduleNamesRefresh(long generation) {
+ NamesRefresh refresh;
+ synchronized (namesMutationLock) {
+ if (generation != namesGeneration.get()
+ || activeNamesRefreshes.containsKey(generation)) {
+ return;
+ }
+ if (activeNamesRefreshes.size() >= MAX_NAMES_REFRESH_FLIGHTS) {
+ pendingNamesRefreshGeneration = generation;
+ return;
+ }
+ refresh = new NamesRefresh(generation);
+ activeNamesRefreshes.put(generation, refresh);
+ if (pendingNamesRefreshGeneration != null &&
pendingNamesRefreshGeneration == generation) {
+ pendingNamesRefreshGeneration = null;
+ }
+ }
+ try {
+ namesRefreshExecutor.execute(() -> {
+ try {
+ loadNames(true, false, refresh.generation);
+ } catch (Exception e) {
+ LOG.warn("Failed to refresh names cache for {}", name, e);
+ } finally {
+ clearNamesRefresh(refresh);
+ }
+ });
+ } catch (RuntimeException e) {
+ clearNamesRefresh(refresh);
+ LOG.warn("Failed to schedule names cache refresh for {}", name, e);
+ }
+ }
+
+ private void clearNamesRefresh(NamesRefresh refresh) {
+ Long pendingGeneration = null;
+ synchronized (namesMutationLock) {
+ if (activeNamesRefreshes.get(refresh.generation) == refresh) {
+ activeNamesRefreshes.remove(refresh.generation);
+ if (pendingNamesRefreshGeneration != null) {
+ pendingGeneration = pendingNamesRefreshGeneration;
+ pendingNamesRefreshGeneration = null;
+ }
+ }
+ }
+ if (pendingGeneration != null) {
+ scheduleNamesRefresh(pendingGeneration);
+ }
}
public String getRemoteName(String localName) {
- return Objects.requireNonNull(namesCache.getIfPresent("")).stream()
+ NamesCacheValue value = namesCache.getIfPresent("");
+ synchronized (namesMutationLock) {
+ if (value != null && value.generation == namesGeneration.get()) {
+ Pair<String, String> pair = value.names.get(localName);
+ if (pair != null) {
+ return pair.key();
+ }
+ if (value.complete) {
+ return null;
+ }
+ }
+ }
+ return getNames(false).stream()
.filter(pair -> pair.value().equals(localName))
.map(Pair::key)
.findFirst()
.orElse(null);
}
+ private void publishNames(NamesCacheValue value) {
+ namesCacheUpdateAction.accept(value.snapshot());
+ }
+
public Optional<T> getMetaObj(String name, long id) {
Optional<T> val = metaObjCache.getIfPresent(name);
- if (val == null || !val.isPresent()) {
- synchronized (metaObjCache) {
- val = metaObjCache.getIfPresent(name);
- if (val != null && val.isPresent()) {
- return val;
- }
- if (LOG.isDebugEnabled()) {
- LOG.debug("trigger getMetaObj in metacache {}, obj name:
{}, id: {}",
- this.name, name, id, new Exception());
- }
- metaObjCache.invalidate(name);
- val = metaObjCache.get(name);
- idToName.put(id, name);
- }
+ if (val != null && val.isPresent()) {
+ return val;
+ }
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("trigger getMetaObj in metacache {}, obj name: {}, id:
{}",
+ this.name, name, id, new Exception());
+ }
+ // Do not hold a cache-wide monitor during the blocking loader call
+ // (buildDbForInit → connector I/O). Caffeine serializes same-key
loads internally;
+ // different keys can load in parallel without blocking
updateCache/invalidate.
+ metaObjCache.invalidate(name);
Review Comment:
[P1] Serialize same-key cold loads before invalidating
Caller B can observe a cold key as absent and pause before this
unconditional invalidation; caller A can then finish and publish its load,
after which B resumes and removes A's newly cached value. For catalog objects
the synchronous removal listener resets A's database, B performs a second load,
and A can still return the detached/reset instance while the cache holds a
different one. Please put the miss recheck, empty-entry invalidation, and load
behind a per-key/striped gate, and add a latch-based regression that forces
this schedule. This is distinct from the existing unrelated-key blocking thread
about the former cache-wide monitor.
--
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]
