924060929 commented on code in PR #66633: URL: https://github.com/apache/doris/pull/66633#discussion_r3841340075
########## fe/fe-connector/fe-connector-cache/src/main/java/org/apache/doris/connector/cache/ScopedMetaCache.java: ########## @@ -0,0 +1,1142 @@ +// 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); + 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"); + } + } + 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(); + synchronized (lease.keyNode) { + VersionedValue<K, V> present = currentVersionedValue(key, path); + if (present != null) { + ownLoad.complete(present.value); + return present.value; + } + } + V loaded = loadAndRecord(key, loadFunction); + if (loaded != null) { + AtomicBoolean commitInvoked = new AtomicBoolean(false); + coordinator.accept(loaded, beforePublication -> { + if (!commitInvoked.compareAndSet(false, true)) { + throw new IllegalStateException("Metadata cache publication callback was invoked twice"); + } + commitLoaded(lease, key, loaded, beforePublication); + }); + if (!commitInvoked.get()) { + throw new IllegalStateException("Metadata cache publication callback was not invoked"); + } + } + 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)) { + synchronized (lease.keyNode) { + lease.keyNode.loadPublicationState.set(new Object()); + publishCommitted(lease, key, value); + } + } + } + + 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)) { + synchronized (lease.keyNode) { Review Comment: Fixed in eb5efc9cd11 by moving per-key synchronization into tryCommitBulk, after the bulk and scope publication read gates. Bulk staging no longer holds the key monitor. Added bulkStagingDoesNotHoldPublicationKey; the 40-test concurrency suite and full FE build pass. ########## fe/fe-connector/fe-connector-cache/src/main/java/org/apache/doris/connector/cache/ScopedMetaCache.java: ########## @@ -0,0 +1,1142 @@ +// 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); + 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"); + } + } + 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(); + synchronized (lease.keyNode) { + VersionedValue<K, V> present = currentVersionedValue(key, path); + if (present != null) { + ownLoad.complete(present.value); + return present.value; + } + } + V loaded = loadAndRecord(key, loadFunction); + if (loaded != null) { + AtomicBoolean commitInvoked = new AtomicBoolean(false); + coordinator.accept(loaded, beforePublication -> { + if (!commitInvoked.compareAndSet(false, true)) { + throw new IllegalStateException("Metadata cache publication callback was invoked twice"); + } + commitLoaded(lease, key, loaded, beforePublication); + }); + if (!commitInvoked.get()) { + throw new IllegalStateException("Metadata cache publication callback was not invoked"); + } + } + 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)) { + synchronized (lease.keyNode) { Review Comment: Fixed in eb5efc9cd11 by routing direct put through guardedCommit and deferring synchronous Caffeine removal callbacks until all publication gates and the key monitor are released. The elected-loader secondary check now uses the same deferral. Added directPutReplacementRemovalRunsAfterPublicationLocks and electedLoaderStaleRemovalRunsAfterPublicationKey. ########## fe/fe-connector/fe-connector-cache/src/main/java/org/apache/doris/connector/cache/ScopedMetaCacheRegistry.java: ########## @@ -0,0 +1,953 @@ +// 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 com.github.benmanes.caffeine.cache.RemovalListener; +import com.github.benmanes.caffeine.cache.Ticker; + +import java.time.Duration; +import java.util.ArrayList; +import java.util.Collections; +import java.util.IdentityHashMap; +import java.util.List; +import java.util.Objects; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; +import java.util.concurrent.Executor; +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; + +/** + * Catalog-local hierarchy shared by independent physical metadata caches. + * + * <p>Logical invalidation replaces one state object. Values capture state identities and are rejected after a + * replacement even when physical Caffeine cleanup is delayed. Each state also owns the exact registrations that + * must be physically removed. Empty child nodes are pruned bottom-up so scope-directory memory remains bounded by + * live values and in-flight loads rather than by every name ever observed. + */ +public final class ScopedMetaCacheRegistry implements AutoCloseable { + private static final Runnable NO_OP = () -> { + }; + private static final BiConsumer<ScopePath.Level, Object> NO_OP_SCOPE = (level, key) -> { + }; + + private final ScopeNode root = new ScopeNode(null, null, ScopePath.Level.CATALOG); + private final Set<ScopedMetaCache<?, ?>> caches = ConcurrentHashMap.newKeySet(); + private final StripedPhaseGate publicationGate = new StripedPhaseGate(); + private final AtomicBoolean closed = new AtomicBoolean(false); + private final LongAdder[] retainedActiveLoads = newCounters(ScopePath.Level.values().length); + private final BiConsumer<ScopePath.Level, Object> afterParentPin; + private final BiConsumer<ScopePath.Level, Object> beforePruneMark; + private final BiConsumer<ScopePath.Level, Object> afterPruneMark; + + public ScopedMetaCacheRegistry() { + this(NO_OP_SCOPE, NO_OP_SCOPE, NO_OP_SCOPE); + } + + ScopedMetaCacheRegistry( + BiConsumer<ScopePath.Level, Object> afterParentPin, + BiConsumer<ScopePath.Level, Object> beforePruneMark, + BiConsumer<ScopePath.Level, Object> afterPruneMark) { + this.afterParentPin = Objects.requireNonNull(afterParentPin, "afterParentPin can not be null"); + this.beforePruneMark = Objects.requireNonNull(beforePruneMark, "beforePruneMark can not be null"); + this.afterPruneMark = Objects.requireNonNull(afterPruneMark, "afterPruneMark can not be null"); + } + + public <K, V> ScopedMetaCache<K, V> createCache(String name, CacheSpec cacheSpec) { + return createCache(name, cacheSpec, null, null, NO_OP, NO_OP); + } + + <K, V> ScopedMetaCache<K, V> createCacheWithRemovalListener( + String name, CacheSpec cacheSpec, RemovalListener<K, V> removalListener, + Duration refreshAfterWrite, Executor refreshExecutor) { + return createCacheWithRemovalListener(name, cacheSpec, null, removalListener, + refreshAfterWrite, refreshExecutor, NO_OP, NO_OP, NO_OP); + } + + <K, V> ScopedMetaCache<K, V> createCacheWithMetaRemovalListener( + String name, CacheSpec cacheSpec, MetaCacheRemovalListener<K, V> removalListener, + Duration refreshAfterWrite, Executor refreshExecutor) { + RemovalListener<K, V> caffeineListener = removalListener == null ? null + : (key, value, cause) -> removalListener.onRemoval( + key, value, MetaCacheRemovalReason.valueOf(cause.name())); + return createCacheWithRemovalListener( + name, cacheSpec, caffeineListener, refreshAfterWrite, refreshExecutor); + } + + <K, V> ScopedMetaCache<K, V> createCache( + String name, + CacheSpec cacheSpec, + Ticker ticker, + BiConsumer<K, V> beforeRemoval) { + return createCache(name, cacheSpec, ticker, beforeRemoval, NO_OP, NO_OP); + } + + <K, V> ScopedMetaCache<K, V> createCache( + String name, + CacheSpec cacheSpec, + Ticker ticker, + BiConsumer<K, V> beforeRemoval, + Runnable afterBulkStage) { + return createCache(name, cacheSpec, ticker, beforeRemoval, NO_OP, afterBulkStage); + } + + <K, V> ScopedMetaCache<K, V> createCache( + String name, + CacheSpec cacheSpec, + Ticker ticker, + BiConsumer<K, V> beforeRemoval, + Runnable afterLoadElection, + Runnable afterBulkStage) { + RemovalListener<K, V> listener = beforeRemoval == null + ? null + : (key, value, cause) -> beforeRemoval.accept(key, value); + return createCacheWithRemovalListener( + name, cacheSpec, ticker, listener, null, null, afterLoadElection, afterBulkStage, NO_OP); + } + + <K, V> ScopedMetaCache<K, V> createCacheWithRefresh( + String name, + CacheSpec cacheSpec, + Duration refreshAfterWrite, + Executor refreshExecutor, + Runnable afterRefreshRegistration) { + return createCacheWithRemovalListener( + name, cacheSpec, null, null, refreshAfterWrite, refreshExecutor, + NO_OP, NO_OP, afterRefreshRegistration); + } + + private <K, V> ScopedMetaCache<K, V> createCacheWithRemovalListener( + String name, + CacheSpec cacheSpec, + Ticker ticker, + RemovalListener<K, V> removalListener, + Duration refreshAfterWrite, + Executor refreshExecutor, + Runnable afterLoadElection, + Runnable afterBulkStage, + Runnable afterRefreshRegistration) { + checkOpen(); + ScopedMetaCache<K, V> cache = new ScopedMetaCache<>( + this, + name, + cacheSpec, + ticker, + removalListener, + refreshAfterWrite, + refreshExecutor, + afterLoadElection, + afterBulkStage, + afterRefreshRegistration); + caches.add(cache); + if (closed.get()) { + caches.remove(cache); + cache.closeFromRegistry(); + throw new IllegalStateException("Scoped meta cache registry is closed"); + } + return cache; + } + + public void invalidate(ScopePath path) { + invalidate(path, NO_OP); + } + + void invalidate(List<ScopePath> paths) { + invalidate(paths, NO_OP); + } + + void invalidate(ScopePath path, Runnable afterStateReplacement) { + Objects.requireNonNull(path, "path can not be null"); + Objects.requireNonNull(afterStateReplacement, "afterStateReplacement can not be null"); + checkOpen(); + InvalidatedScope invalidated = publicationGate.write(() -> { + List<ScopeNode> existing = resolveExisting(path); + bumpPublicationStates(existing); + if (existing.size() != path.level().ordinal() + 1) { + return null; + } + ScopeNode target = existing.get(existing.size() - 1); + return new InvalidatedScope(existing, replaceState(target)); + }); + if (invalidated == null) { + return; + } + afterStateReplacement.run(); + cleanDetachedState(invalidated.oldState); + tryPrune(invalidated.existing); + } + + void invalidate(List<ScopePath> paths, Runnable afterStateReplacement) { + Objects.requireNonNull(paths, "paths can not be null"); + Objects.requireNonNull(afterStateReplacement, "afterStateReplacement can not be null"); + checkOpen(); + List<InvalidatedScope> invalidated = publicationGate.write(() -> { + List<List<ScopeNode>> resolvedScopes = new ArrayList<>(paths.size()); + Set<ScopeNode> targets = Collections.newSetFromMap(new IdentityHashMap<>()); + Set<ScopeNode> publicationNodes = Collections.newSetFromMap(new IdentityHashMap<>()); + for (ScopePath path : paths) { + Objects.requireNonNull(path, "path can not be null"); + List<ScopeNode> existing = resolveExisting(path); + if (existing.size() == path.level().ordinal() + 1) { Review Comment: The rebased implementation includes every existing prefix in publicationNodes before resolving the full invalidation path, so batch invalidation advances cold descendants even when the leaf was not previously registered. The focused cold-descendant concurrency coverage passes on eb5efc9cd11. ########## fe/fe-connector/fe-connector-cache/src/main/java/org/apache/doris/connector/cache/ScopedMetaCache.java: ########## @@ -0,0 +1,1134 @@ +// 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); + 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"); + } + } + 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(); + synchronized (lease.keyNode) { + VersionedValue<K, V> present = currentVersionedValue(key, path); + if (present != null) { + ownLoad.complete(present.value); + return present.value; + } + } + V loaded = loadAndRecord(key, loadFunction); + if (loaded != null) { + AtomicBoolean commitInvoked = new AtomicBoolean(false); + coordinator.accept(loaded, beforePublication -> { + if (!commitInvoked.compareAndSet(false, true)) { + throw new IllegalStateException("Metadata cache publication callback was invoked twice"); + } + commitLoaded(lease, key, loaded, beforePublication); + }); + if (!commitInvoked.get()) { + throw new IllegalStateException("Metadata cache publication callback was not invoked"); + } + } + 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)) { + synchronized (lease.keyNode) { + lease.keyNode.loadPublicationState.set(new Object()); + publishCommitted(lease, key, value); + } + } + } + + 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 == null) { + if (current != null) { + data.asMap().remove(key, current); + lease.keyNode.registration.compareAndSet(current, null); + } + } else { + publishCommitted(lease, key, updatedValue); Review Comment: The rebased implementation handles identity-preserving updates explicitly: when updatedValue == currentValue it advances the load publication state and returns without replace/removal. The focused identity no-op concurrency coverage passes on eb5efc9cd11. -- 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]
