924060929 commented on code in PR #66633: URL: https://github.com/apache/doris/pull/66633#discussion_r3861594191
########## fe/fe-core/src/main/java/org/apache/doris/datasource/metacache/FeMetaCacheEntry.java: ########## @@ -0,0 +1,486 @@ +// 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.ScopePath; +import org.apache.doris.connector.cache.ScopedMetaCache.CacheMetrics; + +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 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 = new CatalogMetaCache(); + private final MetaCache<K, V> data; + + 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")); + } + + private FeMetaCacheEntry(String name, @Nullable Function<K, V> loader, CacheSpec cacheSpec, + ExecutorService refreshExecutor, boolean autoRefresh, boolean contextualOnly, int stripeCount, + @Nullable RemovalListener<K, V> removalListener) { + 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.builder( + name, cacheSpec, ignored -> ScopePath.catalog()); + if (loader != null) { + builder.loader(key -> loadAndPause(key, loader)); + } + if (removalListener != null) { Review Comment: Fixed in 5c0c8571519. Disabled loads no longer invoke the cache-removal callback: unpublished cleanup is a separate discard hook, while the FE adapter only registers removal for values that actually became cache-owned. The returned database object therefore remains initialized and caller-owned. Added a production-listener regression test. ########## fe/fe-connector/fe-connector-cache/src/main/java/org/apache/doris/connector/cache/ScopedMetaCache.java: ########## @@ -0,0 +1,1176 @@ +// 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.connector.cache; + +import org.apache.doris.connector.cache.ScopedMetaCacheRegistry.CacheAddress; +import org.apache.doris.connector.cache.ScopedMetaCacheRegistry.PublicationState; +import org.apache.doris.connector.cache.ScopedMetaCacheRegistry.ScopeLease; +import org.apache.doris.connector.cache.ScopedMetaCacheRegistry.ScopeSnapshot; + +import com.github.benmanes.caffeine.cache.Cache; +import com.github.benmanes.caffeine.cache.Caffeine; +import com.github.benmanes.caffeine.cache.RemovalCause; +import com.github.benmanes.caffeine.cache.RemovalListener; +import com.github.benmanes.caffeine.cache.Ticker; +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.Logger; + +import java.math.BigInteger; +import java.time.Duration; +import java.util.ArrayDeque; +import java.util.Deque; +import java.util.HashMap; +import java.util.Map; +import java.util.NavigableMap; +import java.util.Objects; +import java.util.OptionalLong; +import java.util.TreeMap; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CompletionException; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; +import java.util.concurrent.Executor; +import java.util.concurrent.RejectedExecutionException; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; +import java.util.concurrent.atomic.LongAdder; +import java.util.function.BiConsumer; +import java.util.function.BooleanSupplier; +import java.util.function.Consumer; +import java.util.function.Function; + +/** + * One physical Caffeine cache participating in a {@link ScopedMetaCacheRegistry}. + * + * <p>Every value is wrapped with both its hierarchical scope-state identities and an exact-key state. Hierarchical + * invalidation can therefore detach a whole catalog/database/table/partition subtree, while exact-key invalidation + * fences only one physical cache key. Removal listeners conditionally remove the exact wrapper from its original + * scope bucket and key node, so delayed callbacks cannot delete a replacement. + */ +public final class ScopedMetaCache<K, V> implements AutoCloseable { + private static final Logger LOG = LogManager.getLogger(ScopedMetaCache.class); + private static final Runnable NO_OP = () -> { + }; + + private final ScopedMetaCacheRegistry registry; + private final String name; + private final boolean effectiveEnabled; + private final Cache<K, VersionedValue<K, V>> data; + private final ConcurrentMap<K, KeyNode<K, V>> keyNodes = new ConcurrentHashMap<>(); + private final ConcurrentMap<LoadAddress<K>, CompletableFuture<V>> inFlightLoads = new ConcurrentHashMap<>(); + private final ConcurrentMap<K, VersionedValue<K, V>> refreshing = new ConcurrentHashMap<>(); + private final StripedPhaseGate bulkInvalidationGate = new StripedPhaseGate(); + private final Map<K, BigInteger> exactInvalidations = new HashMap<>(); + private final NavigableMap<BigInteger, Integer> activeBulkStarts = new TreeMap<>(); + private final AtomicBoolean closed = new AtomicBoolean(false); + private final LongAdder requestCount = new LongAdder(); + private final LongAdder hitCount = new LongAdder(); + private final LongAdder missCount = new LongAdder(); + private final LongAdder loadSuccessCount = new LongAdder(); + private final LongAdder loadFailureCount = new LongAdder(); + private final LongAdder totalLoadTimeNanos = new LongAdder(); + private final LongAdder evictionCount = new LongAdder(); + private final LongAdder invalidateCount = new LongAdder(); + private final AtomicReference<Long> lastLoadSuccessTimeMs = new AtomicReference<>(-1L); + private final AtomicReference<Long> lastLoadFailureTimeMs = new AtomicReference<>(-1L); + private final AtomicReference<String> lastError = new AtomicReference<>(""); + private final RemovalListener<K, V> beforeRemoval; + private final Ticker ticker; + private final long refreshAfterWriteNanos; + private final Executor refreshExecutor; + private final Runnable afterLoadElection; + private final Runnable afterBulkStage; + private final Runnable afterRefreshRegistration; + private final ThreadLocal<RemovalDeferral<K, V>> removalDeferrals = + ThreadLocal.withInitial(RemovalDeferral::new); + private BigInteger exactInvalidationSequence = BigInteger.ZERO; + + ScopedMetaCache( + ScopedMetaCacheRegistry registry, + String name, + CacheSpec cacheSpec, + Ticker ticker, + RemovalListener<K, V> beforeRemoval, + Duration refreshAfterWrite, + Executor refreshExecutor, + Runnable afterLoadElection, + Runnable afterBulkStage, + Runnable afterRefreshRegistration) { + this.registry = Objects.requireNonNull(registry, "registry can not be null"); + this.name = Objects.requireNonNull(name, "name can not be null"); + Objects.requireNonNull(cacheSpec, "cacheSpec can not be null"); + this.beforeRemoval = beforeRemoval; + this.ticker = ticker == null ? Ticker.systemTicker() : ticker; + this.refreshAfterWriteNanos = refreshAfterWrite == null ? 0L : refreshAfterWrite.toNanos(); + this.refreshExecutor = refreshExecutor; + this.afterLoadElection = + Objects.requireNonNull(afterLoadElection, "afterLoadElection can not be null"); + this.afterBulkStage = Objects.requireNonNull(afterBulkStage, "afterBulkStage can not be null"); + this.afterRefreshRegistration = Objects.requireNonNull( + afterRefreshRegistration, "afterRefreshRegistration can not be null"); + this.effectiveEnabled = CacheSpec.isCacheEnabled( + cacheSpec.isEnable(), cacheSpec.getTtlSecond(), cacheSpec.getCapacity()); + + Caffeine<Object, Object> builder = Caffeine.newBuilder() + .maximumSize(effectiveEnabled ? cacheSpec.getCapacity() : 0L) + .executor(Runnable::run) + .removalListener(this::onRemoval); + OptionalLong expiry = effectiveEnabled + ? CacheSpec.toExpireAfterAccess(cacheSpec.getTtlSecond()) + : OptionalLong.empty(); + if (expiry.isPresent()) { + builder.expireAfterAccess(Duration.ofSeconds(expiry.getAsLong())); + } + if (ticker != null) { + builder.ticker(this.ticker); + } + this.data = builder.build(); + } + + public String name() { + return name; + } + + public V get(K key, ScopePath path, Function<K, V> loader) { + return getWithPublicationAction(key, path, loader, + (loaded, commit) -> commit.accept(NO_OP)); + } + + public V getWithPublicationAction(K key, ScopePath path, Function<K, V> loader, + BiConsumer<V, Consumer<Runnable>> publicationCoordinator) { + Objects.requireNonNull(key, "key can not be null"); + Objects.requireNonNull(path, "path can not be null"); + Function<K, V> loadFunction = Objects.requireNonNull(loader, "loader can not be null"); + BiConsumer<V, Consumer<Runnable>> coordinator = Objects.requireNonNull( + publicationCoordinator, "publicationCoordinator can not be null"); + checkOpen(); + if (!effectiveEnabled) { + recordAccess(false); + V loaded = loadAndRecord(key, loadFunction); + if (loaded != null) { + AtomicBoolean commitInvoked = new AtomicBoolean(false); + try { + coordinator.accept(loaded, beforePublication -> { + if (!commitInvoked.compareAndSet(false, true)) { + throw new IllegalStateException("Metadata cache publication callback was invoked twice"); + } + Objects.requireNonNull(beforePublication, "beforePublication can not be null").run(); + }); + if (!commitInvoked.get()) { + throw new IllegalStateException("Metadata cache publication callback was not invoked"); + } + } finally { + notifyUnpublishedRemoval(key, loaded); + } + } + return loaded; + } + + VersionedValue<K, V> presentVersioned = currentVersionedValue(key, path); + if (presentVersioned != null) { + recordAccess(true); + scheduleRefresh(key, path, loader, presentVersioned); + return presentVersioned.value; + } + recordAccess(false); + try (PublicationLease<K, V> lease = acquirePublicationLease(key, path, true)) { + LoadAddress<K> loadAddress = new LoadAddress<>(key, path, lease); + CompletableFuture<V> ownLoad = new CompletableFuture<>(); + CompletableFuture<V> existingLoad = inFlightLoads.putIfAbsent(loadAddress, ownLoad); + if (existingLoad != null) { + return awaitLoad(existingLoad); + } + try { + afterLoadElection.run(); + AtomicReference<VersionedValue<K, V>> electedValue = new AtomicReference<>(); + boolean electedValuePresent = deferRemovals(() -> { + VersionedValue<K, V> present; + synchronized (lease.keyNode) { + present = currentVersionedValue(key, path); + } + electedValue.set(present); + return present != null; + }); + if (electedValuePresent) { + VersionedValue<K, V> present = electedValue.get(); + ownLoad.complete(present.value); + return present.value; + } + V loaded = loadAndRecord(key, loadFunction); + if (loaded != null) { + AtomicBoolean commitInvoked = new AtomicBoolean(false); + AtomicBoolean retained = new AtomicBoolean(false); + try { + coordinator.accept(loaded, beforePublication -> { + if (!commitInvoked.compareAndSet(false, true)) { + throw new IllegalStateException( + "Metadata cache publication callback was invoked twice"); + } + retained.set(commitLoaded(lease, key, loaded, beforePublication)); + }); + if (!commitInvoked.get()) { + throw new IllegalStateException("Metadata cache publication callback was not invoked"); + } + } finally { + if (!retained.get()) { + notifyUnpublishedRemoval(key, loaded); Review Comment: Fixed in 5c0c8571519. A miss rejected by concurrent invalidation now stays caller-owned and is returned without invoking the published-value removal listener. Added a deterministic invalidation/publication race test in FeMetaCacheEntryTest and updated the generic concurrency contract. ########## fe/fe-connector/fe-connector-cache/src/main/java/org/apache/doris/connector/cache/ScopedMetaCache.java: ########## @@ -0,0 +1,1176 @@ +// 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.connector.cache; + +import org.apache.doris.connector.cache.ScopedMetaCacheRegistry.CacheAddress; +import org.apache.doris.connector.cache.ScopedMetaCacheRegistry.PublicationState; +import org.apache.doris.connector.cache.ScopedMetaCacheRegistry.ScopeLease; +import org.apache.doris.connector.cache.ScopedMetaCacheRegistry.ScopeSnapshot; + +import com.github.benmanes.caffeine.cache.Cache; +import com.github.benmanes.caffeine.cache.Caffeine; +import com.github.benmanes.caffeine.cache.RemovalCause; +import com.github.benmanes.caffeine.cache.RemovalListener; +import com.github.benmanes.caffeine.cache.Ticker; +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.Logger; + +import java.math.BigInteger; +import java.time.Duration; +import java.util.ArrayDeque; +import java.util.Deque; +import java.util.HashMap; +import java.util.Map; +import java.util.NavigableMap; +import java.util.Objects; +import java.util.OptionalLong; +import java.util.TreeMap; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CompletionException; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; +import java.util.concurrent.Executor; +import java.util.concurrent.RejectedExecutionException; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; +import java.util.concurrent.atomic.LongAdder; +import java.util.function.BiConsumer; +import java.util.function.BooleanSupplier; +import java.util.function.Consumer; +import java.util.function.Function; + +/** + * One physical Caffeine cache participating in a {@link ScopedMetaCacheRegistry}. + * + * <p>Every value is wrapped with both its hierarchical scope-state identities and an exact-key state. Hierarchical + * invalidation can therefore detach a whole catalog/database/table/partition subtree, while exact-key invalidation + * fences only one physical cache key. Removal listeners conditionally remove the exact wrapper from its original + * scope bucket and key node, so delayed callbacks cannot delete a replacement. + */ +public final class ScopedMetaCache<K, V> implements AutoCloseable { + private static final Logger LOG = LogManager.getLogger(ScopedMetaCache.class); + private static final Runnable NO_OP = () -> { + }; + + private final ScopedMetaCacheRegistry registry; + private final String name; + private final boolean effectiveEnabled; + private final Cache<K, VersionedValue<K, V>> data; + private final ConcurrentMap<K, KeyNode<K, V>> keyNodes = new ConcurrentHashMap<>(); + private final ConcurrentMap<LoadAddress<K>, CompletableFuture<V>> inFlightLoads = new ConcurrentHashMap<>(); + private final ConcurrentMap<K, VersionedValue<K, V>> refreshing = new ConcurrentHashMap<>(); + private final StripedPhaseGate bulkInvalidationGate = new StripedPhaseGate(); + private final Map<K, BigInteger> exactInvalidations = new HashMap<>(); + private final NavigableMap<BigInteger, Integer> activeBulkStarts = new TreeMap<>(); + private final AtomicBoolean closed = new AtomicBoolean(false); + private final LongAdder requestCount = new LongAdder(); + private final LongAdder hitCount = new LongAdder(); + private final LongAdder missCount = new LongAdder(); + private final LongAdder loadSuccessCount = new LongAdder(); + private final LongAdder loadFailureCount = new LongAdder(); + private final LongAdder totalLoadTimeNanos = new LongAdder(); + private final LongAdder evictionCount = new LongAdder(); + private final LongAdder invalidateCount = new LongAdder(); + private final AtomicReference<Long> lastLoadSuccessTimeMs = new AtomicReference<>(-1L); + private final AtomicReference<Long> lastLoadFailureTimeMs = new AtomicReference<>(-1L); + private final AtomicReference<String> lastError = new AtomicReference<>(""); + private final RemovalListener<K, V> beforeRemoval; + private final Ticker ticker; + private final long refreshAfterWriteNanos; + private final Executor refreshExecutor; + private final Runnable afterLoadElection; + private final Runnable afterBulkStage; + private final Runnable afterRefreshRegistration; + private final ThreadLocal<RemovalDeferral<K, V>> removalDeferrals = + ThreadLocal.withInitial(RemovalDeferral::new); + private BigInteger exactInvalidationSequence = BigInteger.ZERO; + + ScopedMetaCache( + ScopedMetaCacheRegistry registry, + String name, + CacheSpec cacheSpec, + Ticker ticker, + RemovalListener<K, V> beforeRemoval, + Duration refreshAfterWrite, + Executor refreshExecutor, + Runnable afterLoadElection, + Runnable afterBulkStage, + Runnable afterRefreshRegistration) { + this.registry = Objects.requireNonNull(registry, "registry can not be null"); + this.name = Objects.requireNonNull(name, "name can not be null"); + Objects.requireNonNull(cacheSpec, "cacheSpec can not be null"); + this.beforeRemoval = beforeRemoval; + this.ticker = ticker == null ? Ticker.systemTicker() : ticker; + this.refreshAfterWriteNanos = refreshAfterWrite == null ? 0L : refreshAfterWrite.toNanos(); + this.refreshExecutor = refreshExecutor; + this.afterLoadElection = + Objects.requireNonNull(afterLoadElection, "afterLoadElection can not be null"); + this.afterBulkStage = Objects.requireNonNull(afterBulkStage, "afterBulkStage can not be null"); + this.afterRefreshRegistration = Objects.requireNonNull( + afterRefreshRegistration, "afterRefreshRegistration can not be null"); + this.effectiveEnabled = CacheSpec.isCacheEnabled( + cacheSpec.isEnable(), cacheSpec.getTtlSecond(), cacheSpec.getCapacity()); + + Caffeine<Object, Object> builder = Caffeine.newBuilder() + .maximumSize(effectiveEnabled ? cacheSpec.getCapacity() : 0L) + .executor(Runnable::run) + .removalListener(this::onRemoval); + OptionalLong expiry = effectiveEnabled + ? CacheSpec.toExpireAfterAccess(cacheSpec.getTtlSecond()) + : OptionalLong.empty(); + if (expiry.isPresent()) { + builder.expireAfterAccess(Duration.ofSeconds(expiry.getAsLong())); + } + if (ticker != null) { + builder.ticker(this.ticker); + } + this.data = builder.build(); + } + + public String name() { + return name; + } + + public V get(K key, ScopePath path, Function<K, V> loader) { + return getWithPublicationAction(key, path, loader, + (loaded, commit) -> commit.accept(NO_OP)); + } + + public V getWithPublicationAction(K key, ScopePath path, Function<K, V> loader, + BiConsumer<V, Consumer<Runnable>> publicationCoordinator) { + Objects.requireNonNull(key, "key can not be null"); + Objects.requireNonNull(path, "path can not be null"); + Function<K, V> loadFunction = Objects.requireNonNull(loader, "loader can not be null"); + BiConsumer<V, Consumer<Runnable>> coordinator = Objects.requireNonNull( + publicationCoordinator, "publicationCoordinator can not be null"); + checkOpen(); + if (!effectiveEnabled) { + recordAccess(false); + V loaded = loadAndRecord(key, loadFunction); + if (loaded != null) { + AtomicBoolean commitInvoked = new AtomicBoolean(false); + try { + coordinator.accept(loaded, beforePublication -> { + if (!commitInvoked.compareAndSet(false, true)) { + throw new IllegalStateException("Metadata cache publication callback was invoked twice"); + } + Objects.requireNonNull(beforePublication, "beforePublication can not be null").run(); + }); + if (!commitInvoked.get()) { + throw new IllegalStateException("Metadata cache publication callback was not invoked"); + } + } finally { + notifyUnpublishedRemoval(key, loaded); + } + } + return loaded; + } + + VersionedValue<K, V> presentVersioned = currentVersionedValue(key, path); + if (presentVersioned != null) { + recordAccess(true); + scheduleRefresh(key, path, loader, presentVersioned); + return presentVersioned.value; + } + recordAccess(false); + try (PublicationLease<K, V> lease = acquirePublicationLease(key, path, true)) { + LoadAddress<K> loadAddress = new LoadAddress<>(key, path, lease); + CompletableFuture<V> ownLoad = new CompletableFuture<>(); + CompletableFuture<V> existingLoad = inFlightLoads.putIfAbsent(loadAddress, ownLoad); + if (existingLoad != null) { + return awaitLoad(existingLoad); + } + try { + afterLoadElection.run(); + AtomicReference<VersionedValue<K, V>> electedValue = new AtomicReference<>(); + boolean electedValuePresent = deferRemovals(() -> { + VersionedValue<K, V> present; + synchronized (lease.keyNode) { + present = currentVersionedValue(key, path); + } + electedValue.set(present); + return present != null; + }); + if (electedValuePresent) { + VersionedValue<K, V> present = electedValue.get(); + ownLoad.complete(present.value); + return present.value; + } + V loaded = loadAndRecord(key, loadFunction); + if (loaded != null) { + AtomicBoolean commitInvoked = new AtomicBoolean(false); + AtomicBoolean retained = new AtomicBoolean(false); + try { + coordinator.accept(loaded, beforePublication -> { + if (!commitInvoked.compareAndSet(false, true)) { + throw new IllegalStateException( + "Metadata cache publication callback was invoked twice"); + } + retained.set(commitLoaded(lease, key, loaded, beforePublication)); + }); + if (!commitInvoked.get()) { + throw new IllegalStateException("Metadata cache publication callback was not invoked"); + } + } finally { + if (!retained.get()) { + notifyUnpublishedRemoval(key, loaded); + } + } + } + ownLoad.complete(loaded); + return loaded; + } catch (RuntimeException | Error throwable) { + ownLoad.completeExceptionally(throwable); + throw throwable; + } finally { + inFlightLoads.remove(loadAddress, ownLoad); + } + } + } + + public V getIfPresent(K key, ScopePath path) { + Objects.requireNonNull(key, "key can not be null"); + Objects.requireNonNull(path, "path can not be null"); + checkOpen(); + if (!effectiveEnabled) { + recordAccess(false); + return null; + } + VersionedValue<K, V> versioned = currentVersionedValue(key, path); + recordAccess(versioned != null); + return versioned == null ? null : versioned.value; + } + + private VersionedValue<K, V> currentVersionedValue(K key, ScopePath path) { + VersionedValue<K, V> versioned = data.getIfPresent(key); + if (versioned == null) { + return null; + } + if (!versioned.scopeSnapshot.path().equals(path)) { + return null; + } + if (!versioned.isCurrent(registry, keyNodes)) { + data.asMap().remove(key, versioned); + return null; + } + return versioned; + } + + public void put(K key, ScopePath path, V value) { + Objects.requireNonNull(key, "key can not be null"); + Objects.requireNonNull(path, "path can not be null"); + Objects.requireNonNull(value, "value can not be null"); + checkOpen(); + if (!effectiveEnabled) { + return; + } + try (PublicationLease<K, V> lease = acquirePublicationLease(key, path, false)) { + guardedCommit(lease, () -> { + lease.keyNode.loadPublicationState.set(new Object()); + return publishCommitted(lease, key, value) != null; + }); + } + } + + public boolean compareAndSet(K key, ScopePath path, V expectedValue, V updatedValue) { + Objects.requireNonNull(key, "key can not be null"); + Objects.requireNonNull(path, "path can not be null"); + checkOpen(); + if (!effectiveEnabled) { + return true; + } + try (PublicationLease<K, V> lease = acquirePublicationLease(key, path, false)) { + return guardedCommit(lease, () -> { + VersionedValue<K, V> current = currentVersionedValue(key, path); + V currentValue = current == null ? null : current.value; + if (currentValue != expectedValue) { + return false; + } + lease.keyNode.loadPublicationState.set(new Object()); + if (updatedValue == currentValue) { + return true; + } + if (updatedValue == null) { + if (current != null) { + data.asMap().remove(key, current); + lease.keyNode.registration.compareAndSet(current, null); + } + } else { + publishCommitted(lease, key, updatedValue); + } + return true; + }); + } + } + + public void invalidateKey(K key) { + invalidateKey(key, NO_OP, NO_OP); + } + + void invalidateKey(K key, Runnable afterStateReplacement) { + invalidateKey(key, NO_OP, afterStateReplacement); + } + + void invalidateKey( + K key, Runnable beforeInvalidationLock, Runnable afterStateReplacement) { + Objects.requireNonNull(key, "key can not be null"); + Objects.requireNonNull(beforeInvalidationLock, "beforeInvalidationLock can not be null"); + Objects.requireNonNull(afterStateReplacement, "afterStateReplacement can not be null"); + checkOpen(); + beforeInvalidationLock.run(); + InvalidatedKey<K, V> invalidated = bulkInvalidationGate.write(() -> { + if (closed.get()) { + return null; + } + exactInvalidationSequence = exactInvalidationSequence.add(BigInteger.ONE); + if (!activeBulkStarts.isEmpty()) { + exactInvalidations.put(key, exactInvalidationSequence); + } + KeyNode<K, V> node = keyNodes.get(key); + KeyState invalidatedState = null; + if (node != null) { + invalidatedState = replaceKeyState(node); + } + return new InvalidatedKey<>(node, invalidatedState); + }); + if (invalidated == null || invalidated.node == null) { + return; + } + afterStateReplacement.run(); + VersionedValue<K, V> registered = invalidated.node.registration.get(); + if (registered != null && registered.keyState == invalidated.keyState) { + data.asMap().remove(key, registered); + invalidated.node.registration.compareAndSet(registered, null); + } + tryPruneKey(key, invalidated.node); + } + + public BulkLoadHandle beginBulkLoad(ScopePath parentScope) { + Objects.requireNonNull(parentScope, "parentScope can not be null"); + checkOpen(); + if (!effectiveEnabled) { + return BulkLoadHandle.disabled(this, parentScope); + } + ScopeLease scopeLease = registry.acquire(parentScope); + BigInteger exactSequence = bulkInvalidationGate.write(() -> { + if (closed.get()) { + scopeLease.close(); + throw new IllegalStateException("Scoped meta cache '" + name + "' is closed"); + } + BigInteger sequence = exactInvalidationSequence; + activeBulkStarts.merge(sequence, 1, Integer::sum); + return sequence; + }); + return new BulkLoadHandle( + this, + parentScope, + scopeLease, + scopeLease.publicationState(), + exactSequence); + } + + public boolean publish( + BulkLoadHandle handle, K key, ScopePath actualScope, V value) { + Objects.requireNonNull(handle, "handle can not be null"); + Objects.requireNonNull(key, "key can not be null"); + Objects.requireNonNull(actualScope, "actualScope can not be null"); + Objects.requireNonNull(value, "value can not be null"); + checkOpen(); + handle.checkOwner(this); + if (!handle.parentScope.contains(actualScope)) { + throw new IllegalArgumentException( + "Actual scope " + actualScope + " is outside bulk-load parent " + handle.parentScope); + } + if (!effectiveEnabled) { + return false; + } + try (PublicationLease<K, V> lease = acquirePublicationLease(key, actualScope, false)) { + VersionedValue<K, V> staged = newVersionedValue(lease, key, value); + afterBulkStage.run(); + return handle.tryCommit(key, lease, staged); + } + } + + public CacheMetrics metrics() { + return bulkInvalidationGate.read(() -> new CacheMetrics( + data.estimatedSize(), + keyNodes.size(), + inFlightLoads.size(), + activeBulkStarts.values().stream().mapToInt(Integer::intValue).sum(), + exactInvalidations.size(), + effectiveEnabled, + requestCount.sum(), + hitCount.sum(), + missCount.sum(), + loadSuccessCount.sum(), + loadFailureCount.sum(), + totalLoadTimeNanos.sum(), + evictionCount.sum(), + invalidateCount.sum(), + lastLoadSuccessTimeMs.get(), + lastLoadFailureTimeMs.get(), + lastError.get())); + } + + int refreshingCountForTest() { + return refreshing.size(); + } + + public void forEach(BiConsumer<K, V> consumer) { + Objects.requireNonNull(consumer, "consumer can not be null"); + data.asMap().forEach((key, versioned) -> { + if (versioned.isCurrent(registry, keyNodes)) { + consumer.accept(key, versioned.value); + } + }); + } + + private V loadAndRecord(K key, Function<K, V> loader) { + long startNanos = System.nanoTime(); + try { + V loaded = loader.apply(key); + loadSuccessCount.increment(); + lastLoadSuccessTimeMs.set(System.currentTimeMillis()); + return loaded; + } catch (RuntimeException | Error throwable) { + loadFailureCount.increment(); + lastLoadFailureTimeMs.set(System.currentTimeMillis()); + lastError.set(throwable.toString()); + throw throwable; + } finally { + totalLoadTimeNanos.add(System.nanoTime() - startNanos); + } + } + + private void recordAccess(boolean hit) { + requestCount.increment(); + if (hit) { + hitCount.increment(); + } else { + missCount.increment(); + } + } + + public void cleanUp() { + data.cleanUp(); + } + + @Override + public void close() { + if (!closed.compareAndSet(false, true)) { + return; + } + registry.removeCache(this); + closePhysicalState(); + } + + void closeFromRegistry() { + if (closed.compareAndSet(false, true)) { + closePhysicalState(); + } + } + + void removeExpectedRaw(Object rawKey, Object expectedValue) { + @SuppressWarnings("unchecked") + K key = (K) rawKey; + @SuppressWarnings("unchecked") + VersionedValue<K, V> versionedValue = (VersionedValue<K, V>) expectedValue; + data.asMap().remove(key, versionedValue); + } + + private PublicationLease<K, V> acquirePublicationLease( + K key, ScopePath path, boolean fenceAgainstDirectPublication) { + while (true) { + ScopeLease scopeLease = registry.acquire(path); + KeyNode<K, V> keyNode = keyNodes.computeIfAbsent(key, ignored -> new KeyNode<>()); + keyNode.activeLoads.incrementAndGet(); + KeyState keyState = keyNode.current.get(); + Object loadPublicationState = + fenceAgainstDirectPublication ? keyNode.loadPublicationState.get() : null; + if (keyNodes.get(key) == keyNode && scopeLease.isCurrent()) { + return new PublicationLease<>( + this, key, scopeLease, keyNode, keyState, loadPublicationState); + } + releaseKey(key, keyNode); + scopeLease.close(); + } + } + + private VersionedValue<K, V> publish(PublicationLease<K, V> lease, K key, V value) { + if (!lease.isCurrent()) { + return null; + } + VersionedValue<K, V> versioned = newVersionedValue(lease, key, value); + install(versioned, lease); + if (!lease.isCurrent()) { + data.asMap().remove(key, versioned); + return null; + } + return versioned; + } + + private VersionedValue<K, V> publishCommitted(PublicationLease<K, V> lease, K key, V value) { + return publish(lease, key, value); + } + + private boolean commitLoaded(PublicationLease<K, V> lease, K key, V value, Runnable beforePublication) { + Runnable action = Objects.requireNonNull(beforePublication, "beforePublication can not be null"); + return guardedCommit(lease, () -> { + action.run(); + return publishCommitted(lease, key, value) != null; + }); + } + + private boolean guardedCommit(PublicationLease<K, V> lease, BooleanSupplier commitAction) { + return deferRemovals(() -> bulkInvalidationGate.readBoolean( + () -> lease.scopeLease.commitIfPublicationCurrent( + lease.scopePublicationState, () -> { + synchronized (lease.keyNode) { + return lease.isCurrent() && commitAction.getAsBoolean(); + } + }))); + } + + private VersionedValue<K, V> newVersionedValue( + PublicationLease<K, V> lease, K key, V value) { + CacheAddress address = new CacheAddress(this, key); + return new VersionedValue<>( + key, value, address, lease.scopeLease.snapshot(), lease.keyNode, lease.keyState, ticker.read()); + } + + private void scheduleRefresh(K key, ScopePath path, Function<K, V> loader, VersionedValue<K, V> current) { + if (refreshAfterWriteNanos == 0L || ticker.read() - current.writeTimeNanos < refreshAfterWriteNanos + || refreshing.putIfAbsent(key, current) != null) { + return; + } + PublicationLease<K, V> lease; + try { + afterRefreshRegistration.run(); + lease = acquirePublicationLease(key, path, true); + } catch (RuntimeException | Error throwable) { + refreshing.remove(key, current); + throw throwable; + } + if (data.getIfPresent(key) != current || !current.isCurrent(registry, keyNodes)) { + lease.close(); + refreshing.remove(key, current); + return; + } + try { + refreshExecutor.execute(() -> { Review Comment: Fixed in 5c0c8571519. Refresh publication admission is now acquired inside the executor task rather than before enqueue, and close clears the refresh marker. A queued task discarded by shutdownNow therefore owns no lease or publication state. Added a deterministic queued-task close test. ########## fe/fe-connector/fe-connector-cache/src/main/java/org/apache/doris/connector/cache/ScopedMetaCache.java: ########## @@ -0,0 +1,1176 @@ +// 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.connector.cache; + +import org.apache.doris.connector.cache.ScopedMetaCacheRegistry.CacheAddress; +import org.apache.doris.connector.cache.ScopedMetaCacheRegistry.PublicationState; +import org.apache.doris.connector.cache.ScopedMetaCacheRegistry.ScopeLease; +import org.apache.doris.connector.cache.ScopedMetaCacheRegistry.ScopeSnapshot; + +import com.github.benmanes.caffeine.cache.Cache; +import com.github.benmanes.caffeine.cache.Caffeine; +import com.github.benmanes.caffeine.cache.RemovalCause; +import com.github.benmanes.caffeine.cache.RemovalListener; +import com.github.benmanes.caffeine.cache.Ticker; +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.Logger; + +import java.math.BigInteger; +import java.time.Duration; +import java.util.ArrayDeque; +import java.util.Deque; +import java.util.HashMap; +import java.util.Map; +import java.util.NavigableMap; +import java.util.Objects; +import java.util.OptionalLong; +import java.util.TreeMap; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CompletionException; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; +import java.util.concurrent.Executor; +import java.util.concurrent.RejectedExecutionException; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; +import java.util.concurrent.atomic.LongAdder; +import java.util.function.BiConsumer; +import java.util.function.BooleanSupplier; +import java.util.function.Consumer; +import java.util.function.Function; + +/** + * One physical Caffeine cache participating in a {@link ScopedMetaCacheRegistry}. + * + * <p>Every value is wrapped with both its hierarchical scope-state identities and an exact-key state. Hierarchical + * invalidation can therefore detach a whole catalog/database/table/partition subtree, while exact-key invalidation + * fences only one physical cache key. Removal listeners conditionally remove the exact wrapper from its original + * scope bucket and key node, so delayed callbacks cannot delete a replacement. + */ +public final class ScopedMetaCache<K, V> implements AutoCloseable { + private static final Logger LOG = LogManager.getLogger(ScopedMetaCache.class); + private static final Runnable NO_OP = () -> { + }; + + private final ScopedMetaCacheRegistry registry; + private final String name; + private final boolean effectiveEnabled; + private final Cache<K, VersionedValue<K, V>> data; + private final ConcurrentMap<K, KeyNode<K, V>> keyNodes = new ConcurrentHashMap<>(); + private final ConcurrentMap<LoadAddress<K>, CompletableFuture<V>> inFlightLoads = new ConcurrentHashMap<>(); + private final ConcurrentMap<K, VersionedValue<K, V>> refreshing = new ConcurrentHashMap<>(); + private final StripedPhaseGate bulkInvalidationGate = new StripedPhaseGate(); + private final Map<K, BigInteger> exactInvalidations = new HashMap<>(); + private final NavigableMap<BigInteger, Integer> activeBulkStarts = new TreeMap<>(); + private final AtomicBoolean closed = new AtomicBoolean(false); + private final LongAdder requestCount = new LongAdder(); + private final LongAdder hitCount = new LongAdder(); + private final LongAdder missCount = new LongAdder(); + private final LongAdder loadSuccessCount = new LongAdder(); + private final LongAdder loadFailureCount = new LongAdder(); + private final LongAdder totalLoadTimeNanos = new LongAdder(); + private final LongAdder evictionCount = new LongAdder(); + private final LongAdder invalidateCount = new LongAdder(); + private final AtomicReference<Long> lastLoadSuccessTimeMs = new AtomicReference<>(-1L); + private final AtomicReference<Long> lastLoadFailureTimeMs = new AtomicReference<>(-1L); + private final AtomicReference<String> lastError = new AtomicReference<>(""); + private final RemovalListener<K, V> beforeRemoval; + private final Ticker ticker; + private final long refreshAfterWriteNanos; + private final Executor refreshExecutor; + private final Runnable afterLoadElection; + private final Runnable afterBulkStage; + private final Runnable afterRefreshRegistration; + private final ThreadLocal<RemovalDeferral<K, V>> removalDeferrals = + ThreadLocal.withInitial(RemovalDeferral::new); + private BigInteger exactInvalidationSequence = BigInteger.ZERO; + + ScopedMetaCache( + ScopedMetaCacheRegistry registry, + String name, + CacheSpec cacheSpec, + Ticker ticker, + RemovalListener<K, V> beforeRemoval, + Duration refreshAfterWrite, + Executor refreshExecutor, + Runnable afterLoadElection, + Runnable afterBulkStage, + Runnable afterRefreshRegistration) { + this.registry = Objects.requireNonNull(registry, "registry can not be null"); + this.name = Objects.requireNonNull(name, "name can not be null"); + Objects.requireNonNull(cacheSpec, "cacheSpec can not be null"); + this.beforeRemoval = beforeRemoval; + this.ticker = ticker == null ? Ticker.systemTicker() : ticker; + this.refreshAfterWriteNanos = refreshAfterWrite == null ? 0L : refreshAfterWrite.toNanos(); + this.refreshExecutor = refreshExecutor; + this.afterLoadElection = + Objects.requireNonNull(afterLoadElection, "afterLoadElection can not be null"); + this.afterBulkStage = Objects.requireNonNull(afterBulkStage, "afterBulkStage can not be null"); + this.afterRefreshRegistration = Objects.requireNonNull( + afterRefreshRegistration, "afterRefreshRegistration can not be null"); + this.effectiveEnabled = CacheSpec.isCacheEnabled( + cacheSpec.isEnable(), cacheSpec.getTtlSecond(), cacheSpec.getCapacity()); + + Caffeine<Object, Object> builder = Caffeine.newBuilder() + .maximumSize(effectiveEnabled ? cacheSpec.getCapacity() : 0L) + .executor(Runnable::run) + .removalListener(this::onRemoval); + OptionalLong expiry = effectiveEnabled + ? CacheSpec.toExpireAfterAccess(cacheSpec.getTtlSecond()) + : OptionalLong.empty(); + if (expiry.isPresent()) { + builder.expireAfterAccess(Duration.ofSeconds(expiry.getAsLong())); + } + if (ticker != null) { + builder.ticker(this.ticker); + } + this.data = builder.build(); + } + + public String name() { + return name; + } + + public V get(K key, ScopePath path, Function<K, V> loader) { + return getWithPublicationAction(key, path, loader, + (loaded, commit) -> commit.accept(NO_OP)); + } + + public V getWithPublicationAction(K key, ScopePath path, Function<K, V> loader, + BiConsumer<V, Consumer<Runnable>> publicationCoordinator) { + Objects.requireNonNull(key, "key can not be null"); + Objects.requireNonNull(path, "path can not be null"); + Function<K, V> loadFunction = Objects.requireNonNull(loader, "loader can not be null"); + BiConsumer<V, Consumer<Runnable>> coordinator = Objects.requireNonNull( + publicationCoordinator, "publicationCoordinator can not be null"); + checkOpen(); + if (!effectiveEnabled) { + recordAccess(false); + V loaded = loadAndRecord(key, loadFunction); + if (loaded != null) { + AtomicBoolean commitInvoked = new AtomicBoolean(false); + try { + coordinator.accept(loaded, beforePublication -> { + if (!commitInvoked.compareAndSet(false, true)) { + throw new IllegalStateException("Metadata cache publication callback was invoked twice"); + } + Objects.requireNonNull(beforePublication, "beforePublication can not be null").run(); + }); + if (!commitInvoked.get()) { + throw new IllegalStateException("Metadata cache publication callback was not invoked"); + } + } finally { + notifyUnpublishedRemoval(key, loaded); + } + } + return loaded; + } + + VersionedValue<K, V> presentVersioned = currentVersionedValue(key, path); + if (presentVersioned != null) { + recordAccess(true); + scheduleRefresh(key, path, loader, presentVersioned); + return presentVersioned.value; + } + recordAccess(false); + try (PublicationLease<K, V> lease = acquirePublicationLease(key, path, true)) { + LoadAddress<K> loadAddress = new LoadAddress<>(key, path, lease); + CompletableFuture<V> ownLoad = new CompletableFuture<>(); + CompletableFuture<V> existingLoad = inFlightLoads.putIfAbsent(loadAddress, ownLoad); + if (existingLoad != null) { + return awaitLoad(existingLoad); + } + try { + afterLoadElection.run(); + AtomicReference<VersionedValue<K, V>> electedValue = new AtomicReference<>(); + boolean electedValuePresent = deferRemovals(() -> { + VersionedValue<K, V> present; + synchronized (lease.keyNode) { + present = currentVersionedValue(key, path); + } + electedValue.set(present); + return present != null; + }); + if (electedValuePresent) { + VersionedValue<K, V> present = electedValue.get(); + ownLoad.complete(present.value); + return present.value; + } + V loaded = loadAndRecord(key, loadFunction); + if (loaded != null) { + AtomicBoolean commitInvoked = new AtomicBoolean(false); + AtomicBoolean retained = new AtomicBoolean(false); + try { + coordinator.accept(loaded, beforePublication -> { + if (!commitInvoked.compareAndSet(false, true)) { + throw new IllegalStateException( + "Metadata cache publication callback was invoked twice"); + } + retained.set(commitLoaded(lease, key, loaded, beforePublication)); + }); + if (!commitInvoked.get()) { + throw new IllegalStateException("Metadata cache publication callback was not invoked"); + } + } finally { + if (!retained.get()) { + notifyUnpublishedRemoval(key, loaded); + } + } + } + ownLoad.complete(loaded); + return loaded; + } catch (RuntimeException | Error throwable) { + ownLoad.completeExceptionally(throwable); + throw throwable; + } finally { + inFlightLoads.remove(loadAddress, ownLoad); + } + } + } + + public V getIfPresent(K key, ScopePath path) { + Objects.requireNonNull(key, "key can not be null"); + Objects.requireNonNull(path, "path can not be null"); + checkOpen(); + if (!effectiveEnabled) { + recordAccess(false); + return null; + } + VersionedValue<K, V> versioned = currentVersionedValue(key, path); + recordAccess(versioned != null); + return versioned == null ? null : versioned.value; + } + + private VersionedValue<K, V> currentVersionedValue(K key, ScopePath path) { + VersionedValue<K, V> versioned = data.getIfPresent(key); + if (versioned == null) { + return null; + } + if (!versioned.scopeSnapshot.path().equals(path)) { + return null; + } + if (!versioned.isCurrent(registry, keyNodes)) { + data.asMap().remove(key, versioned); + return null; + } + return versioned; + } + + public void put(K key, ScopePath path, V value) { + Objects.requireNonNull(key, "key can not be null"); + Objects.requireNonNull(path, "path can not be null"); + Objects.requireNonNull(value, "value can not be null"); + checkOpen(); + if (!effectiveEnabled) { + return; + } + try (PublicationLease<K, V> lease = acquirePublicationLease(key, path, false)) { + guardedCommit(lease, () -> { + lease.keyNode.loadPublicationState.set(new Object()); + return publishCommitted(lease, key, value) != null; + }); + } + } + + public boolean compareAndSet(K key, ScopePath path, V expectedValue, V updatedValue) { + Objects.requireNonNull(key, "key can not be null"); + Objects.requireNonNull(path, "path can not be null"); + checkOpen(); + if (!effectiveEnabled) { + return true; + } + try (PublicationLease<K, V> lease = acquirePublicationLease(key, path, false)) { + return guardedCommit(lease, () -> { + VersionedValue<K, V> current = currentVersionedValue(key, path); + V currentValue = current == null ? null : current.value; + if (currentValue != expectedValue) { + return false; + } + lease.keyNode.loadPublicationState.set(new Object()); + if (updatedValue == currentValue) { + return true; + } + if (updatedValue == null) { + if (current != null) { + data.asMap().remove(key, current); + lease.keyNode.registration.compareAndSet(current, null); + } + } else { + publishCommitted(lease, key, updatedValue); + } + return true; + }); + } + } + + public void invalidateKey(K key) { + invalidateKey(key, NO_OP, NO_OP); + } + + void invalidateKey(K key, Runnable afterStateReplacement) { + invalidateKey(key, NO_OP, afterStateReplacement); + } + + void invalidateKey( + K key, Runnable beforeInvalidationLock, Runnable afterStateReplacement) { + Objects.requireNonNull(key, "key can not be null"); + Objects.requireNonNull(beforeInvalidationLock, "beforeInvalidationLock can not be null"); + Objects.requireNonNull(afterStateReplacement, "afterStateReplacement can not be null"); + checkOpen(); + beforeInvalidationLock.run(); + InvalidatedKey<K, V> invalidated = bulkInvalidationGate.write(() -> { + if (closed.get()) { + return null; + } + exactInvalidationSequence = exactInvalidationSequence.add(BigInteger.ONE); + if (!activeBulkStarts.isEmpty()) { + exactInvalidations.put(key, exactInvalidationSequence); + } + KeyNode<K, V> node = keyNodes.get(key); + KeyState invalidatedState = null; + if (node != null) { + invalidatedState = replaceKeyState(node); + } + return new InvalidatedKey<>(node, invalidatedState); + }); + if (invalidated == null || invalidated.node == null) { + return; + } + afterStateReplacement.run(); + VersionedValue<K, V> registered = invalidated.node.registration.get(); + if (registered != null && registered.keyState == invalidated.keyState) { + data.asMap().remove(key, registered); + invalidated.node.registration.compareAndSet(registered, null); + } + tryPruneKey(key, invalidated.node); + } + + public BulkLoadHandle beginBulkLoad(ScopePath parentScope) { + Objects.requireNonNull(parentScope, "parentScope can not be null"); + checkOpen(); + if (!effectiveEnabled) { + return BulkLoadHandle.disabled(this, parentScope); + } + ScopeLease scopeLease = registry.acquire(parentScope); + BigInteger exactSequence = bulkInvalidationGate.write(() -> { + if (closed.get()) { + scopeLease.close(); + throw new IllegalStateException("Scoped meta cache '" + name + "' is closed"); + } + BigInteger sequence = exactInvalidationSequence; + activeBulkStarts.merge(sequence, 1, Integer::sum); + return sequence; + }); + return new BulkLoadHandle( + this, + parentScope, + scopeLease, + scopeLease.publicationState(), + exactSequence); + } + + public boolean publish( + BulkLoadHandle handle, K key, ScopePath actualScope, V value) { + Objects.requireNonNull(handle, "handle can not be null"); + Objects.requireNonNull(key, "key can not be null"); + Objects.requireNonNull(actualScope, "actualScope can not be null"); + Objects.requireNonNull(value, "value can not be null"); + checkOpen(); + handle.checkOwner(this); + if (!handle.parentScope.contains(actualScope)) { + throw new IllegalArgumentException( + "Actual scope " + actualScope + " is outside bulk-load parent " + handle.parentScope); + } + if (!effectiveEnabled) { + return false; + } + try (PublicationLease<K, V> lease = acquirePublicationLease(key, actualScope, false)) { + VersionedValue<K, V> staged = newVersionedValue(lease, key, value); + afterBulkStage.run(); + return handle.tryCommit(key, lease, staged); + } + } + + public CacheMetrics metrics() { + return bulkInvalidationGate.read(() -> new CacheMetrics( + data.estimatedSize(), + keyNodes.size(), + inFlightLoads.size(), + activeBulkStarts.values().stream().mapToInt(Integer::intValue).sum(), + exactInvalidations.size(), + effectiveEnabled, + requestCount.sum(), + hitCount.sum(), + missCount.sum(), + loadSuccessCount.sum(), + loadFailureCount.sum(), + totalLoadTimeNanos.sum(), + evictionCount.sum(), + invalidateCount.sum(), + lastLoadSuccessTimeMs.get(), + lastLoadFailureTimeMs.get(), + lastError.get())); + } + + int refreshingCountForTest() { + return refreshing.size(); + } + + public void forEach(BiConsumer<K, V> consumer) { + Objects.requireNonNull(consumer, "consumer can not be null"); + data.asMap().forEach((key, versioned) -> { + if (versioned.isCurrent(registry, keyNodes)) { + consumer.accept(key, versioned.value); + } + }); + } + + private V loadAndRecord(K key, Function<K, V> loader) { + long startNanos = System.nanoTime(); + try { + V loaded = loader.apply(key); + loadSuccessCount.increment(); + lastLoadSuccessTimeMs.set(System.currentTimeMillis()); + return loaded; + } catch (RuntimeException | Error throwable) { + loadFailureCount.increment(); + lastLoadFailureTimeMs.set(System.currentTimeMillis()); + lastError.set(throwable.toString()); + throw throwable; + } finally { + totalLoadTimeNanos.add(System.nanoTime() - startNanos); + } + } + + private void recordAccess(boolean hit) { + requestCount.increment(); + if (hit) { + hitCount.increment(); + } else { + missCount.increment(); + } + } + + public void cleanUp() { + data.cleanUp(); + } + + @Override + public void close() { + if (!closed.compareAndSet(false, true)) { + return; + } + registry.removeCache(this); + closePhysicalState(); + } + + void closeFromRegistry() { + if (closed.compareAndSet(false, true)) { + closePhysicalState(); + } + } + + void removeExpectedRaw(Object rawKey, Object expectedValue) { + @SuppressWarnings("unchecked") + K key = (K) rawKey; + @SuppressWarnings("unchecked") + VersionedValue<K, V> versionedValue = (VersionedValue<K, V>) expectedValue; + data.asMap().remove(key, versionedValue); + } + + private PublicationLease<K, V> acquirePublicationLease( + K key, ScopePath path, boolean fenceAgainstDirectPublication) { + while (true) { + ScopeLease scopeLease = registry.acquire(path); + KeyNode<K, V> keyNode = keyNodes.computeIfAbsent(key, ignored -> new KeyNode<>()); + keyNode.activeLoads.incrementAndGet(); + KeyState keyState = keyNode.current.get(); + Object loadPublicationState = + fenceAgainstDirectPublication ? keyNode.loadPublicationState.get() : null; + if (keyNodes.get(key) == keyNode && scopeLease.isCurrent()) { + return new PublicationLease<>( + this, key, scopeLease, keyNode, keyState, loadPublicationState); + } + releaseKey(key, keyNode); + scopeLease.close(); + } + } + + private VersionedValue<K, V> publish(PublicationLease<K, V> lease, K key, V value) { + if (!lease.isCurrent()) { + return null; + } + VersionedValue<K, V> versioned = newVersionedValue(lease, key, value); + install(versioned, lease); + if (!lease.isCurrent()) { + data.asMap().remove(key, versioned); + return null; + } + return versioned; + } + + private VersionedValue<K, V> publishCommitted(PublicationLease<K, V> lease, K key, V value) { + return publish(lease, key, value); + } + + private boolean commitLoaded(PublicationLease<K, V> lease, K key, V value, Runnable beforePublication) { + Runnable action = Objects.requireNonNull(beforePublication, "beforePublication can not be null"); + return guardedCommit(lease, () -> { + action.run(); + return publishCommitted(lease, key, value) != null; + }); + } + + private boolean guardedCommit(PublicationLease<K, V> lease, BooleanSupplier commitAction) { + return deferRemovals(() -> bulkInvalidationGate.readBoolean( + () -> lease.scopeLease.commitIfPublicationCurrent( + lease.scopePublicationState, () -> { + synchronized (lease.keyNode) { + return lease.isCurrent() && commitAction.getAsBoolean(); + } + }))); + } + + private VersionedValue<K, V> newVersionedValue( + PublicationLease<K, V> lease, K key, V value) { + CacheAddress address = new CacheAddress(this, key); + return new VersionedValue<>( + key, value, address, lease.scopeLease.snapshot(), lease.keyNode, lease.keyState, ticker.read()); + } + + private void scheduleRefresh(K key, ScopePath path, Function<K, V> loader, VersionedValue<K, V> current) { + if (refreshAfterWriteNanos == 0L || ticker.read() - current.writeTimeNanos < refreshAfterWriteNanos + || refreshing.putIfAbsent(key, current) != null) { + return; + } + PublicationLease<K, V> lease; + try { + afterRefreshRegistration.run(); + lease = acquirePublicationLease(key, path, true); + } catch (RuntimeException | Error throwable) { + refreshing.remove(key, current); + throw throwable; + } + if (data.getIfPresent(key) != current || !current.isCurrent(registry, keyNodes)) { + lease.close(); + refreshing.remove(key, current); + return; + } + try { + refreshExecutor.execute(() -> { + try (PublicationLease<K, V> ignored = lease) { + if (closed.get()) { + return; + } + V refreshed = loadAndRecord(key, loader); + if (refreshed != null) { + boolean retained = false; + try { + retained = replaceRefreshExpected(lease, key, current, refreshed); + } finally { + if (!retained) { + notifyUnpublishedRemoval(key, refreshed); + } + } + } + } catch (RuntimeException | Error throwable) { + LOG.warn("Scoped metadata cache refresh failed", throwable); + } finally { + refreshing.remove(key, current); + } + }); + } catch (RejectedExecutionException exception) { + lease.close(); + refreshing.remove(key, current); + } + } + + private void install( + VersionedValue<K, V> versioned, PublicationLease<K, V> lease) { + registry.register(versioned.address, versioned, versioned.scopeSnapshot); + lease.keyNode.registration.set(versioned); + data.asMap().put(versioned.key, versioned); + } + + private boolean replaceRefreshExpected(PublicationLease<K, V> lease, K key, + VersionedValue<K, V> expected, V refreshed) { + return guardedCommit(lease, () -> { + if (data.getIfPresent(key) != expected || !expected.isCurrent(registry, keyNodes)) { + return false; + } + VersionedValue<K, V> replacement = newVersionedValue(lease, key, refreshed); Review Comment: Fixed in 5c0c8571519. When refresh returns the same value identity, the existing wrapper remains published and only its write timestamp is renewed. No removal/discard callback runs for that identity; rejected identity refresh also avoids double retirement. Both paths now have regression tests. ########## fe/fe-core/src/main/java/org/apache/doris/datasource/ExternalTable.java: ########## @@ -243,10 +243,6 @@ public String getEngine() { /** * Returns the effective meta cache engine for this table. */ - public String getMetaCacheEngine() { Review Comment: 已在 5c0c8571519 删除这段过期注释。 ########## fe/fe-connector/fe-connector-hive/src/main/java/org/apache/doris/connector/hive/HiveConnector.java: ########## @@ -135,10 +138,12 @@ public HiveConnector(Map<String, String> properties, ConnectorContext context) { this.props = HiveCatalogProperties.of(properties); this.properties = props.getRaw(); this.context = context; - this.fileListingCache = new HiveFileListingCache(props); + this.fileListingCache = new HiveFileListingCache(metaCache, props); Review Comment: 这里 connector 显式声明的是 connector-specific 的 key、scope、loader 以及资源释放语义；缓存容器、层级注册、统一失效、并发 publication 和 lifecycle 已由 CatalogMetaCache/ScopedMetaCache 在 core runtime 统一管理。保留 definition 在 connector 侧可以避免 core 反向依赖插件类型，也能覆盖 Iceberg 这类带独立资源引用的对象。你提到的统一内存预算/治理确实还没有在本 PR 解决；这部分按计划放到 #66717 的 master 版本中接入统一 runtime，本 PR 不再扩这个 scope。 -- 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]
