github-actions[bot] commented on code in PR #65126:
URL: https://github.com/apache/doris/pull/65126#discussion_r3599912102
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/ExternalDatabase.java:
##########
@@ -599,10 +589,148 @@ private String getLocalTableName(String tableName,
boolean isReplay) {
return finalName;
}
+ private NameCacheValue getTableNamesValue(boolean allowLoad) {
+ if (tableNames == null) {
+ return null;
+ }
+ return allowLoad ? tableNames.get("") : tableNames.getIfPresent("");
+ }
+
+ // Centralize names-negative-lookup handling so all table lookup paths
share the same config-driven policy.
+ @Nullable
+ private <R> R resolveTableNameFromSnapshot(String lookupName, boolean
isReplay,
+ Function<NameCacheValue, R> resolver) {
+ NameCacheValue cached = getTableNamesValue(false);
+ if (cached == null) {
+ if (isReplay) {
+ return null;
+ }
+ NameCacheValue loaded = getTableNamesValue(true);
+ return loaded == null ? null : resolver.apply(loaded);
+ }
+ R resolved = resolver.apply(cached);
+ if (resolved != null || isReplay ||
!Config.enable_external_meta_cache_name_miss_refresh) {
+ return resolved;
+ }
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("refresh table names after hot-snapshot miss, catalog:
{}, db: {}, lookup: {}",
+ getCatalog().getName(), getFullName(), lookupName);
+ }
+ resetMetaCacheNames();
+ NameCacheValue refreshed = getTableNamesValue(true);
+ return refreshed == null ? null : resolver.apply(refreshed);
+ }
+
+ private List<String> listLocalTableNamesFromCache() {
+ NameCacheValue namesValue = java.util.Objects.requireNonNull(
+ getTableNamesValue(true), "table names cache can not be null");
+ return namesValue.localNames();
+ }
+
+ private List<String> listLocalTableNamesWithoutCache(SessionContext
sessionContext) {
+ return
listTableNames(sessionContext).stream().map(Pair::value).collect(Collectors.toList());
+ }
+
+ @Nullable
+ private String getRemoteTableName(String localTableName, boolean isReplay)
{
+ // Route local-to-remote resolution through the shared helper so miss
reload stays consistent with lookups.
+ return resolveTableNameFromSnapshot(localTableName, isReplay,
+ namesValue ->
namesValue.remoteNameOfLocalName(localTableName));
+ }
+
+ private Optional<Pair<String, String>>
findTableNamePairWithoutCache(SessionContext sessionContext,
+ String requestedTableName) {
+ return listTableNames(sessionContext).stream()
+ .filter(pair -> matchesLocalTableName(pair.value(),
requestedTableName))
+ .findFirst();
+ }
+
+ private boolean matchesLocalTableName(String localTableName, String
requestedTableName) {
+ if (isStoredTableNamesLowerCase()) {
+ return
localTableName.equals(requestedTableName.toLowerCase(Locale.ROOT));
+ }
+ if (isTableNamesCaseInsensitive()) {
+ return localTableName.equalsIgnoreCase(requestedTableName);
+ }
+ return localTableName.equals(requestedTableName);
+ }
+
+ private String resolveTableNameForInvalidation(String tableName) {
+ if (isStoredTableNamesLowerCase()) {
+ return tableName.toLowerCase(Locale.ROOT);
+ }
+ if (!isTableNamesCaseInsensitive()) {
+ return tableName;
+ }
+
+ String localTableName = getLocalTableName(tableName, true);
+ if (localTableName != null) {
+ return localTableName;
+ }
+ T cachedTable = tables.findIfPresent(key ->
key.equalsIgnoreCase(tableName));
+ if (cachedTable != null) {
+ return cachedTable.getName();
+ }
+ return tableIdToName.values().stream()
+ .filter(name -> name.equalsIgnoreCase(tableName))
+ .findFirst()
+ .orElse(tableName);
+ }
+
+ private void updateTableCache(T table, String remoteTableName, String
localTableName) {
+ updateTableCache(table, remoteTableName, localTableName, false);
+ }
+
+ protected void updateTableCache(T table, String remoteTableName, String
localTableName,
+ boolean forceUpdateCacheState) {
+ buildMetaCache();
+ // Runtime incremental events only maintain names and object entries
that are already hot. The ID map is a
+ // lightweight lookup index and must always track registered objects
so normal by-ID lookup can load on demand.
+ if (forceUpdateCacheState) {
+ tableNames.compute("", (ignored, current) ->
+ (current == null ? NameCacheValue.empty() :
current).withName(remoteTableName, localTableName));
+ } else if (tableNames.getIfPresent("") != null) {
Review Comment:
While a slow names load is in flight, the entry is still absent from
Caffeine. A concurrent CREATE_TABLE event therefore fails this guard, performs
no `compute`, and leaves the load generation unchanged; the loader can then
publish the pre-create snapshot after the event. DROP has the symmetric
stale-positive race, and the catalog database-name branches use the same
pattern. Please advance the generation even when the entry is cold (for example
with an unconditional null-preserving compute/touch that does not preheat it),
and test an event paused against an in-flight manual names load.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/ExternalDatabase.java:
##########
@@ -599,10 +589,148 @@ private String getLocalTableName(String tableName,
boolean isReplay) {
return finalName;
}
+ private NameCacheValue getTableNamesValue(boolean allowLoad) {
+ if (tableNames == null) {
+ return null;
+ }
+ return allowLoad ? tableNames.get("") : tableNames.getIfPresent("");
+ }
+
+ // Centralize names-negative-lookup handling so all table lookup paths
share the same config-driven policy.
+ @Nullable
+ private <R> R resolveTableNameFromSnapshot(String lookupName, boolean
isReplay,
+ Function<NameCacheValue, R> resolver) {
+ NameCacheValue cached = getTableNamesValue(false);
+ if (cached == null) {
+ if (isReplay) {
+ return null;
+ }
+ NameCacheValue loaded = getTableNamesValue(true);
+ return loaded == null ? null : resolver.apply(loaded);
+ }
+ R resolved = resolver.apply(cached);
+ if (resolved != null || isReplay ||
!Config.enable_external_meta_cache_name_miss_refresh) {
+ return resolved;
+ }
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("refresh table names after hot-snapshot miss, catalog:
{}, db: {}, lookup: {}",
+ getCatalog().getName(), getFullName(), lookupName);
+ }
+ resetMetaCacheNames();
+ NameCacheValue refreshed = getTableNamesValue(true);
+ return refreshed == null ? null : resolver.apply(refreshed);
+ }
+
+ private List<String> listLocalTableNamesFromCache() {
+ NameCacheValue namesValue = java.util.Objects.requireNonNull(
+ getTableNamesValue(true), "table names cache can not be null");
+ return namesValue.localNames();
+ }
+
+ private List<String> listLocalTableNamesWithoutCache(SessionContext
sessionContext) {
+ return
listTableNames(sessionContext).stream().map(Pair::value).collect(Collectors.toList());
+ }
+
+ @Nullable
+ private String getRemoteTableName(String localTableName, boolean isReplay)
{
+ // Route local-to-remote resolution through the shared helper so miss
reload stays consistent with lookups.
+ return resolveTableNameFromSnapshot(localTableName, isReplay,
+ namesValue ->
namesValue.remoteNameOfLocalName(localTableName));
+ }
+
+ private Optional<Pair<String, String>>
findTableNamePairWithoutCache(SessionContext sessionContext,
+ String requestedTableName) {
+ return listTableNames(sessionContext).stream()
+ .filter(pair -> matchesLocalTableName(pair.value(),
requestedTableName))
+ .findFirst();
+ }
+
+ private boolean matchesLocalTableName(String localTableName, String
requestedTableName) {
+ if (isStoredTableNamesLowerCase()) {
+ return
localTableName.equals(requestedTableName.toLowerCase(Locale.ROOT));
+ }
+ if (isTableNamesCaseInsensitive()) {
+ return localTableName.equalsIgnoreCase(requestedTableName);
+ }
+ return localTableName.equals(requestedTableName);
+ }
+
+ private String resolveTableNameForInvalidation(String tableName) {
+ if (isStoredTableNamesLowerCase()) {
+ return tableName.toLowerCase(Locale.ROOT);
+ }
+ if (!isTableNamesCaseInsensitive()) {
+ return tableName;
+ }
+
+ String localTableName = getLocalTableName(tableName, true);
+ if (localTableName != null) {
+ return localTableName;
+ }
+ T cachedTable = tables.findIfPresent(key ->
key.equalsIgnoreCase(tableName));
+ if (cachedTable != null) {
+ return cachedTable.getName();
+ }
+ return tableIdToName.values().stream()
+ .filter(name -> name.equalsIgnoreCase(tableName))
+ .findFirst()
+ .orElse(tableName);
+ }
+
+ private void updateTableCache(T table, String remoteTableName, String
localTableName) {
+ updateTableCache(table, remoteTableName, localTableName, false);
+ }
+
+ protected void updateTableCache(T table, String remoteTableName, String
localTableName,
+ boolean forceUpdateCacheState) {
+ buildMetaCache();
+ // Runtime incremental events only maintain names and object entries
that are already hot. The ID map is a
+ // lightweight lookup index and must always track registered objects
so normal by-ID lookup can load on demand.
+ if (forceUpdateCacheState) {
+ tableNames.compute("", (ignored, current) ->
+ (current == null ? NameCacheValue.empty() :
current).withName(remoteTableName, localTableName));
+ } else if (tableNames.getIfPresent("") != null) {
+ // The outer hot-entry check only skips a pointless compute on
cold state. current may still
+ // become null here if another thread invalidates the names entry
between getIfPresent and compute.
+ tableNames.compute("", (ignored, current) ->
+ current == null ? null : current.withName(remoteTableName,
localTableName));
+ }
+ if (forceUpdateCacheState || tables.getIfPresent(localTableName) !=
null) {
+ tables.put(localTableName, table);
+ }
+ tableIdToName.put(table.getId(), localTableName);
Review Comment:
A CREATE_TABLE event now leaves names and objects cold but always adds this
ID entry. If the table is already gone remotely when the matching DROP_TABLE
event runs, `CatalogMgr.unregisterExternalTable()` performs a load-through
lookup, gets null from the post-drop name list, and returns for
`ignoreIfExists=true` before `unregisterTable()` can remove this entry.
Repeated cold create/drop churn therefore accumulates stale ID mappings (and
by-ID lookups keep attempting to load absent tables). Please make the ignored
event-miss path perform canonical cache/ID cleanup, and cover the production
CREATE-event -> remote drop -> DROP-event wrapper rather than only calling
`unregisterTable()` directly.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/metacache/MetaCacheEntry.java:
##########
@@ -222,48 +315,133 @@ private V getWithManualLoad(K key, Function<K, V>
loadFunction) {
return value;
}
+ // Keep the slow miss load under the per-key load lock so concurrent
misses for the same key
+ // are still deduplicated. publishLock(key) only protects the short
publication window.
synchronized (loadLock(key)) {
value = data.asMap().get(key);
if (value != null) {
return value;
}
- long generation = invalidateGeneration.get();
- V loaded = loadAndTrack(key, loadFunction);
- if (generation != invalidateGeneration.get()) {
- return loaded;
+ long generation;
+ // Snapshot generation only after re-checking the cache under the
publication lock so
+ // public mutations cannot slip between the miss observation and
the captured version.
+ synchronized (publishLock(key)) {
+ value = data.asMap().get(key);
+ if (value != null) {
+ return value;
+ }
+ generation = generationOf(key);
}
-
- // Keep null results uncached so manual miss load matches
LoadingCache null-return behavior.
+ V loaded = loadAndTrack(key, loadFunction);
if (loaded == null) {
return null;
}
- // Leave a narrow hook for tests to pause exactly before the cache
put race window.
- beforeManualCachePutForTest(key, loaded);
- data.put(key, loaded);
- if (generation != invalidateGeneration.get()) {
- removeLoadedValue(key, loaded);
+ synchronized (publishLock(key)) {
+ if (generation != generationOf(key)) {
+ return loaded;
+ }
+ // Leave a narrow hook for tests to pause exactly before the
cache put race window.
+ beforeManualCachePutForTest(key, loaded);
+ putLoadedValueWithoutGenerationBump(key, loaded);
+ // Re-check after the put because
invalidateAll()/invalidateIf() may bump stripe generations
+ // outside publishLock while this thread is publishing a
loaded value.
+ if (generation != generationOf(key)) {
+ removeLoadedValueWithoutGenerationBump(key, loaded);
+ }
}
return loaded;
}
}
+ // Keep internal load write-back separate from public mutation so it does
not advance generation.
+ private void putLoadedValueWithoutGenerationBump(K key, V loaded) {
+ data.put(key, loaded);
+ }
+
// Remove only the value loaded by the current request and keep newer
replacements intact.
- private void removeLoadedValue(K key, V loaded) {
+ private void removeLoadedValueWithoutGenerationBump(K key, V loaded) {
data.asMap().computeIfPresent(key, (ignored, currentValue) ->
currentValue == loaded ? null : currentValue);
}
+ private CacheLoader<K, V> newCacheLoader() {
+ return new CacheLoader<K, V>() {
+ @Override
+ public V load(K key) {
+ return loadFromDefaultLoader(key);
+ }
+
+ @Override
+ public CompletableFuture<V> asyncReload(K key, V oldValue,
Executor executor) {
+ long generation = generationOf(key);
+ CompletableFuture<V> result = new CompletableFuture<>();
+ CompletableFuture.supplyAsync(() ->
loadFromDefaultLoader(key), executor)
+ .whenComplete((loaded, error) -> {
+ if (error != null) {
+ result.completeExceptionally(error);
+ return;
+ }
+ synchronized (publishLock(key)) {
+ if (generation == generationOf(key)) {
+ result.complete(loaded);
Review Comment:
This generation check is not atomic with the actual Caffeine refresh
publication because `bumpAllGenerations()` runs without `publishLock`. If K is
capacity-evicted while its refresh is running, a bulk invalidation can bump
generations and finish scanning while K is absent; `result.complete(loaded)`
can then run Caffeine 2.9.3's refresh callback, whose `currentValue == null`
branch reinserts K after the invalidation. Please linearize bulk generation
changes with refresh completion/publication even for evicted keys, and add a
deterministic eviction + bulk-invalidation race test.
--
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]
