924060929 commented on code in PR #66633: URL: https://github.com/apache/doris/pull/66633#discussion_r3869505857
########## 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")); + } + + 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) { + builder.removalListener((key, value, reason) -> + removalListener.onRemoval(key, value, toCaffeineRemovalCause(reason))); + } + if (autoRefresh && Config.external_cache_refresh_time_minutes > 0) { + builder.refreshAfterWrite( + Duration.ofMinutes(Config.external_cache_refresh_time_minutes), refreshExecutor); + } + data = owner.create(builder.build()); + } + + public String name() { + return name; + } + + public V get(K key) { + if (loader == null) { + throw new UnsupportedOperationException(String.format( + "Entry '%s' requires a contextual miss loader.", name)); + } + return data.get(key); + } + + public V get(K key, Function<K, V> missLoader) { + Function<K, V> nonNullLoader = Objects.requireNonNull(missLoader, "missLoader can not be null"); + return data.get(key, loadKey -> loadAndPause(loadKey, nonNullLoader)); + } + + public V getAndRunIfCurrent(K key, BiConsumer<K, V> currentValueAction) { + return getAndRunIfCurrent(key, (ignored, value) -> true, currentValueAction); + } + + public V getAndRunIfCurrent(K key, BiPredicate<K, V> actionRequired, + BiConsumer<K, V> currentValueAction) { + BiPredicate<K, V> required = Objects.requireNonNull(actionRequired, "actionRequired can not be null"); + BiConsumer<K, V> action = Objects.requireNonNull(currentValueAction, "currentValueAction can not be null"); + Function<K, V> loadFunction = Objects.requireNonNull(loader, "loader can not be null"); + V cached = data.getIfPresent(key); + if (cached != null && !required.test(key, cached)) { + return cached; + } + StripeState<K> stripe = stripeState(key); + ActionToken<K> token; + synchronized (stripe) { + token = beginAction(stripe, key); + } + try { + AtomicBoolean actionPublished = new AtomicBoolean(false); + V value = cached == null + ? data.getWithPublicationAction(key, loadKey -> loadAndPause(loadKey, loadFunction), + (loaded, commit) -> { + beforeCurrentValueActionForTest(key, loaded); + synchronized (stripe) { + if (isCurrent(stripe, token) && required.test(key, loaded)) { + commit.accept(() -> action.accept(key, loaded)); + actionPublished.set(true); + } else { + commit.accept(() -> { + }); + } + } + }) + : cached; + if (value == null) { + return null; + } + if (actionPublished.get()) { + return value; + } + if (!required.test(key, value)) { + return value; + } + beforeCurrentValueActionForTest(key, value); + synchronized (stripe) { + boolean current = isCurrent(stripe, token) + && (!effectiveEnabled || data.getIfPresent(key) == value); + if (current && required.test(key, value) && isCurrent(stripe, token)) { + try { + action.accept(key, value); + } catch (RuntimeException | Error throwable) { + data.invalidateKey(key); + throw throwable; + } + } + } + return value; + } finally { + synchronized (stripe) { + endAction(stripe, token); + } + } + } + + public V getIfPresent(K key) { + return data.getIfPresent(key); + } + + @Nullable + public V findIfPresent(Predicate<K> keyPredicate) { + Objects.requireNonNull(keyPredicate, "keyPredicate can not be null"); + List<V> result = new ArrayList<>(1); + data.forEach((key, value) -> { + if (result.isEmpty() && keyPredicate.test(key)) { + result.add(value); + } + }); + return result.isEmpty() ? null : result.get(0); + } + + public void put(K key, V value) { + Objects.requireNonNull(value, "value can not be null"); + StripeState<K> stripe = stripeState(key); + synchronized (stripe) { + bumpAction(stripe, key); + beforePublicMutationWriteForTest(key); + data.put(key, value); + } + } + + public V compute(K key, BiFunction<K, V, V> remappingFunction) { + return computeAndRun(key, remappingFunction, () -> { + }); + } + + public V computeAndRun(K key, BiFunction<K, V, V> remappingFunction, Runnable afterMutation) { + BiFunction<K, V, V> remapper = Objects.requireNonNull(remappingFunction, "remappingFunction can not be null"); + Runnable action = Objects.requireNonNull(afterMutation, "afterMutation can not be null"); + StripeState<K> stripe = stripeState(key); + synchronized (stripe) { + bumpAction(stripe, key); + while (true) { + V current = data.getIfPresent(key); + V updated = effectiveEnabled ? remapper.apply(key, current) : null; + beforePublicMutationWriteForTest(key); + if (data.compareAndSet(key, current, updated)) { + action.run(); + return updated; + } + } + } + } + + public V computeAfterValidation(K key, BiFunction<K, V, V> remappingFunction, Runnable validationAction) { + BiFunction<K, V, V> remapper = Objects.requireNonNull(remappingFunction, "remappingFunction can not be null"); + Runnable validation = Objects.requireNonNull(validationAction, "validationAction can not be null"); + StripeState<K> stripe = stripeState(key); + synchronized (stripe) { + while (true) { + V current = data.getIfPresent(key); + V updated = effectiveEnabled ? remapper.apply(key, current) : null; + validation.run(); Review Comment: 已修复，提交是 529ed2d3801。 之前注册数据库时，ID 映射会先对查询可见，但对应的数据库对象还没有提交完成。并发的按 ID 查询可能趁这个空档放入另一份对象；事件更新随后替换它时，又会把查询已经拿到的对象重置掉。 现在 ID 映射和数据库对象放在公共 cache 的同一次提交保护里：只有确认旧对象没有变化后才更新 ID，并且在 ID 更新和对象更新全部完成前，不允许并发加载插入另一份对象。表缓存也用了同样的处理。 我补了可重复的 latch 并发测试，专门覆盖“ID 已可见、对象还没更新”这个时序。验证通过：公共 cache 111 个测试、FeMetaCacheEntryTest 14 个测试、ExternalCatalogTest 和 ExternalDatabaseTest 共 81 个测试，以及完整 ./build.sh --fe。 -- 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]
