github-actions[bot] commented on code in PR #66717: URL: https://github.com/apache/doris/pull/66717#discussion_r3820183763
########## fe/fe-core/src/main/java/org/apache/doris/datasource/hive/HiveCacheSizeEstimator.java: ########## @@ -0,0 +1,62 @@ +// 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.hive; + +import org.apache.doris.datasource.hive.HiveExternalMetaCache.HivePartitionValues; +import org.apache.doris.datasource.hive.HiveExternalMetaCache.PartitionValueCacheKey; +import org.apache.doris.datasource.metacache.MetaCacheSizeEstimate; +import org.apache.doris.datasource.metacache.MetaCacheWeightUtils; + +/** Constant-time retained-weight formula for Hive partition-value cache entries. */ +final class HiveCacheSizeEstimator { + // Calibrated against complete 4.1 object graphs. The payload reserve covers the partition + // name plus derived value/literal strings and therefore remains skew-sensitive. + private static final long ENTRY_BASE_BYTES = objectBytes(2L * 1024L); + private static final long PARTITION_BASE_BYTES = objectBytes(896L); + private static final long PARTITION_COLUMN_BYTES = objectBytes(256L); + // One copy is retained as the partition name and another in the decoded partition values. + private static final long PARTITION_NAME_PAYLOAD_COPIES = 2L; + + private HiveCacheSizeEstimator() { + } + + private static long objectBytes(long bytes) { + return MetaCacheWeightUtils.estimatedObjectBytes(bytes); + } + + static MetaCacheSizeEstimate estimatePartitionValuesEntry( + PartitionValueCacheKey key, HivePartitionValues value) { + if (!MetaCacheWeightUtils.isSupportedJvmObjectLayout()) { + return MetaCacheSizeEstimate.incomplete("unsupported_jvm_object_alignment"); + } + long partitionCount = value.getIdToPartitionItem() == null + ? 0L : value.getIdToPartitionItem().size(); + long perPartitionBytes = MetaCacheWeightUtils.saturatedAdd( + PARTITION_BASE_BYTES, + MetaCacheWeightUtils.saturatedMultiply( + value.getPartitionColumnCount(), PARTITION_COLUMN_BYTES)); + long bytes = MetaCacheWeightUtils.saturatedAdd( + ENTRY_BASE_BYTES, MetaCacheWeightUtils.estimatedNameMappingBytes(key.getNameMapping())); Review Comment: [P2] Account for the retained key width on empty tables `PartitionValueCacheKey` retains an immutable copy of `types`, but this formula only charges partition-column width inside `partitionCount * perPartitionBytes`. For an empty partitioned table, `partitionCount` is zero, so a key with one type and one with many types receive the same estimate even though the immutable list and backing references retained by Caffeine grow with width. Charge the key list and slots independently of partition count, and add an empty-table narrow-versus-wide calibration rooted at both the key and value. ########## fe/fe-core/src/main/java/org/apache/doris/datasource/metacache/ExternalMetaCacheBudgetManager.java: ########## @@ -0,0 +1,587 @@ +// 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.logging.log4j.LogManager; +import org.apache.logging.log4j.Logger; + +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.Optional; +import java.util.OptionalLong; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.RejectedExecutionException; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicLong; +import java.util.function.LongUnaryOperator; +import java.util.stream.Collectors; + +/** + * FE-wide admission accounting for managed external metadata caches. + * + * <p>All changes are serialized by one short critical section. Cache loads and + * estimators run outside it, so the lock only protects a few arithmetic and map + * operations while making global/catalog/entry reservation atomic. + */ +public final class ExternalMetaCacheBudgetManager { + private static final Logger LOG = LogManager.getLogger(ExternalMetaCacheBudgetManager.class); + private static final ExecutorService PEER_RECLAIM_EXECUTOR = Executors.newSingleThreadExecutor(runnable -> { + Thread thread = new Thread(runnable, "external-meta-cache-peer-reclaim"); + thread.setDaemon(true); + return thread; + }); + + public static final String CATALOG_MAX_WEIGHT_PROPERTY = "meta.cache.max-weight"; + + private final Object lock = new Object(); + private final OptionalLong globalMaxWeight; + private final Map<Long, Bucket> catalogBuckets = new HashMap<>(); + private final Map<EntryScope, Bucket> entryBuckets = new HashMap<>(); + private final Map<EntryScope, EntryBudget> entryBudgets = new HashMap<>(); + private long globalUsedWeight; + private final AtomicLong globalRejectedCount = new AtomicLong(); + + public ExternalMetaCacheBudgetManager(OptionalLong globalMaxWeight) { + this.globalMaxWeight = Objects.requireNonNull(globalMaxWeight, "globalMaxWeight"); + if (globalMaxWeight.isPresent() && globalMaxWeight.getAsLong() <= 0) { + throw new IllegalArgumentException("global max weight must be positive when enabled"); + } + } + + public static ExternalMetaCacheBudgetManager fromConfig() { + String configured = Config.external_meta_cache_max_weight; + long parsed = CacheSpec.parseWeight( + configured, + "external_meta_cache_max_weight", + true, + Runtime.getRuntime().maxMemory()); + if (configured.trim().endsWith("%") && parsed == 0L) { + throw new IllegalArgumentException( + "external_meta_cache_max_weight percentage must be greater than 0%"); + } + return new ExternalMetaCacheBudgetManager(parsed == 0L ? OptionalLong.empty() : OptionalLong.of(parsed)); + } + + public OptionalLong parseCatalogMaxWeight(Map<String, String> catalogProperties) { + String configured = catalogProperties.get(CATALOG_MAX_WEIGHT_PROPERTY); + if (configured == null) { + return OptionalLong.empty(); + } + long parsed = CacheSpec.parseWeight(configured, CATALOG_MAX_WEIGHT_PROPERTY, false, 0L); + if (parsed <= 0) { + throw new IllegalArgumentException(CATALOG_MAX_WEIGHT_PROPERTY + " must be positive"); + } + return OptionalLong.of(parsed); + } + + /** Validate a catalog limit at DDL time against this FE's configured global bound. */ + public OptionalLong validateCatalogMaxWeight(Map<String, String> catalogProperties) { + OptionalLong catalogMaxWeight = parseCatalogMaxWeight(catalogProperties); + validateHierarchy(catalogMaxWeight, OptionalLong.empty()); + return catalogMaxWeight; + } + + /** + * Create the budget handle used by one physical per-catalog cache entry. + */ + public EntryBudget createEntryBudget(long catalogId, String engine, String entryName, + OptionalLong catalogMaxWeight, OptionalLong entryMaxWeight) { + Objects.requireNonNull(engine, "engine"); + Objects.requireNonNull(entryName, "entryName"); + Objects.requireNonNull(catalogMaxWeight, "catalogMaxWeight"); + Objects.requireNonNull(entryMaxWeight, "entryMaxWeight"); + validateCatalogEntryHierarchy(catalogMaxWeight, entryMaxWeight); + + OptionalLong effectiveMax = minimumPresent(globalMaxWeight, catalogMaxWeight, entryMaxWeight); + if (!effectiveMax.isPresent()) { + throw new IllegalArgumentException("entry budget requires at least one configured weight bound"); + } + + EntryScope scope = new EntryScope(catalogId, engine, entryName); + synchronized (lock) { + Bucket catalogBucket = catalogBuckets.get(catalogId); + long catalogLimit = minimumLimit(globalMaxWeight, catalogMaxWeight); + if (catalogBucket == null) { + catalogBucket = new Bucket(catalogLimit); + catalogBuckets.put(catalogId, catalogBucket); + } else if (catalogBucket.maxWeight != catalogLimit) { + throw new IllegalStateException("Conflicting catalog cache max weight for catalog " + catalogId); + } + + if (entryBuckets.containsKey(scope)) { + throw new IllegalStateException("Duplicated external meta cache budget: " + scope); + } + Bucket entryBucket = new Bucket(effectiveMax.getAsLong()); + EntryBudget entryBudget = new EntryBudget( + this, scope, catalogBucket, entryBucket, effectiveMax.getAsLong()); + entryBuckets.put(scope, entryBucket); + entryBudgets.put(scope, entryBudget); + return entryBudget; + } + } + + public OptionalLong getGlobalMaxWeight() { + return globalMaxWeight; + } + + public long getGlobalUsedWeight() { + synchronized (lock) { + return globalUsedWeight; + } + } + + public long getGlobalRejectedCount() { + return globalRejectedCount.get(); + } + + public void validateHierarchy(OptionalLong catalogMaxWeight, OptionalLong entryMaxWeight) { + if (globalMaxWeight.isPresent() && catalogMaxWeight.isPresent() + && catalogMaxWeight.getAsLong() > globalMaxWeight.getAsLong()) { + throw new IllegalArgumentException(CATALOG_MAX_WEIGHT_PROPERTY + " can not exceed FE global max weight"); + } + OptionalLong parent = catalogMaxWeight.isPresent() ? catalogMaxWeight : globalMaxWeight; + if (parent.isPresent() && entryMaxWeight.isPresent() + && entryMaxWeight.getAsLong() > parent.getAsLong()) { + throw new IllegalArgumentException("entry max weight can not exceed its parent max weight"); + } + } + + /** + * Validate persisted catalog-to-entry hierarchy without comparing it with this FE's local + * global bound. Catalog properties are validated on the master, while the global percentage + * is resolved independently from each FE's heap. Runtime admission therefore clamps to the + * local global limit instead of rejecting a catalog accepted on a larger master. + */ + public void validateCatalogEntryHierarchy(OptionalLong catalogMaxWeight, OptionalLong entryMaxWeight) { + OptionalLong parent = catalogMaxWeight; + if (parent.isPresent() && entryMaxWeight.isPresent() + && entryMaxWeight.getAsLong() > parent.getAsLong()) { + throw new IllegalArgumentException("entry max weight can not exceed its parent max weight"); + } + } + + private Optional<AdmissionReservation> tryReserve(EntryBudget entryBudget, long bytes) { + checkWeight(bytes); + synchronized (lock) { + if (entryBudget.closed) { + return Optional.empty(); + } + if (!fits(limitOf(globalMaxWeight), globalUsedWeight, bytes) + || !fits(entryBudget.catalogBucket.maxWeight, entryBudget.catalogBucket.usedWeight, bytes) + || !fits(entryBudget.entryBucket.maxWeight, entryBudget.entryBucket.usedWeight, bytes)) { + entryBudget.rejectedCount.incrementAndGet(); + globalRejectedCount.incrementAndGet(); + return Optional.empty(); + } + addUsed(entryBudget, bytes); + return Optional.of(new AdmissionReservation(this, entryBudget, bytes)); + } + } + + private boolean resize(AdmissionReservation reservation, long newBytes) { + checkWeight(newBytes); + synchronized (lock) { + if (!reservation.active || reservation.entryBudget.closed) { + return false; + } + long delta = newBytes - reservation.bytes; + if (delta > 0 && (!fits(limitOf(globalMaxWeight), globalUsedWeight, delta) + || !fits(reservation.entryBudget.catalogBucket.maxWeight, + reservation.entryBudget.catalogBucket.usedWeight, delta) + || !fits(reservation.entryBudget.entryBucket.maxWeight, + reservation.entryBudget.entryBucket.usedWeight, delta))) { + reservation.entryBudget.rejectedCount.incrementAndGet(); + globalRejectedCount.incrementAndGet(); + return false; + } + if (delta >= 0) { + addUsed(reservation.entryBudget, delta); + } else { + subtractUsed(reservation.entryBudget, -delta); + } + reservation.bytes = newBytes; + return true; + } + } + + private void release(AdmissionReservation reservation) { + synchronized (lock) { + if (!reservation.active) { + return; + } + if (reservation.entryBudget.closed) { + reservation.bytes = 0L; + reservation.active = false; + return; + } + subtractUsed(reservation.entryBudget, reservation.bytes); + reservation.bytes = 0L; + reservation.active = false; + } + } + + private void close(EntryBudget entryBudget) { + synchronized (lock) { + if (entryBudget.closed) { + return; + } + if (entryBudget.entryBucket.usedWeight != 0L) { + long leakedWeight = entryBudget.entryBucket.usedWeight; + LOG.error("Force-closing external metadata cache budget {} with {} bytes still reserved", + entryBudget.scope, leakedWeight); + if (leakedWeight <= globalUsedWeight + && leakedWeight <= entryBudget.catalogBucket.usedWeight) { + globalUsedWeight -= leakedWeight; + entryBudget.catalogBucket.usedWeight -= leakedWeight; + entryBudget.entryBucket.usedWeight = 0L; + } else { + LOG.error("External metadata cache accounting is inconsistent while closing {}; " + + "globalUsed={}, catalogUsed={}, entryUsed={}", + entryBudget.scope, globalUsedWeight, + entryBudget.catalogBucket.usedWeight, leakedWeight); + globalUsedWeight = Math.max(0L, globalUsedWeight - leakedWeight); + entryBudget.catalogBucket.usedWeight = Math.max( + 0L, entryBudget.catalogBucket.usedWeight - leakedWeight); + entryBudget.entryBucket.usedWeight = 0L; + } + } + entryBudget.closed = true; + entryBudget.reclaimer = null; + entryBuckets.remove(entryBudget.scope, entryBudget.entryBucket); + entryBudgets.remove(entryBudget.scope, entryBudget); + Bucket catalogBucket = entryBudget.catalogBucket; + boolean catalogStillReferenced = entryBuckets.keySet().stream() Review Comment: [P2] Avoid scanning every budget while holding the global lock This close path runs once for each weighted entry, but while holding the FE-wide accounting lock it traverses every remaining entry scope to discover whether this catalog still has an entry. Dropping or reconfiguring one catalog therefore performs repeated O(total weighted entries) scans in the critical section, blocking unrelated reservation, release, statistics, creation, and close traffic across the FE. This remains after fixing the separate lifecycle-close and reclaim-sort threads. Please keep a per-catalog live-entry count in the bucket so this decision is O(1), and cover unrelated reservation progress during a many-catalog close. -- 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]
