FeMetaCacheEntry.java
// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements. See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership. The ASF licenses this file
// to you under the Apache License, Version 2.0 (the
// "License"); you may not use this file except in compliance
// with the License. You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing,
// software distributed under the License is distributed on an
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
// KIND, either express or implied. See the License for the
// specific language governing permissions and limitations
// under the License.
package org.apache.doris.datasource.metacache;
import org.apache.doris.common.Config;
import org.apache.doris.connector.cache.CacheSpec;
import org.apache.doris.connector.cache.CatalogMetaCache;
import org.apache.doris.connector.cache.MetaCache;
import org.apache.doris.connector.cache.MetaCacheDefinition;
import org.apache.doris.connector.cache.MetaCacheRemovalReason;
import org.apache.doris.connector.cache.MetaCacheSizeEstimators;
import org.apache.doris.connector.cache.ScopePath;
import com.github.benmanes.caffeine.cache.RemovalCause;
import com.github.benmanes.caffeine.cache.RemovalListener;
import java.time.Duration;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReferenceArray;
import java.util.function.BiConsumer;
import java.util.function.BiFunction;
import java.util.function.BiPredicate;
import java.util.function.Function;
import java.util.function.Predicate;
import java.util.function.Supplier;
import javax.annotation.Nullable;
/**
* FE naming-cache adapter over the shared connector-cache runtime.
*
* <p>The common runtime owns value storage, load deduplication, generations, eviction and leak-free indexes. This
* adapter only retains the short per-key publication window needed to update FE's auxiliary ID/name indexes together
* with a database or table object.
*/
public class FeMetaCacheEntry<K, V> {
private static final int SINGLE_KEY_STRIPES = 1;
private static final class StripeState<K> {
@Nullable
private Map<K, ActionState> activeActions;
}
private static final class ActionState {
private long generation;
private int references;
}
private static final class ActionToken<K> {
private final K key;
private final ActionState state;
private final long generation;
private ActionToken(K key, ActionState state) {
this.key = key;
this.state = state;
this.generation = state.generation;
}
}
private final String name;
@Nullable
private final Function<K, V> loader;
private final CacheSpec cacheSpec;
private final boolean effectiveEnabled;
private final boolean autoRefresh;
private final int stripeCount;
private final AtomicReferenceArray<StripeState<K>> stripeStates;
private final CatalogMetaCache owner;
private final ScopePath invalidationScope;
private final MetaCache<K, V> data;
private final BiConsumer<K, V> discardedValueAction;
private final Object lifecycleLock = new Object();
private Runnable retirementCleanup;
private final AtomicBoolean registrationReleased = new AtomicBoolean();
private volatile boolean retired;
private int activeUses;
public FeMetaCacheEntry(String name, Function<K, V> loader, CacheSpec cacheSpec, ExecutorService refreshExecutor) {
this(name, loader, cacheSpec, refreshExecutor, true, false, defaultObjectStripeCount(), null);
}
public FeMetaCacheEntry(String name, Function<K, V> loader, CacheSpec cacheSpec, ExecutorService refreshExecutor,
boolean autoRefresh) {
this(name, loader, cacheSpec, refreshExecutor, autoRefresh, false, defaultObjectStripeCount(), null);
}
public FeMetaCacheEntry(String name, @Nullable Function<K, V> loader, CacheSpec cacheSpec,
ExecutorService refreshExecutor, boolean autoRefresh, boolean contextualOnly) {
this(name, loader, cacheSpec, refreshExecutor, autoRefresh, contextualOnly,
defaultObjectStripeCount(), null);
}
public FeMetaCacheEntry(String name, Function<K, V> loader, CacheSpec cacheSpec, ExecutorService refreshExecutor,
boolean autoRefresh, int stripeCount) {
this(name, loader, cacheSpec, refreshExecutor, autoRefresh, false, stripeCount, null);
}
public FeMetaCacheEntry(String name, @Nullable Function<K, V> loader, CacheSpec cacheSpec,
ExecutorService refreshExecutor, boolean autoRefresh, boolean contextualOnly, int stripeCount) {
this(name, loader, cacheSpec, refreshExecutor, autoRefresh, contextualOnly, stripeCount, null);
}
public static <K, V> FeMetaCacheEntry<K, V> withSyncRemovalListener(String name, Function<K, V> loader,
CacheSpec cacheSpec, ExecutorService refreshExecutor, RemovalListener<K, V> removalListener) {
return withSyncRemovalListener(name, loader, cacheSpec, refreshExecutor,
defaultObjectStripeCount(), removalListener);
}
public static <K, V> FeMetaCacheEntry<K, V> withSyncRemovalListener(String name, Function<K, V> loader,
CacheSpec cacheSpec, ExecutorService refreshExecutor, int stripeCount,
RemovalListener<K, V> removalListener) {
return new FeMetaCacheEntry<>(name, loader, cacheSpec, refreshExecutor, false, false, stripeCount,
Objects.requireNonNull(removalListener, "removalListener can not be null"));
}
FeMetaCacheEntry(String name, @Nullable Function<K, V> loader, CacheSpec cacheSpec,
ExecutorService refreshExecutor, boolean autoRefresh, boolean contextualOnly, int stripeCount,
@Nullable RemovalListener<K, V> removalListener) {
this(CatalogMetaCache.unmanaged(), ignored -> ScopePath.catalog(), ScopePath.catalog(),
name, loader, cacheSpec, refreshExecutor, autoRefresh, contextualOnly,
stripeCount, removalListener, false, MetaCacheDefinition.BudgetPolicy.WEIGHTED_PAYLOAD);
}
public static <K, V> FeMetaCacheEntry<K, V> shared(CatalogMetaCache owner,
Function<K, ScopePath> scopeResolver, ScopePath invalidationScope,
String name, Function<K, V> loader, CacheSpec cacheSpec, ExecutorService refreshExecutor,
boolean autoRefresh, int stripeCount, @Nullable RemovalListener<K, V> removalListener) {
return new FeMetaCacheEntry<>(owner, scopeResolver, invalidationScope.in(ScopePath.Coordinates.LOCAL),
name, loader, cacheSpec,
refreshExecutor, autoRefresh, false, stripeCount, removalListener, true,
MetaCacheDefinition.BudgetPolicy.WEIGHTED_PAYLOAD);
}
/** Lookup handles retain their existing count/TTL policy; schema payloads use the owner's byte budget. */
public static <K, V> FeMetaCacheEntry<K, V> sharedLookup(CatalogMetaCache owner,
Function<K, ScopePath> scopeResolver, ScopePath invalidationScope,
String name, Function<K, V> loader, CacheSpec cacheSpec, ExecutorService refreshExecutor,
boolean autoRefresh, int stripeCount, @Nullable RemovalListener<K, V> removalListener) {
return new FeMetaCacheEntry<>(owner, scopeResolver, invalidationScope.in(ScopePath.Coordinates.LOCAL),
name, loader, cacheSpec,
refreshExecutor, autoRefresh, false, stripeCount, removalListener, true,
MetaCacheDefinition.BudgetPolicy.COUNT_ONLY);
}
private FeMetaCacheEntry(CatalogMetaCache owner, Function<K, ScopePath> scopeResolver,
ScopePath invalidationScope, String name, @Nullable Function<K, V> loader, CacheSpec cacheSpec,
ExecutorService refreshExecutor, boolean autoRefresh, boolean contextualOnly, int stripeCount,
@Nullable RemovalListener<K, V> removalListener, boolean retireDiscardedValues,
MetaCacheDefinition.BudgetPolicy budgetPolicy) {
this.owner = Objects.requireNonNull(owner, "owner");
discardedValueAction = retireDiscardedValues && removalListener != null
? (key, value) -> removalListener.onRemoval(key, value, RemovalCause.EXPLICIT) : null;
this.invalidationScope = Objects.requireNonNull(invalidationScope, "invalidationScope");
Function<K, ScopePath> entryScope = retireDiscardedValues
? key -> scopeResolver.apply(key).in(ScopePath.Coordinates.LOCAL) : scopeResolver;
this.name = Objects.requireNonNull(name, "name can not be null");
this.loader = loader;
this.cacheSpec = Objects.requireNonNull(cacheSpec, "cacheSpec can not be null");
this.autoRefresh = autoRefresh;
Objects.requireNonNull(refreshExecutor, "refreshExecutor can not be null");
if (contextualOnly && loader != null) {
throw new IllegalArgumentException("contextual-only entry loader must be null");
}
if (contextualOnly && autoRefresh) {
throw new IllegalArgumentException("contextual-only entry can not enable auto refresh");
}
if (!contextualOnly) {
Objects.requireNonNull(loader, "loader can not be null");
}
if (removalListener != null && autoRefresh) {
throw new IllegalArgumentException("sync removal listener cache can not enable refreshAfterWrite");
}
if (stripeCount < 1) {
throw new IllegalArgumentException("stripeCount must be positive");
}
this.stripeCount = stripeCount;
stripeStates = new AtomicReferenceArray<>(stripeCount);
if (stripeCount == SINGLE_KEY_STRIPES) {
stripeStates.set(0, new StripeState<>());
}
effectiveEnabled = CacheSpec.isCacheEnabled(
cacheSpec.isEnable(), cacheSpec.getTtlSecond(), cacheSpec.getCapacity());
MetaCacheDefinition.Builder<K, V> builder = MetaCacheDefinition.<K, V>builder(
name, cacheSpec, entryScope)
.sizeEstimator(MetaCacheSizeEstimators.reflective())
.budgetPolicy(budgetPolicy);
if (loader != null) {
builder.loader(key -> loadAndPause(key, loader));
}
if (removalListener != null) {
builder.removalListener((key, value, reason) ->
removalListener.onRemoval(key, value, toCaffeineRemovalCause(reason)));
// An uncached database is still returned to its caller. Retire its cache registrations too,
// so the caller uses the transient lookup path rather than leaking a new set of child entries.
if (discardedValueAction != null) {
builder.discardListener(discardedValueAction);
}
}
if (autoRefresh && Config.external_cache_refresh_time_minutes > 0) {
builder.refreshAfterWrite(
Duration.ofMinutes(Config.external_cache_refresh_time_minutes), refreshExecutor);
}
data = owner.create(builder.build());
}
public String name() {
return name;
}
public V get(K key) {
if (loader == null) {
throw new UnsupportedOperationException(String.format(
"Entry '%s' requires a contextual miss loader.", name));
}
return useCache(() -> data.get(key), () -> loadWithoutCaching(key, loader));
}
public V get(K key, Function<K, V> missLoader) {
Function<K, V> nonNullLoader = Objects.requireNonNull(missLoader, "missLoader can not be null");
return useCache(() -> data.get(key, loadKey -> loadAndPause(loadKey, nonNullLoader)),
() -> loadWithoutCaching(key, nonNullLoader));
}
public V getAndRunIfCurrent(K key, BiConsumer<K, V> currentValueAction) {
return getAndRunIfCurrent(key, (ignored, value) -> true, currentValueAction);
}
public V getAndRunIfCurrent(K key, BiPredicate<K, V> actionRequired,
BiConsumer<K, V> currentValueAction) {
BiPredicate<K, V> required = Objects.requireNonNull(actionRequired, "actionRequired can not be null");
BiConsumer<K, V> action = Objects.requireNonNull(currentValueAction, "currentValueAction can not be null");
Function<K, V> loadFunction = Objects.requireNonNull(loader, "loader can not be null");
return useCache(() -> getAndRunIfCurrentCached(key, required, action, loadFunction),
() -> loadWithoutCaching(key, loadFunction));
}
private V getAndRunIfCurrentCached(K key, BiPredicate<K, V> required,
BiConsumer<K, V> action, Function<K, V> loadFunction) {
V cached = data.getIfPresent(key);
if (cached != null && !required.test(key, cached)) {
return cached;
}
StripeState<K> stripe = stripeState(key);
ActionToken<K> token;
synchronized (stripe) {
token = beginAction(stripe, key);
}
try {
AtomicBoolean actionPublished = new AtomicBoolean(false);
V value = cached == null
? data.getWithPublicationAction(key, loadKey -> loadAndPause(loadKey, loadFunction),
(loaded, commit) -> {
beforeCurrentValueActionForTest(key, loaded);
synchronized (stripe) {
if (isCurrent(stripe, token) && required.test(key, loaded)) {
commit.accept(() -> action.accept(key, loaded));
actionPublished.set(true);
} else {
commit.accept(() -> {
});
}
}
})
: cached;
if (value == null) {
return null;
}
if (actionPublished.get()) {
return value;
}
if (!required.test(key, value)) {
return value;
}
beforeCurrentValueActionForTest(key, value);
synchronized (stripe) {
boolean current = isCurrent(stripe, token)
&& (!effectiveEnabled || data.getIfPresent(key) == value);
if (current && required.test(key, value) && isCurrent(stripe, token)) {
try {
data.runIfCurrent(key, value, () -> action.accept(key, value));
} catch (RuntimeException | Error throwable) {
data.invalidateKey(key);
throw throwable;
}
}
}
return value;
} finally {
synchronized (stripe) {
endAction(stripe, token);
}
}
}
public V getIfPresent(K key) {
return useCache(() -> data.getIfPresent(key), () -> null);
}
@Nullable
public V findIfPresent(Predicate<K> keyPredicate) {
return useCache(() -> findIfPresentCached(keyPredicate), () -> null);
}
private V findIfPresentCached(Predicate<K> keyPredicate) {
Objects.requireNonNull(keyPredicate, "keyPredicate can not be null");
List<V> result = new ArrayList<>(1);
data.forEach((key, value) -> {
if (result.isEmpty() && keyPredicate.test(key)) {
result.add(value);
}
});
return result.isEmpty() ? null : result.get(0);
}
public void put(K key, V value) {
useCache(() -> putCached(key, value));
}
private void putCached(K key, V value) {
Objects.requireNonNull(value, "value can not be null");
StripeState<K> stripe = stripeState(key);
synchronized (stripe) {
bumpAction(stripe, key);
beforePublicMutationWriteForTest(key);
data.put(key, value);
}
}
public V compute(K key, BiFunction<K, V, V> remappingFunction) {
return computeAndRun(key, remappingFunction, () -> {
});
}
public V computeAndRun(K key, BiFunction<K, V, V> remappingFunction, Runnable afterMutation) {
return useCache(() -> computeAndRunCached(key, remappingFunction, afterMutation), () -> null);
}
private V computeAndRunCached(K key, BiFunction<K, V, V> remappingFunction, Runnable afterMutation) {
BiFunction<K, V, V> remapper = Objects.requireNonNull(remappingFunction, "remappingFunction can not be null");
Runnable action = Objects.requireNonNull(afterMutation, "afterMutation can not be null");
StripeState<K> stripe = stripeState(key);
synchronized (stripe) {
bumpAction(stripe, key);
while (true) {
V current = data.getIfPresent(key);
V updated = effectiveEnabled ? remapper.apply(key, current) : null;
beforePublicMutationWriteForTest(key);
if (data.compareAndSet(key, current, updated)) {
action.run();
return updated;
}
}
}
}
public V computeAfterValidation(K key, BiFunction<K, V, V> remappingFunction, Runnable validationAction) {
return useCache(() -> computeAfterValidationCached(key, remappingFunction, validationAction), () -> null);
}
private V computeAfterValidationCached(K key, BiFunction<K, V, V> remappingFunction, Runnable validationAction) {
BiFunction<K, V, V> remapper = Objects.requireNonNull(remappingFunction, "remappingFunction can not be null");
Runnable validation = Objects.requireNonNull(validationAction, "validationAction can not be null");
StripeState<K> stripe = stripeState(key);
synchronized (stripe) {
while (true) {
V current = data.getIfPresent(key);
V updated = effectiveEnabled ? remapper.apply(key, current) : null;
validation.run();
bumpAction(stripe, key);
beforePublicMutationWriteForTest(key);
if (data.compareAndSet(key, current, updated)) {
return updated;
}
}
}
}
/**
* Runs the action only after the expected cache value is confirmed, while concurrent loads are prevented from
* publishing until both the action and the value update finish. The action runs while this entry's stripe and
* the shared-cache publication locks are held, so it must be short, non-blocking, and must not re-enter this
* entry. If it throws, the value is not updated, but completed external side effects are not rolled back.
*/
public V computeWithCommitAction(K key, BiFunction<K, V, V> remappingFunction, Runnable commitAction) {
return useCache(() -> computeWithCommitActionCached(key, remappingFunction, commitAction), () -> null);
}
private V computeWithCommitActionCached(K key, BiFunction<K, V, V> remappingFunction, Runnable commitAction) {
BiFunction<K, V, V> remapper = Objects.requireNonNull(remappingFunction, "remappingFunction can not be null");
Runnable action = Objects.requireNonNull(commitAction, "commitAction can not be null");
StripeState<K> stripe = stripeState(key);
synchronized (stripe) {
while (true) {
V current = data.getIfPresent(key);
V updated = effectiveEnabled ? remapper.apply(key, current) : null;
bumpAction(stripe, key);
beforePublicMutationWriteForTest(key);
if (data.compareAndSet(key, current, updated, action)) {
return updated;
}
}
}
}
public void invalidateKey(K key) {
invalidateKeyAndRun(key, () -> {
});
}
public void invalidateKeyAndRun(K key, Runnable afterInvalidation) {
Runnable action = Objects.requireNonNull(afterInvalidation, "afterInvalidation can not be null");
useCache(() -> {
StripeState<K> stripe = stripeState(key);
synchronized (stripe) {
bumpAction(stripe, key);
data.invalidateKey(key);
action.run();
}
return null;
}, () -> {
action.run();
return null;
});
}
public void invalidateAll() {
useCache(this::invalidateAllCached);
}
private void invalidateAllCached() {
for (int i = 0; i < stripeCount; i++) {
StripeState<K> stripe = stripeStates.get(i);
if (stripe != null) {
synchronized (stripe) {
if (stripe.activeActions != null) {
stripe.activeActions.values().forEach(state -> state.generation++);
}
}
}
}
owner.invalidateScope(invalidationScope);
}
/** Auxiliary state has the same accepted-operation lifetime as this exact physical registration. */
public void onRetired(Runnable cleanup) {
synchronized (lifecycleLock) {
if (retired || retirementCleanup != null) {
throw new IllegalStateException("Retirement cleanup must be bound once before retirement");
}
retirementCleanup = Objects.requireNonNull(cleanup, "cleanup");
}
}
private void releaseRegistration() {
if (registrationReleased.compareAndSet(false, true)) {
try {
owner.remove(data);
} finally {
if (retirementCleanup != null) {
retirementCleanup.run();
}
}
}
}
/**
* Stops accepting cached operations without waiting for a remote load. Accepted operations pin the physical
* entry until they finish; later reads use the loader without publishing cache values or ID side effects.
* Cleanup removes this exact registration, never the current database scope or a replacement instance.
*/
public void close() {
boolean remove;
synchronized (lifecycleLock) {
retired = true;
remove = activeUses == 0;
}
if (remove) {
releaseRegistration();
}
}
public void forEach(BiConsumer<K, V> consumer) {
useCache(() -> data.forEach(consumer));
}
private void useCache(Runnable operation) {
useCache(() -> {
operation.run();
return null;
}, () -> null);
}
private <R> R useCache(Supplier<R> operation, Supplier<R> uncached) {
boolean accepted;
synchronized (lifecycleLock) {
accepted = !retired;
if (accepted) {
activeUses++;
}
}
if (!accepted) {
return uncached.get();
}
try {
return operation.get();
} finally {
boolean remove;
synchronized (lifecycleLock) {
activeUses--;
remove = retired && activeUses == 0;
}
if (remove) {
releaseRegistration();
}
}
}
public MetaCacheEntryStats stats() {
return MetaCacheEntryStats.from(data);
}
public static int defaultObjectStripeCount() {
return Config.external_meta_cache_object_entry_lock_stripes;
}
public static int singleKeyStripeCount() {
return SINGLE_KEY_STRIPES;
}
int stripeCountForTest() {
return stripeCount;
}
int initializedStripeCountForTest() {
int initialized = 0;
for (int i = 0; i < stripeCount; i++) {
if (stripeStates.get(i) != null) {
initialized++;
}
}
return initialized;
}
int activeActionReferenceCountForTest() {
int references = 0;
for (int i = 0; i < stripeCount; i++) {
StripeState<K> stripe = stripeStates.get(i);
if (stripe != null) {
synchronized (stripe) {
if (stripe.activeActions != null) {
for (ActionState state : stripe.activeActions.values()) {
references += state.references;
}
}
}
}
}
return references;
}
void putSharedForTest(K key, V value) {
data.put(key, value);
}
void beforeManualCachePutForTest(K key, V loaded) {
}
void beforePublicMutationWriteForTest(K key) {
}
protected void beforeCurrentValueActionForTest(K key, V value) {
}
private V loadAndPause(K key, Function<K, V> loadFunction) {
V loaded = loadFunction.apply(key);
if (loaded != null) {
beforeManualCachePutForTest(key, loaded);
}
return loaded;
}
private V loadWithoutCaching(K key, Function<K, V> loadFunction) {
V loaded = loadFunction.apply(key);
if (loaded != null && discardedValueAction != null) {
discardedValueAction.accept(key, loaded);
}
return loaded;
}
private static com.github.benmanes.caffeine.cache.RemovalCause toCaffeineRemovalCause(
MetaCacheRemovalReason reason) {
return com.github.benmanes.caffeine.cache.RemovalCause.valueOf(reason.name());
}
private StripeState<K> stripeState(K key) {
int hash = key == null ? 0 : key.hashCode();
int index = (hash & Integer.MAX_VALUE) % stripeCount;
StripeState<K> current = stripeStates.get(index);
if (current != null) {
return current;
}
StripeState<K> created = new StripeState<>();
if (stripeStates.compareAndSet(index, null, created)) {
return created;
}
return stripeStates.get(index);
}
private ActionToken<K> beginAction(StripeState<K> stripe, K key) {
if (stripe.activeActions == null) {
stripe.activeActions = new HashMap<>();
}
ActionState state = stripe.activeActions.computeIfAbsent(key, ignored -> new ActionState());
state.references++;
return new ActionToken<>(key, state);
}
private boolean isCurrent(StripeState<K> stripe, ActionToken<K> token) {
return !retired && stripe.activeActions != null
&& stripe.activeActions.get(token.key) == token.state
&& token.state.generation == token.generation;
}
private void bumpAction(StripeState<K> stripe, K key) {
if (stripe.activeActions != null) {
ActionState state = stripe.activeActions.get(key);
if (state != null) {
state.generation++;
}
}
}
private void endAction(StripeState<K> stripe, ActionToken<K> token) {
ActionState state = stripe.activeActions.get(token.key);
if (state != token.state || state.references <= 0) {
throw new IllegalStateException("Invalid naming-cache action state");
}
state.references--;
if (state.references == 0) {
stripe.activeActions.remove(token.key);
if (stripe.activeActions.isEmpty()) {
stripe.activeActions = null;
}
}
}
}