github-actions[bot] commented on code in PR #66717:
URL: https://github.com/apache/doris/pull/66717#discussion_r3827238177
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/ExternalMetaCacheMgr.java:
##########
@@ -299,16 +530,54 @@ public MetaCacheEntryStats getEntryStats() {
private void initEngineCaches() {
registerBuiltinEngineCaches();
+ bindCatalogPreparers();
+ }
+
+ private void bindCatalogPreparers() {
+ for (ExternalMetaCache cache : cacheRegistry.allCaches()) {
+ String engine = cache.engine();
+ cache.bindCatalogPreparer(catalogId ->
tryPrepareCatalogByEngine(catalogId, engine));
+ }
+ }
+
+ /**
+ * Re-prepare a catalog whose group was retired between a caller's
preparation and its lookup.
+ * Lookups may run inside a cache loader, and retirement holds the
lifecycle lock while it
+ * closes groups, so this never blocks: when the fence is contended the
lookup keeps its
+ * pre-existing failure and the next call prepares under the new policy.
+ */
+ private void tryPrepareCatalogByEngine(long catalogId, String engine) {
+ ExternalMetaCache targetCache = this.engine(engine);
+ if (targetCache.isCatalogInitialized(catalogId)) {
+ return;
+ }
+ Lock lifecycleLock = catalogLifecycleLocks.get(catalogId);
+ if (!lifecycleLock.tryLock()) {
Review Comment:
[P1] Preserve lookups across a contended policy handoff
When an ALTER removes the entry group while holding this lifecycle stripe, a
lookup that already completed its initial prepare can reach
`requireCatalogEntryGroup()` after the detach. This `tryLock()` then fails, the
preparer returns without publishing a replacement, and
`requireCatalogEntryGroup()` immediately throws `IllegalStateException` even
though the catalog still exists. The new reprepare test only exercises the case
after removal has released the stripe. Please make retirement/replacement an
atomic handoff or provide a safe retry so a concurrent cache-policy ALTER
cannot fail an otherwise valid query.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/metacache/MetaCacheEntry.java:
##########
@@ -198,62 +552,738 @@ public MetaCacheEntryStats stats() {
failureCount,
totalLoadTime,
totalLoadCount == 0 ? 0D : (double) totalLoadTime /
totalLoadCount,
- cacheStats.evictionCount(),
+ MetaCacheWeightUtils.saturatedAdd(
+ cacheStats.evictionCount(), localEvictionCount.get()),
invalidateCount.get(),
lastLoadSuccessTimeMs.get(),
lastLoadFailureTimeMs.get(),
- lastError.get());
+ lastError.get(),
+ weightBounded,
+ weightBounded ? cacheSpec.getMaxWeight().getAsLong() : -1L,
+ weightBounded ? entryBudget.getUsedWeight() : -1L,
+ weightBounded ? MetaCacheWeightUtils.saturatedAdd(
+ automaticEvictionWeight.get(),
localEvictionWeight.get()) : -1L,
+ weightBounded ? weightAdmissionRejectedCount.get() : -1L,
+ weightBounded ? entryBudget.getCatalogMaxWeight() : -1L,
+ weightBounded ? entryBudget.getCatalogUsedWeight() : -1L,
+ weightBounded ? entryBudget.getGlobalMaxWeight() : -1L,
+ weightBounded ? entryBudget.getGlobalUsedWeight() : -1L,
+ weightBounded ? lastWeightRejectReason.get() : "");
+ }
+
+ public boolean isWeightBounded() {
+ return weightBounded;
+ }
+
+ /** True when this entry stores values at all (enabled with a positive
capacity or weight). */
+ public boolean isEffectivelyEnabled() {
+ return effectiveEnabled;
+ }
+
+ /**
+ * True when publication sizing can lead to a weighted admission: an entry
may be configured
+ * with a weight bound yet be ineffective (max-weight 0, disabled, zero
TTL or capacity), in
+ * which case preparing values for publication is pure waste.
+ */
+ public boolean isWeightAccounting() {
+ return weightBounded && effectiveEnabled;
+ }
+
+ private AdmissionResult admitWeightedValue(
+ K key, V value, @Nullable V expectedCurrent, boolean
requireExpected,
+ @Nullable KeyMutationToken expectedMutation, long
expectedReservationGeneration,
+ boolean advanceMutationOnAdmission) {
+ if (closed.get()) {
+ return AdmissionResult.DISABLED;
+ }
+ MetaCacheSizeEstimate estimate;
+ try {
+ estimate = Objects.requireNonNull(sizeEstimator.estimate(key,
value), "size estimate");
+ } catch (IllegalArgumentException e) {
+ throw e;
+ } catch (RuntimeException e) {
+ rejectWeight("invalid_estimate");
+ return AdmissionResult.REJECTED;
+ }
+ if (!estimate.isComplete()) {
+ rejectWeight(estimate.getIncompleteReason());
+ return AdmissionResult.REJECTED;
+ }
+
+ long estimatedPayloadBytes = estimate.getBytes();
+ // A retained non-null key/value plus Caffeine node can never consume
zero bytes. Treat a
+ // complete zero as an estimator contract violation so an omitted
formula cannot bypass
+ // every quota and admit an unbounded number of zero-weight entries.
+ if (estimatedPayloadBytes == 0L) {
+ rejectWeight("invalid_estimate");
+ return AdmissionResult.REJECTED;
+ }
+ long newWeight = MetaCacheWeightUtils.saturatedAdd(
+ estimatedPayloadBytes, FIXED_ENTRY_ACCOUNTING_OVERHEAD_BYTES);
+ synchronized (admissionLock) {
+ if (closed.get()) {
+ return AdmissionResult.DISABLED;
+ }
+ if (expectedMutation != null && !isKeyMutationCurrent(key,
expectedMutation)) {
+ return AdmissionResult.NOT_CURRENT;
+ }
+ V oldValue = data.asMap().get(key);
+ ReservationRecord record = reservations.get(key);
+ if (expectedReservationGeneration >= 0L
+ && (record == null || record.generation !=
expectedReservationGeneration)) {
+ return AdmissionResult.NOT_CURRENT;
+ }
+ if (requireExpected && oldValue != expectedCurrent) {
+ return AdmissionResult.NOT_CURRENT;
+ }
+ if (oldValue == null && record != null) {
+ // The previous generation was already removed by Caffeine; if
that removal was
+ // an eviction whose asynchronous cleanup has not run yet,
account it here.
+ if (pendingEvictionGenerations.remove(key, record.generation))
{
+ automaticEvictionWeight.accumulateAndGet(
+ record.weight, MetaCacheWeightUtils::saturatedAdd);
+ }
+ reservations.remove(key, record);
+ record.reservation.release();
+ record = null;
+ }
+ if (oldValue != null && (record == null || !record.published)) {
+ rejectWeight("missing_reservation");
+ return AdmissionResult.REJECTED;
+ }
+
+ if (record == null) {
+ Optional<AdmissionReservation> reservation =
reserveWithLocalEviction(key, newWeight);
+ if (!reservation.isPresent()) {
+ rejectWeight("budget_exceeded");
+ return AdmissionResult.REJECTED;
+ }
+ ReservationRecord newRecord = new ReservationRecord(
+ newWeight, reservation.get(),
nextReservationGeneration());
+ if (advanceMutationOnAdmission) {
+ advanceKeyMutation(key);
+ }
+ reservations.put(key, newRecord);
+ try {
+ beforeWeightedCachePutForTest(key, value);
+ data.put(key, value);
+ if (reservations.get(key) == newRecord &&
data.asMap().get(key) == value) {
+ newRecord.published = true;
+ notifyReplacement(key, null, value);
+ }
+ return AdmissionResult.ADMITTED;
+ } catch (RuntimeException | Error e) {
+ reservations.remove(key, newRecord);
+ newRecord.reservation.release();
+ throw e;
+ }
+ }
+
+ ReservationRecord previousRecord = record;
+ long reservedWeight = Math.max(previousRecord.weight, newWeight);
+ if (!resizeWithLocalEviction(key, previousRecord.reservation,
reservedWeight)) {
+ rejectWeight("budget_exceeded");
+ return AdmissionResult.REJECTED;
+ }
+ if (advanceMutationOnAdmission) {
+ advanceKeyMutation(key);
+ }
+ ReservationRecord newRecord = new ReservationRecord(
+ newWeight, previousRecord.reservation,
nextReservationGeneration());
+ reservations.put(key, newRecord);
+ try {
+ beforeWeightedCachePutForTest(key, value);
+ data.put(key, value);
+ boolean retained = reservations.get(key) == newRecord &&
data.asMap().get(key) == value;
+ if (retained) {
+ newRecord.published = true;
+ }
+ if (retained && reservedWeight != newWeight &&
!newRecord.reservation.tryResize(newWeight)) {
+ throw new IllegalStateException("failed to release cache
replacement reservation delta");
+ }
+ if (retained) {
+ notifyReplacement(key, oldValue, value);
+ }
+ return AdmissionResult.ADMITTED;
+ } catch (RuntimeException | Error e) {
+ if (reservations.replace(key, newRecord, previousRecord)) {
+ if (data.asMap().get(key) == null) {
+ reservations.remove(key, previousRecord);
+ previousRecord.reservation.release();
+ } else if
(!previousRecord.reservation.tryResize(previousRecord.weight)) {
+ throw new IllegalStateException("failed to roll back
cache replacement reservation", e);
+ }
+ }
+ throw e;
+ }
+ }
+ }
+
+ private Optional<AdmissionReservation> reserveWithLocalEviction(K
incomingKey, long bytes) {
+ if (bytes > entryBudget.getEffectiveMaxWeight()) {
+ return Optional.empty();
+ }
+ Optional<AdmissionReservation> reservation =
entryBudget.tryReserve(bytes);
+ while (!reservation.isPresent()) {
+ AtomicReference<Optional<AdmissionReservation>> retried =
+ new AtomicReference<>(Optional.empty());
+ int evicted = evictLocalColdest(incomingKey,
LOCAL_EVICTION_BATCH_SIZE, () -> {
+ Optional<AdmissionReservation> attempt =
entryBudget.tryReserve(bytes);
+ retried.set(attempt);
+ return attempt.isPresent();
+ });
+ reservation = retried.get();
+ if (reservation.isPresent()) {
+ break;
+ }
+ if (evicted == 0) {
+ entryBudget.requestPeerReclaim(bytes);
+ break;
+ }
+ }
+ return reservation;
+ }
+
+ private boolean resizeWithLocalEviction(K incomingKey,
AdmissionReservation reservation, long newBytes) {
+ if (newBytes > entryBudget.getEffectiveMaxWeight()) {
+ return false;
+ }
+ if (reservation.tryResize(newBytes)) {
+ return true;
+ }
+ while (true) {
+ AtomicBoolean resized = new AtomicBoolean();
+ int evicted = evictLocalColdest(incomingKey,
LOCAL_EVICTION_BATCH_SIZE, () -> {
+ if (reservation.tryResize(newBytes)) {
+ resized.set(true);
+ return true;
+ }
+ return false;
+ });
+ if (resized.get()) {
+ return true;
+ }
+ if (evicted == 0) {
+ entryBudget.requestPeerReclaim(Math.max(0L, newBytes -
reservation.getBytes()));
+ return false;
+ }
+ }
+ }
+
+ /**
+ * Evict up to {@code limit} coldest values, retrying the caller's goal
after every single
+ * eviction: a skewed cache where one large cold value creates all needed
headroom must not
+ * discard the rest of the selected batch.
+ */
+ private int evictLocalColdest(K incomingKey, int limit, BooleanSupplier
stopAfterEviction) {
+ if (!data.policy().eviction().isPresent()) {
+ return 0;
+ }
+ Map<K, V> coldest = data.policy().eviction().get().coldest(limit);
+ int evicted = 0;
+ for (Map.Entry<K, V> candidate : coldest.entrySet()) {
+ if (Objects.equals(candidate.getKey(), incomingKey)) {
+ continue;
+ }
+ V current = data.asMap().get(candidate.getKey());
+ ReservationRecord record = reservations.get(candidate.getKey());
+ long evictedWeight = record != null && record.published && current
!= null ? record.weight : 0L;
+ if (current == candidate.getValue() &&
data.asMap().remove(candidate.getKey(), current)) {
+ if (record != null) {
+ releaseReservation(candidate.getKey(), record.generation);
+ }
+ localEvictionCount.incrementAndGet();
+ localEvictionWeight.accumulateAndGet(evictedWeight,
MetaCacheWeightUtils::saturatedAdd);
+ evicted++;
+ if (stopAfterEviction.getAsBoolean()) {
+ return evicted;
+ }
+ }
+ }
+ return evicted;
+ }
+
+ private long reclaimForPeer(long targetBytes) {
+ if (targetBytes <= 0L || closed.get()) {
+ return 0L;
+ }
+ synchronized (admissionLock) {
+ long before = entryBudget.getUsedWeight();
+ long reclaimed = 0L;
+ while (reclaimed < targetBytes
+ && evictLocalColdest(null, LOCAL_EVICTION_BATCH_SIZE, () ->
+ Math.max(0L, before - entryBudget.getUsedWeight())
>= targetBytes) > 0) {
+ reclaimed = Math.max(0L, before - entryBudget.getUsedWeight());
+ }
+ return reclaimed;
+ }
+ }
+
+ private int weigh(K key, V value) {
+ ReservationRecord record = reservations.get(key);
+ // Every supported write path installs the reservation record before
calling data.put.
+ // Missing ownership is an invariant violation, so fail closed without
invoking an O(n)
+ // estimator from Caffeine's hot weigher callback.
+ long weight = record == null ? Integer.MAX_VALUE : record.weight;
+ return weight >= Integer.MAX_VALUE ? Integer.MAX_VALUE : (int) weight;
+ }
+
+ private void onRemoval(@Nullable K key, @Nullable V value, RemovalCause
cause) {
+ if (key == null) {
+ return;
+ }
+ if (!weightBounded && !generationFencedRefresh && removalListener ==
null) {
+ return;
+ }
+ if (closed.get()) {
+ return;
+ }
+ // Replacement transfers the existing reservation to the newly
published generation. A
+ // soft-value collection instead reports a null value with COLLECTED
and must release it.
+ if (cause == RemovalCause.REPLACED) {
+ return;
+ }
+ if (removalListener != null) {
+ pendingRemovalNotifications.add(new RemovedToken<>(key,
removalToken(value)));
+ scheduleRemovalCleanup();
+ }
+ if (Thread.holdsLock(admissionLock)) {
+ // Other removals have already removed the Caffeine mapping and
can release their owner
+ // inline. A stale callback cannot release a replacement while its
mapping is visible.
+ if (data.asMap().get(key) == null) {
+ if (weightBounded) {
+ ReservationRecord record = reservations.get(key);
+ if (record != null) {
+ if (cause.wasEvicted()) {
+ automaticEvictionWeight.accumulateAndGet(
+ record.weight,
MetaCacheWeightUtils::saturatedAdd);
+ }
+ releaseReservation(key, record.generation);
+ }
+ } else {
+ RefreshRecord record = refreshRecords.get(key);
+ if (record != null) {
+ releaseRefreshRecord(key, record.generation);
+ }
+ }
+ }
+ return;
+ }
+ beforeRemovalOwnerSnapshotForTest(key);
+ long ownerGeneration = currentOwnerGeneration(key);
Review Comment:
[P2] Attribute delayed evictions to the removed generation
A COLLECTED/EXPIRED callback can pause after Caffeine removes G1 while a
concurrent put releases G1 and publishes G2. This lookup then returns G2, so
the old eviction is recorded under G2; cleanup sees G2 still mapped and drops
the marker without adding G1's weight. Quota ownership stays safe, but the new
exact `eviction_weight` telemetry under-reports precisely this delayed-callback
case. Please retain the removed generation/weight independently of the current
owner and extend the existing delayed-collection test to assert G1 is counted
exactly once.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergTableCacheValue.java:
##########
@@ -17,25 +17,179 @@
package org.apache.doris.datasource.iceberg;
-import com.google.common.base.Suppliers;
+import org.apache.doris.datasource.NameMapping;
+import org.apache.doris.datasource.metacache.MetaCacheSizeEstimate;
+import org.apache.doris.datasource.metacache.MetaCacheSizeEstimator;
+
+import org.apache.iceberg.HasTableOperations;
import org.apache.iceberg.Table;
+import org.apache.iceberg.TableMetadata;
+import org.apache.iceberg.io.FileIO;
+import org.apache.iceberg.io.SupportsStorageCredentials;
-import java.util.function.Supplier;
+import java.util.Objects;
+import java.util.Optional;
public class IcebergTableCacheValue {
- private final Table icebergTable;
- private final Supplier<IcebergSnapshotCacheValue> latestSnapshotCacheValue;
+ private volatile Table icebergTable;
+ private String retainedCurrentSnapshotJson;
+ private volatile boolean queryIsolationPrepared;
+ private long retainedTablePayloadBytes;
+ private MetaCacheSizeEstimate sizeEstimate;
- public IcebergTableCacheValue(Table icebergTable,
Supplier<IcebergSnapshotCacheValue> latestSnapshotCacheValue) {
- this.icebergTable = icebergTable;
- this.latestSnapshotCacheValue =
Suppliers.memoize(latestSnapshotCacheValue::get);
+ public IcebergTableCacheValue(Table icebergTable) {
+ this.icebergTable =
IcebergSnapshotCacheValue.retainTableGeneration(icebergTable);
}
public Table getIcebergTable() {
+ Table retainedTable = icebergTable;
+ return queryIsolationPrepared ||
IcebergSnapshotCacheValue.isNonGrowingGeneration(retainedTable)
+ ? IcebergSnapshotCacheValue.createQueryScopedTable(
+ retainedTable, retainedCurrentSnapshotJson)
+ : retainedTable;
+ }
+
+ public Table getWritableIcebergTable(Table liveTable) {
+ Table retainedTable = icebergTable;
+ return IcebergSnapshotCacheValue.createWritableTable(retainedTable,
liveTable);
+ }
+
+ synchronized MetaCacheSizeEstimate prepareForCachePublication(NameMapping
key) {
+ if (sizeEstimate == null) {
+ sizeEstimate =
MetaCacheSizeEstimator.estimateSafely("iceberg_table_preparation_failed",
+ () -> {
+ // Order matters: serializing a v1 snapshot
materializes its transient
+ // manifest list, which the payload accounting
rejects, so account first.
+ retainedTablePayloadBytes =
+
IcebergCacheSizeEstimator.retainedTablePayloadBytes(icebergTable);
+ retainedCurrentSnapshotJson =
+
IcebergSnapshotCacheValue.retainCurrentSnapshotJson(icebergTable);
+ return
IcebergCacheSizeEstimator.estimateTableEntry(key, this);
+ });
+ if (sizeEstimate.isComplete()) {
+ icebergTable =
IcebergSnapshotCacheValue.retainNonGrowingGeneration(icebergTable);
+ queryIsolationPrepared = true;
+ }
+ }
+ return sizeEstimate;
+ }
+
+ public MetaCacheSizeEstimate getSizeEstimate() {
+ return sizeEstimate == null
+ ? MetaCacheSizeEstimate.incomplete("not_prepared") :
sizeEstimate;
+ }
+
+ Table getRetainedIcebergTable() {
return icebergTable;
}
- public IcebergSnapshotCacheValue getLatestSnapshotCacheValue() {
- return latestSnapshotCacheValue.get();
+ synchronized Table newQueryScopedTable() {
+ if (!queryIsolationPrepared) {
+ // A failed optional size preparation must only reject weighted
cache admission. Do not
+ // repeat the same unsupported metadata access on the query path
and turn it into a
+ // table-load failure; this value is not retained by the weighted
cache in that case.
+ if (sizeEstimate != null && !sizeEstimate.isComplete()) {
+ return icebergTable;
+ }
+ retainedCurrentSnapshotJson =
+
IcebergSnapshotCacheValue.retainCurrentSnapshotJson(icebergTable);
+ icebergTable =
IcebergSnapshotCacheValue.retainNonGrowingGeneration(icebergTable);
+ queryIsolationPrepared = true;
+ }
+ return IcebergSnapshotCacheValue.createQueryScopedTable(
+ icebergTable, retainedCurrentSnapshotJson);
+ }
+
+ String getRetainedCurrentSnapshotJson() {
+ return retainedCurrentSnapshotJson;
+ }
+
+ boolean isQueryIsolationPrepared() {
+ return queryIsolationPrepared;
+ }
+
+ long getRetainedTablePayloadBytes() {
+ return retainedTablePayloadBytes;
+ }
+
+ long getRetainedCurrentSnapshotPayloadBytes() {
+ return IcebergSnapshotCacheValue.retainedSnapshotJsonBytes(
+ retainedCurrentSnapshotJson);
+ }
+
+ Optional<String> getTableUuid() {
+ TableMetadata metadata = retainedMetadata();
+ return metadata == null || metadata.uuid() == null ||
metadata.uuid().isEmpty()
+ ? Optional.empty() : Optional.of(metadata.uuid());
+ }
+
+ boolean isSamePhysicalGeneration(IcebergTableCacheValue other) {
+ if (other == null) {
+ return false;
+ }
+ TableMetadata left = retainedMetadata();
+ TableMetadata right = other.retainedMetadata();
+ return left != null && right != null
+ && Objects.equals(left.uuid(), right.uuid())
+ && Objects.equals(left.metadataFileLocation(),
right.metadataFileLocation());
+ }
+
+ /**
+ * Same metadata generation served through the same operational resources.
A catalog that
+ * reloads the same metadata file may still hand out a new FileIO carrying
rotated vended
+ * credentials; projections frozen on the previous handle must not outlive
that rotation.
+ */
+ boolean isSameOperationalGeneration(IcebergTableCacheValue other) {
+ return isSamePhysicalGeneration(other) &&
sharesOperationalResources(other.icebergTable);
+ }
+
+ boolean sharesOperationalResources(Table table) {
+ return sharesOperationalResources(icebergTable, table);
+ }
+
+ /** True when both tables read and write through the same FileIO,
encryption and locations. */
+ static boolean sharesOperationalResources(Table left, Table right) {
+ if (left == right) {
+ return true;
+ }
+ if (left == null || right == null) {
+ return false;
+ }
+ return sameFileIo(left.io(), right.io())
+ && sameResource(left.encryption(), right.encryption())
+ && sameResource(left.locationProvider(),
right.locationProvider());
+ }
+
+ private static boolean sameFileIo(FileIO left, FileIO right) {
+ if (left == right) {
+ return true;
+ }
+ if (left == null || right == null || left.getClass() !=
right.getClass()) {
+ return false;
+ }
+ try {
+ // Vended credentials live in the FileIO properties / storage
credentials; equal
+ // configuration means equal credentials even across catalog
reload instances.
+ return Objects.equals(left.properties(), right.properties())
+ && Objects.equals(storageCredentials(left),
storageCredentials(right));
+ } catch (RuntimeException e) {
+ // A FileIO that cannot expose its configuration cannot prove it
is unchanged.
+ return false;
+ }
+ }
+
+ private static Object storageCredentials(FileIO fileIO) {
+ return fileIO instanceof SupportsStorageCredentials
+ ? ((SupportsStorageCredentials) fileIO).credentials() : null;
+ }
+
+ private static boolean sameResource(Object left, Object right) {
+ return left == right || (left != null && right != null &&
left.getClass() == right.getClass());
Review Comment:
[P1] Do not equate same-class resource replacements
The FileIO-rotation follow-up now proves equivalence from properties and
credentials, but this helper treats any two same-class encryption managers or
location providers as unchanged. Distinct instances can carry different KMS
clients, sessions, credentials, or custom provider state;
`FrozenTableOperations` strongly retains the old instances, so both
snapshot-hit revalidation and table-replacement invalidation keep using them.
This is the remaining same-class case not covered by the existing FileIO
thread. Please use an immutable state fingerprint when one exists, or fail
closed to identity/value equality, and add same-class replacement tests for
both resources.
--
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]
