github-actions[bot] commented on code in PR #66717: URL: https://github.com/apache/doris/pull/66717#discussion_r3775088151
########## fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergSnapshotEntryKey.java: ########## @@ -0,0 +1,107 @@ +// 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.iceberg; + +import org.apache.doris.datasource.NameMapping; + +import org.apache.iceberg.HasTableOperations; +import org.apache.iceberg.Snapshot; +import org.apache.iceberg.Table; +import org.apache.iceberg.TableMetadata; + +import java.util.Objects; +import java.util.Optional; + +/** Stable identity for an Iceberg snapshot projection built from one frozen metadata generation. */ +public final class IcebergSnapshotEntryKey { + private final NameMapping nameMapping; + private final String metadataFileLocation; + private final long snapshotId; + private final int schemaId; + private final int defaultSpecId; + + private IcebergSnapshotEntryKey(NameMapping nameMapping, String metadataFileLocation, + long snapshotId, int schemaId, int defaultSpecId) { + this.nameMapping = Objects.requireNonNull(nameMapping, "nameMapping can not be null"); + this.metadataFileLocation = Objects.requireNonNull( + metadataFileLocation, "metadataFileLocation can not be null"); + this.snapshotId = snapshotId; + this.schemaId = schemaId; + this.defaultSpecId = defaultSpecId; + } + + /** + * Build a key from the same retained table generation that will be used by the value loader. + * Tables without a stable metadata location intentionally bypass the snapshot cache. + */ + public static Optional<IcebergSnapshotEntryKey> tryCreate(NameMapping nameMapping, Table retainedTable) { + if (!(retainedTable instanceof HasTableOperations)) { + return Optional.empty(); + } + TableMetadata metadata = ((HasTableOperations) retainedTable).operations().current(); + if (metadata == null || metadata.metadataFileLocation() == null + || metadata.metadataFileLocation().isEmpty()) { + return Optional.empty(); + } + Snapshot snapshot = metadata.currentSnapshot(); + long snapshotId = snapshot == null ? IcebergUtils.UNKNOWN_SNAPSHOT_ID : snapshot.snapshotId(); + return Optional.of(new IcebergSnapshotEntryKey(nameMapping, metadata.metadataFileLocation(), Review Comment: [P1] Include the physical table generation in this key. HadoopCatalog reuses the deterministic `metadata/v1.metadata.json` path after a purged same-name drop/recreate, and an empty replacement also resets snapshot/schema/spec IDs to `-1/0/0`; its UUID is new, but every field here collides. After the table entry refreshes, this contextual entry can therefore return the old retained table. The same collision also passes `isSameGeneration()`, which accepts equal locations without checking UUID. Please key/fence on UUID or a table-entry generation and cover an empty HadoopCatalog drop/recreate. ########## fe/fe-core/src/main/java/org/apache/doris/datasource/ExternalCatalog.java: ########## @@ -1367,9 +1381,21 @@ public int hashCode() { public void notifyPropertiesUpdated(Map<String, String> updatedProps) { CatalogIf.super.notifyPropertiesUpdated(updatedProps); String schemaCacheTtl = updatedProps.getOrDefault(SCHEMA_CACHE_TTL_SECOND, null); - if (java.util.Objects.nonNull(schemaCacheTtl)) { - ExternalMetaCacheMgr extMetaCacheMgr = Env.getCurrentEnv().getExtMetaCacheMgr(); + ExternalMetaCacheMgr extMetaCacheMgr = Env.getCurrentEnv().getExtMetaCacheMgr(); Review Comment: [P1] Fence this new quota invalidation against an in-flight first initialization. `prepareCatalogByEngine()` can copy the old properties while no group exists; if ALTER commits this setting next, `removeCatalog()` skips the absent group, and the delayed initializer then publishes the old count-only policy indefinitely. That silently defeats the configured memory bound. Please version/serialize the property snapshot with removal and publication, and add a paused ALTER-vs-init test that verifies the new weighted policy wins. ########## fe/fe-core/src/main/java/org/apache/doris/datasource/metacache/ExternalMetaCacheBudgetManager.java: ########## @@ -0,0 +1,386 @@ +// 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 java.util.HashMap; +import java.util.Map; +import java.util.Objects; +import java.util.Optional; +import java.util.OptionalLong; +import java.util.concurrent.atomic.AtomicLong; + +/** + * 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 { + 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 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"); + } + if (globalMaxWeight.isPresent() && parsed > globalMaxWeight.getAsLong()) { Review Comment: [P1] Do not reject replayed catalogs against this FE's local global cap. `external_meta_cache_max_weight` is per-FE and may be a percentage of local heap, while `meta.cache.max-weight` is persisted after validation only on the master. For example, a 4 GB catalog cap accepted with `global=20%` on a 32 GB master will fail every lazy cache initialization on an 8 GB observer, because replay skips DDL validation and this check runs on access. Please let the local global bucket clamp the effective admission limit (while keeping DDL hierarchy validation), and cover heterogeneous-heap replay. ########## fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergTableCacheValue.java: ########## @@ -17,25 +17,45 @@ package org.apache.doris.datasource.iceberg; -import com.google.common.base.Suppliers; -import org.apache.iceberg.Table; +import org.apache.doris.datasource.NameMapping; +import org.apache.doris.datasource.metacache.MetaCacheSizeEstimate; +import org.apache.doris.datasource.metacache.MetaCacheSizeEstimator; -import java.util.function.Supplier; +import org.apache.iceberg.Table; public class IcebergTableCacheValue { - private final Table icebergTable; - private final Supplier<IcebergSnapshotCacheValue> latestSnapshotCacheValue; + private Table icebergTable; + private MetaCacheSizeEstimate sizeEstimate; - public IcebergTableCacheValue(Table icebergTable, Supplier<IcebergSnapshotCacheValue> latestSnapshotCacheValue) { + public IcebergTableCacheValue(Table icebergTable) { this.icebergTable = icebergTable; - this.latestSnapshotCacheValue = Suppliers.memoize(latestSnapshotCacheValue::get); } public Table getIcebergTable() { return icebergTable; } - public IcebergSnapshotCacheValue getLatestSnapshotCacheValue() { - return latestSnapshotCacheValue.get(); + public Table getWritableIcebergTable() { + return IcebergSnapshotCacheValue.createServiceBackedTable(icebergTable); + } + + MetaCacheSizeEstimate prepareForCachePublication(NameMapping key) { + if (sizeEstimate == null) { + sizeEstimate = MetaCacheSizeEstimator.estimateSafely("iceberg_table_preparation_failed", () -> { + icebergTable = IcebergSnapshotCacheValue.detachTableGeneration(icebergTable); + IcebergSnapshotCacheValue.materializeCurrentSnapshotManifests(icebergTable); Review Comment: [P2] Keep manifest materialization inside the catalog authentication scope. The loader's `getExecutionAuthenticator().execute(...)` ends after `ops.loadTable()`, but weighted preparation later calls `dataManifests(table.io())` / `deleteManifests(table.io())` here. For the Kerberized Hadoop catalog, credentials are supplied only inside `HadoopExecutionAuthenticator.execute`, so this manifest-list read can fail; `estimateSafely` then marks the value incomplete and every weighted table lookup is returned uncached (the snapshot estimator has the same problem). Please run remote-I/O preparation under the owning catalog authenticator, with a credential-scoped admission/hit regression. ########## fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/PaimonExternalMetaCache.java: ########## @@ -86,7 +100,14 @@ public Table getPaimonTable(NameMapping nameMapping) { public PaimonSnapshotCacheValue getSnapshotCache(ExternalTable dorisTable) { NameMapping nameMapping = dorisTable.getOrBuildNameMapping(); - return tableEntry.get(nameMapping.getCtlId()).get(nameMapping).getLatestSnapshotCacheValue(); + PaimonTableCacheValue tableValue = tableEntry.get(nameMapping.getCtlId()).get(nameMapping); Review Comment: [P2] Avoid resolving the remote fence before every snapshot-cache lookup. `loadFence()` runs before `snapshotEntry.get()`, and its path calls `copyWithLatestSchema()`, `latestSnapshot()`, and `schemaManager().latest()`, so even a hit on an admitted snapshot still performs latest-metadata discovery. Before this change, `PaimonTableCacheValue` memoized the projection, so stable repeated reads avoided that work. Please retain or refresh the fence under the table generation (or otherwise put discovery behind a cache) and add a repeated-hit call-count test. -- 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]
