github-actions[bot] commented on code in PR #66717:
URL: https://github.com/apache/doris/pull/66717#discussion_r3799411592
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/PaimonExternalMetaCache.java:
##########
@@ -86,41 +104,85 @@ 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);
+ PaimonSnapshot fence = loadLatestSnapshotFence(nameMapping,
tableValue.getPaimonTable()).getSnapshot();
+ PaimonSnapshotEntryKey key = PaimonSnapshotEntryKey.of(
Review Comment:
[P2] Retire superseded snapshots within the current table generation
This method re-reads the latest fence on every call and includes its
snapshot/schema IDs in the key, but `retireTableGeneration` only removes keys
from other synthetic table generations. Each commit observed before
`tableEntry` refresh thus publishes a new full partition/table projection while
earlier same-generation values can never be looked up again; they remain
charged until the next successful base refresh, 24-hour access expiry, or
capacity/weight eviction, so a busy table can displace or reject current
metadata. Please race-safely retain only the newest latest key per
`(NameMapping, tableGeneration)`, and test advancing IDs without replacing
`tableValue`, including reversed concurrent completion.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergExternalMetaCache.java:
##########
@@ -159,25 +285,32 @@ public void invalidateCatalogEntries(long catalogId) {
}
private IcebergTableCacheValue loadTableCacheValue(NameMapping
nameMapping) {
- CatalogIf catalog =
Env.getCurrentEnv().getCatalogMgr().getCatalog(nameMapping.getCtlId());
+ CatalogIf catalog = getCatalog(nameMapping.getCtlId());
if (catalog == null) {
throw new RuntimeException(String.format("Cannot find catalog %d
when loading table %s/%s.",
nameMapping.getCtlId(), nameMapping.getLocalDbName(),
nameMapping.getLocalTblName()));
}
IcebergMetadataOps ops = resolveMetadataOps(catalog);
- try {
- Table table = ((ExternalCatalog)
catalog).getExecutionAuthenticator()
- .execute(() ->
ops.loadTable(nameMapping.getRemoteDbName(), nameMapping.getRemoteTblName()));
- ExternalTable dorisTable = findExternalTable(nameMapping, ENGINE);
- return new IcebergTableCacheValue(table, () ->
loadSnapshotProjection(dorisTable, table));
- } catch (Exception e) {
- throw new RuntimeException(ExceptionUtils.getRootCauseMessage(e),
e);
- }
+ return executeAuthenticated(catalog, () -> {
+ Table table = ops.loadTable(nameMapping.getRemoteDbName(),
nameMapping.getRemoteTblName());
+ IcebergTableCacheValue value = new IcebergTableCacheValue(table);
+ MetaCacheEntry<NameMapping, IcebergTableCacheValue> currentEntry =
+ tableEntry.getIfInitialized(nameMapping.getCtlId());
+ if (currentEntry != null && currentEntry.isWeightBounded()) {
Review Comment:
[P2] Skip publication sizing when the entry is ineffective
An entry can be ineffective yet remain `isWeightBounded()` (for example
`max-weight=0`, or `enable=false`/zero TTL/capacity combined with a direct,
catalog, or global weight). This check therefore runs full
`prepareTableForCachePublication()` on every uncached lookup even though
`MetaCacheEntry` cannot admit the value, repeatedly walking/serializing the
retained metadata graph solely for governance that can never produce a hit. The
snapshot path at lines 187/196 and manifest accounting flag at line 272 use the
same mismatched predicate. Please require both weight-bounded and effectively
enabled for publication preparation/accounting, with ineffective weighted
table/snapshot/manifest regressions.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergExternalMetaCache.java:
##########
@@ -108,13 +128,92 @@ public Table getIcebergTable(ExternalTable dorisTable) {
return
tableEntry.get(nameMapping.getCtlId()).get(nameMapping).getIcebergTable();
}
+ public Table getWritableIcebergTable(ExternalTable dorisTable) {
+ NameMapping nameMapping = dorisTable.getOrBuildNameMapping();
+ CatalogIf catalog = getCatalog(nameMapping.getCtlId());
+ if (catalog == null) {
+ throw new RuntimeException("Cannot find catalog " +
nameMapping.getCtlId()
+ + " when loading a writable Iceberg table");
+ }
+ IcebergMetadataOps ops = resolveMetadataOps(catalog);
+ // DDL/actions must start from the live catalog generation. DML that
was planned against a
+ // retained read generation wraps this live table separately in
IcebergTransaction.
+ return executeAuthenticated(catalog, () -> ops.loadTable(
+ nameMapping.getRemoteDbName(),
nameMapping.getRemoteTblName()));
+ }
+
+ Table getQueryScopedIcebergTable(ExternalTable dorisTable) {
+ NameMapping nameMapping = dorisTable.getOrBuildNameMapping();
+ MetaCacheEntry<NameMapping, IcebergTableCacheValue> entry =
+ tableEntry.get(nameMapping.getCtlId());
+ IcebergTableCacheValue tableValue =
+ entry.get(nameMapping);
+ return createQueryTable(nameMapping, tableValue);
+ }
+
+ private Table createQueryTable(
+ NameMapping nameMapping, IcebergTableCacheValue tableValue) {
+ boolean isolateForQueries = tableValue.isQueryIsolationPrepared()
+ || snapshotEntry.get(nameMapping.getCtlId()).isWeightBounded();
+ if (!isolateForQueries) {
+ return tableValue.getIcebergTable();
+ }
+ Table queryTable = tableValue.newQueryScopedTable();
+ IcebergSnapshotCacheValue.loadQueryMetadataForStatement(queryTable);
+ return queryTable;
+ }
+
public IcebergSnapshotCacheValue getSnapshotCache(ExternalTable
dorisTable) {
NameMapping nameMapping = dorisTable.getOrBuildNameMapping();
- return
tableEntry.get(nameMapping.getCtlId()).get(nameMapping).getLatestSnapshotCacheValue();
+ IcebergTableCacheValue tableValue =
+ tableEntry.get(nameMapping.getCtlId()).get(nameMapping);
+ Table retainedTable = tableValue.getRetainedIcebergTable();
+ java.util.Optional<IcebergSnapshotEntryKey> optionalKey =
+ IcebergSnapshotEntryKey.tryCreate(nameMapping, retainedTable);
+ if (!optionalKey.isPresent()) {
+ boolean isolateForQueries = tableValue.isQueryIsolationPrepared();
+ return executeAuthenticated(nameMapping.getCtlId(),
+ () -> loadSnapshotProjection(
+ dorisTable,
+ isolateForQueries ?
tableValue.newQueryScopedTable()
+ : tableValue.getIcebergTable(),
+ tableValue.getRetainedIcebergTable(),
+ tableValue.getRetainedCurrentSnapshotJson(),
isolateForQueries));
+ }
+ IcebergSnapshotEntryKey key = optionalKey.get();
+ MetaCacheEntry<IcebergSnapshotEntryKey, IcebergSnapshotCacheValue>
entry =
+ snapshotEntry.get(nameMapping.getCtlId());
+ boolean isolateForQueries = tableValue.isQueryIsolationPrepared()
+ || entry.isWeightBounded();
+ IcebergSnapshotCacheValue snapshotValue = entry.get(key,
+ ignored -> executeAuthenticated(nameMapping.getCtlId(), () -> {
+ Table projectionTable = isolateForQueries
+ ? tableValue.newQueryScopedTable() :
tableValue.getIcebergTable();
+ IcebergSnapshotCacheValue value = loadSnapshotProjection(
+ dorisTable, projectionTable,
+ tableValue.getRetainedIcebergTable(),
+ tableValue.getRetainedCurrentSnapshotJson(),
isolateForQueries);
+ if (entry.isWeightBounded()) {
+ value.prepareForCachePublication(key);
+ }
+ return value;
+ }));
+ MetaCacheEntry<NameMapping, IcebergTableCacheValue> tables =
tableEntry.get(nameMapping.getCtlId());
+ IcebergTableCacheValue currentTable =
tables.peekIfPresent(nameMapping);
+ if (tables.isEffectivelyEnabled()
Review Comment:
[P1] Revalidate snapshot resources when the base is ineffective
A valid `meta.cache.iceberg.table.max-weight=0` makes the base entry
ineffective while leaving snapshot caching enabled by default;
`table.enable=false` with an explicit `snapshot.enable=true` reaches the same
state. The table lookup then returns a fresh uncached handle on every call, but
this guard skips all checks and a same-physical-key snapshot hit keeps the old
`FrozenTableOperations`/`FileIO`; no base replacement listener can retire it,
so regularly accessed scans can reuse expired vended credentials indefinitely.
Please compare the fresh `tableValue` with retained snapshot resources whenever
the base is ineffective (or disable that child state), and test same-metadata
credential rotation under both configurations.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/PaimonExternalMetaCache.java:
##########
@@ -86,41 +104,85 @@ 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);
+ PaimonSnapshot fence = loadLatestSnapshotFence(nameMapping,
tableValue.getPaimonTable()).getSnapshot();
+ PaimonSnapshotEntryKey key = PaimonSnapshotEntryKey.of(
+ nameMapping, fence, tableValue.getGeneration());
+ MetaCacheEntry<PaimonSnapshotEntryKey, PaimonSnapshotCacheValue> entry
=
+ snapshotEntry.get(nameMapping.getCtlId());
+ PaimonSnapshotCacheValue snapshotValue = entry.get(key,
+ ignored -> executeAuthenticated(nameMapping,
+ () -> latestSnapshotProjectionLoader.loadAtFence(
+ nameMapping, fence,
tableValue.getGeneration())));
+ if (!isCurrentTableGeneration(nameMapping,
tableValue.getGeneration())) {
+ entry.invalidateKeyIfSame(key, snapshotValue);
+ }
+ return snapshotValue;
}
public PaimonSnapshotCacheValue loadSnapshotProjection(ExternalTable
dorisTable, Table effectiveTable) {
- return
latestSnapshotProjectionLoader.load(dorisTable.getOrBuildNameMapping(),
effectiveTable);
+ NameMapping nameMapping = dorisTable.getOrBuildNameMapping();
+ return executeAuthenticated(nameMapping,
+ () -> latestSnapshotProjectionLoader.load(nameMapping,
effectiveTable));
}
public PaimonSnapshotCacheValue loadLatestSnapshotFence(ExternalTable
dorisTable) {
NameMapping nameMapping = dorisTable.getOrBuildNameMapping();
- Table table =
tableEntry.get(nameMapping.getCtlId()).get(nameMapping).getPaimonTable();
- return latestSnapshotProjectionLoader.loadFence(nameMapping, table);
+ PaimonTableCacheValue tableValue =
tableEntry.get(nameMapping.getCtlId()).get(nameMapping);
+ return loadLatestSnapshotFence(nameMapping,
tableValue.getPaimonTable());
}
public PaimonSnapshotCacheValue loadSnapshotAtFence(
ExternalTable dorisTable, PaimonSnapshot fence) {
NameMapping nameMapping = dorisTable.getOrBuildNameMapping();
- return latestSnapshotProjectionLoader.loadAtFence(nameMapping, fence);
+ return executeAuthenticated(nameMapping,
+ () -> latestSnapshotProjectionLoader.loadAtFence(nameMapping,
fence));
}
public PaimonSnapshotCacheValue loadSnapshotAtFence(
ExternalTable dorisTable, Table effectiveTable, PaimonSnapshot
fence) {
- return latestSnapshotProjectionLoader.loadEffectiveAtFence(
- dorisTable.getOrBuildNameMapping(), effectiveTable, fence);
+ NameMapping nameMapping = dorisTable.getOrBuildNameMapping();
+ return executeAuthenticated(nameMapping,
+ () -> latestSnapshotProjectionLoader.loadEffectiveAtFence(
+ nameMapping, effectiveTable, fence));
}
public PaimonSchemaCacheValue getPaimonSchemaCacheValue(NameMapping
nameMapping, long schemaId) {
- SchemaCacheValue schemaCacheValue =
schemaEntry.get(nameMapping.getCtlId())
- .get(new PaimonSchemaCacheKey(nameMapping, schemaId));
+ PaimonTableCacheValue tableValue =
tableEntry.get(nameMapping.getCtlId()).get(nameMapping);
+ return getPaimonSchemaCacheValue(
+ nameMapping, schemaId, tableValue.getGeneration(),
tableValue.getPaimonTable());
+ }
+
+ PaimonSchemaCacheValue getPaimonSchemaCacheValue(
+ NameMapping nameMapping, long schemaId, long tableGeneration,
Table retainedTable) {
+ PaimonSchemaCacheKey key = new PaimonSchemaCacheKey(nameMapping,
tableGeneration, schemaId);
+ if (tableGeneration <= 0L) {
+ return (PaimonSchemaCacheValue) executeAuthenticated(nameMapping,
+ () -> loadSchemaCacheValue(key, retainedTable));
+ }
+ MetaCacheEntry<PaimonSchemaCacheKey, SchemaCacheValue> entry =
schemaEntry.get(nameMapping.getCtlId());
+ SchemaCacheValue schemaCacheValue = entry.get(key,
+ ignored -> executeAuthenticated(nameMapping,
+ () -> loadSchemaCacheValue(key, retainedTable)));
+ if (!isCurrentTableGeneration(nameMapping, tableGeneration)) {
+ entry.invalidateKeyIfSame(key, schemaCacheValue);
+ }
return (PaimonSchemaCacheValue) schemaCacheValue;
}
+ /**
+ * Snapshot and schema projections are keyed by the synthetic generation
of the base table
+ * handle they were derived from. A generation that is no longer published
(replaced, expired,
+ * or never admitted because its weight estimate was rejected) can never
be looked up again,
+ * so its projections must not stay behind in the child entries.
+ */
+ private boolean isCurrentTableGeneration(NameMapping nameMapping, long
tableGeneration) {
+ PaimonTableCacheValue currentTable =
tableEntry.get(nameMapping.getCtlId()).peekIfPresent(nameMapping);
Review Comment:
[P2] Do not leave child caches permanently cold when the base is ineffective
A valid `meta.cache.paimon.table.max-weight=0` makes the base ineffective
while leaving both children enabled by default; `table.enable=false` also
leaves schema enabled and can explicitly re-enable snapshots. Every table
lookup then returns an uncached value with a fresh synthetic generation, so
this check always fails and immediately discards the just-loaded child.
Repeated schema calls reload the table/schema, while snapshot calls repeat
fence discovery and full partition enumeration even though those caches report
enabled. Please either effectively disable/reject dependent children when the
base is ineffective, or give these loads a safe reusable physical identity, and
cover repeated schema/snapshot lookups under these configurations.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/metacache/MetaCacheEntry.java:
##########
@@ -198,62 +542,691 @@ 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;
+ }
+
+ 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()) {
+ int evicted = evictLocalColdest(incomingKey,
LOCAL_EVICTION_BATCH_SIZE);
+ if (evicted == 0) {
+ entryBudget.requestPeerReclaim(bytes);
+ break;
+ }
+ reservation = entryBudget.tryReserve(bytes);
+ }
+ 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) {
+ int evicted = evictLocalColdest(incomingKey,
LOCAL_EVICTION_BATCH_SIZE);
+ if (evicted == 0) {
+ entryBudget.requestPeerReclaim(Math.max(0L, newBytes -
reservation.getBytes()));
+ return false;
+ }
+ if (reservation.tryResize(newBytes)) {
+ return true;
+ }
+ }
+ }
+
+ private int evictLocalColdest(K incomingKey, int limit) {
+ 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++;
+ }
+ }
+ 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) > 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 RemovedValue<>(key, value));
Review Comment:
[P1] Keep queued removal values inside the memory budget
`pendingRemovalNotifications` now strongly owns the entire removed `V`, but
removals performed under `admissionLock` release that value's reservation
immediately below. `invalidateAll`/`invalidateIf` and local/peer eviction can
therefore admit replacements while the single process-wide cleanup thread still
holds retired Paimon table graphs; each callback also scans snapshot/schema
entries, so remove/refill churn can build an unbounded queue outside
entry/catalog/global accounting. Please either keep the reservation until the
notification drops `V` or enqueue only the generation/token needed by the
listener, and test a blocked cleanup plus repeated invalidate/refill cycle.
--
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]
