github-actions[bot] commented on code in PR #65126:
URL: https://github.com/apache/doris/pull/65126#discussion_r3654115240
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/metacache/MetaCacheEntry.java:
##########
@@ -205,65 +363,199 @@ public MetaCacheEntryStats stats() {
lastError.get());
}
- // Read the config dynamically so existing cache entries follow runtime
config updates.
- private boolean isManualMissLoadEnabled() {
- return Config.enable_external_meta_cache_manual_miss_load;
- }
-
// Execute slow miss loads outside Caffeine's sync load path and suppress
stale write-back after invalidation.
- private V getWithManualLoad(K key, Function<K, V> loadFunction) {
+ private V getWithManualLoad(K key, Function<K, V> loadFunction,
+ @Nullable BiPredicate<K, V> currentValueActionRequired,
+ @Nullable BiConsumer<K, V> currentValueAction) {
if (!effectiveEnabled) {
- // Bypass cache entirely when the entry is disabled so manual miss
load does not relax disable semantics.
- return loadAndTrack(key, loadFunction);
+ if (currentValueAction == null) {
+ // Preserve the ordinary disabled-entry path without adding
publication coordination.
+ return loadAndTrack(key, loadFunction);
+ }
+ // Bypass object publication when disabled, but still fence an
auxiliary-index update against invalidation.
+ long generation;
+ synchronized (publishLock(key)) {
+ generation = generationOf(key);
+ }
+ V loaded = loadAndTrack(key, loadFunction);
+ if (loaded == null) {
+ return loaded;
+ }
+ beforeCurrentValueActionForTest(key, loaded);
+ synchronized (publishLock(key)) {
+ if (generation == generationOf(key)
+ && currentValueActionRequired.test(key, loaded)) {
+ currentValueAction.accept(key, loaded);
+ }
+ }
+ return loaded;
}
V value = data.getIfPresent(key);
if (value != null) {
+ runCurrentValueActionIfPresent(
+ key, value, currentValueActionRequired,
currentValueAction);
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) {
+ runCurrentValueActionIfPresent(
+ key, value, currentValueActionRequired,
currentValueAction);
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) {
+ if (currentValueAction != null
+ && currentValueActionRequired.test(key, value)) {
+ currentValueAction.accept(key, value);
+ }
+ 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);
+ } else if (currentValueAction != null
+ && currentValueActionRequired.test(key, loaded)) {
+ currentValueAction.accept(key, loaded);
+ }
}
return loaded;
}
}
+ private void runCurrentValueActionIfPresent(K key, V value,
+ @Nullable BiPredicate<K, V> currentValueActionRequired,
+ @Nullable BiConsumer<K, V> currentValueAction) {
+ if (currentValueAction == null) {
+ return;
+ }
+ if (!currentValueActionRequired.test(key, value)) {
+ return;
+ }
+ beforeCurrentValueActionForTest(key, value);
+ synchronized (publishLock(key)) {
+ if (data.asMap().get(key) == value
+ && currentValueActionRequired.test(key, value)) {
+ currentValueAction.accept(key, value);
+ }
+ }
+ }
+
+ // 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);
Review Comment:
**[P2] Linearize refresh admission with key invalidation**
`asyncReload()` snapshots the generation without `publishLock`, while
`invalidateKeyAndRun()` bumps that generation before removing K. A cache read
in that gap can still observe `oldValue` and start a refresh that records the
post-invalidation generation; after K is removed, completion sees the same
generation and Caffeine's `currentValue == null` branch can insert the loaded
value again. This needs no capacity eviction, so it is distinct from the
existing bulk-invalidation/eviction thread and applies to routine per-key
invalidation. Please snapshot under `publishLock` only if
`data.asMap().get(key) == oldValue`, and add a deterministic bump-before-remove
overlap test.
##########
fe/fe-core/src/main/java/org/apache/doris/catalog/RefreshManager.java:
##########
@@ -172,45 +172,65 @@ public void replayRefreshTable(ExternalObjectLog log) {
LOG.warn("failed to find db when replaying refresh table: {}",
log.debugForRefreshTable());
return;
}
- Optional<? extends ExternalTable> table;
- if (!Strings.isNullOrEmpty(log.getTableName())) {
- table = db.get().getTableForReplay(log.getTableName());
+ String localTableName;
Review Comment:
**[P1] Replay named refreshes when the database cache is disabled**
The new cold-table fallback is reached only after `getDbForReplay()`
succeeds, but this PR supports global TTL 0 as a disabled object cache: named
database lookups return transient objects and never populate `databases`.
Catalog-scoped `meta.cache.<engine>.<entry>.ttl-second` overrides can still
keep the independent schema/partition/file entries enabled, so queries can fill
those caches while follower replay returns at the database miss above even
though current logs contain canonical `dbName` and `tableName`. That leaves
follower engine metadata stale. This is distinct from the existing cold-table
thread, whose database remains hot. For name-bearing logs, invalidate by the
logged database/table names before requiring a cached database object, and add
a TTL-0 replay test with an enabled engine entry.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/metacache/MetaCacheEntry.java:
##########
@@ -107,11 +166,18 @@ public MetaCacheEntry(String name, @Nullable Function<K,
V> loader, CacheSpec ca
maxSize,
true,
null);
- this.loadingData =
cacheFactory.buildCache(this::loadFromDefaultLoader, refreshExecutor);
+ // Build through a dedicated loader so refresh reload can check
generation before publishing.
+ CacheLoader<K, V> cacheLoader = newCacheLoader();
+ if (syncRemovalListener) {
+ this.loadingData =
cacheFactory.buildCacheWithSyncRemovalListener(cacheLoader, removalListener);
+ } else {
+ this.loadingData = cacheFactory.buildCache(cacheLoader,
refreshExecutor);
+ }
this.data = loadingData;
// Initialize striped locks eagerly to keep the hot path
allocation-free.
for (int i = 0; i < loadLocks.length; i++) {
loadLocks[i] = new Object();
+ publishLocks[i] = new Object();
Review Comment:
**[P2] Avoid allocating 512 locks for every database cache entry**
This initialization eagerly creates two lock objects per stripe for every
`MetaCacheEntry`. With the default 256 stripes, each database whose table
metadata is accessed now retains 512 lock objects plus two reference arrays and
a 256-long generation array for its table cache. The catalog object cache can
retain 1000 such databases by default, so a catalog that touches tables across
them keeps roughly 512k lock objects / about 12 MiB of coordination state
before metadata, and multiple catalogs multiply it. Disabled entries still
perform the same eager allocation because these structures are created before
`effectiveEnabled` is used. Please use lazy/shared striping (or otherwise size
it to useful capacity) and add a scale/allocation guard.
--
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]
