CalvinKirs commented on code in PR #66717:
URL: https://github.com/apache/doris/pull/66717#discussion_r3836927125
##########
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);
+ if (ownerGeneration >= 0L) {
+ beforeRemovalReleaseForTest(key);
+ if (closed.get()) {
+ return;
+ }
+ if (cause.wasEvicted()) {
+ pendingEvictionGenerations.merge(key, ownerGeneration,
Math::max);
+ }
+ pendingRemovalGenerations.merge(key, ownerGeneration, Math::max);
+ if (closed.get()) {
+ pendingRemovalGenerations.remove(key, ownerGeneration);
+ pendingEvictionGenerations.remove(key, ownerGeneration);
+ return;
+ }
+ scheduleRemovalCleanup();
+ }
+ }
+
+ private void scheduleRemovalCleanup() {
+ if (closed.get()) {
+ return;
+ }
+ if (removalCleanupScheduled.compareAndSet(false, true)) {
+ try {
+ REMOVAL_CLEANUP_EXECUTOR.execute(this::drainRemovalCleanups);
+ } catch (RejectedExecutionException e) {
+ removalCleanupScheduled.set(false);
+ LOG.warn("Failed to schedule removal cleanup for external
metadata cache entry {}", name, e);
+ }
+ }
+ }
+
+ private void drainRemovalCleanups() {
+ try {
+ drainRemovalNotifications();
Review Comment:
Fixed in bfca15ad5d4: drainRemovalCleanups now releases dead reservations
before running dependency-retirement listeners, and the removal callback queues
the reservation before the notification so the worker can never observe the
notification without the matching quota release. Regression
testDeadReservationsReleaseBeforeDependencyRetirement holds a listener on a
latch after an automatic eviction and asserts the global quota is already
released while the listener is still blocked.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/PaimonExternalMetaCache.java:
##########
@@ -86,41 +114,315 @@ public Table getPaimonTable(NameMapping nameMapping) {
public PaimonSnapshotCacheValue getSnapshotCache(ExternalTable dorisTable)
{
NameMapping nameMapping = dorisTable.getOrBuildNameMapping();
- return
tableEntry.get(nameMapping.getCtlId()).get(nameMapping).getLatestSnapshotCacheValue();
+ MetaCacheEntry<NameMapping, PaimonTableCacheValue> tables =
tableEntry.get(nameMapping.getCtlId());
+ PaimonTableCacheValue tableValue = tables.get(nameMapping);
+ if (tables.isEffectivelyEnabled()) {
+ // Serve the memoized latest projection of this table generation
while it is still
+ // published: the latest read is as stale-until-TTL/refresh as the
cached table
+ // handle itself and costs no snapshot IO, preserving the
pre-existing external
+ // metadata cache contract. The fence is re-observed only when no
projection of this
+ // generation is reachable anymore (first read, expiry, weight
eviction, explicit
+ // invalidation), which is also when rollback ordering below
matters.
+ ObservedFence observed = latestObservedFences.get(
+ new LatestFenceOwner(nameMapping,
tableValue.getGeneration()));
+ if (observed != null) {
+ PaimonSnapshotCacheValue memoized =
+
snapshotEntry.get(nameMapping.getCtlId()).peekIfPresent(observed.key);
+ if (memoized != null) {
+ return memoized;
+ }
+ }
+ }
+ PaimonSnapshot fence = loadLatestSnapshotFence(nameMapping,
tableValue).getSnapshot();
+ if (!tables.isEffectivelyEnabled()) {
+ // Projections are keyed by the synthetic generation of a
published table handle. An
+ // ineffective table entry publishes nothing, so nothing keyed by
this load could ever
+ // be looked up again: serve it directly instead of churning the
snapshot entry.
+ return executeForGeneration(tableValue, nameMapping,
+ () -> latestSnapshotProjectionLoader.loadAtFence(
+ nameMapping, fence, tableValue.getGeneration()))
+ .bindCapturedAuthenticator(tableValue.getAuthenticator());
+ }
+ // Order fence observations, not snapshot ids: a rollback moves the
latest snapshot
+ // backwards, and a concurrent call may finish after a later
observation (reversed
+ // completion). Either way the most recently observed fence is the one
future lookups read.
+ long observation = fenceObservations.incrementAndGet();
Review Comment:
Fixed in bfca15ad5d4: fence capture and observation-number assignment are
serialized per owner (fenceCaptureLocks, entries retired together with their
owners across generation retirement, unpublished-generation cleanup and catalog
purges), so the observation order always matches the fence-read order and a
capture pausing between read and increment can no longer replace a newer
published fence with an older one. Regression
testFenceCaptureAndObservationAssignmentAreSerializedPerOwner blocks a capture
inside the fence read (before the counter), races a newer capture, and asserts
serialization plus the newest-read fence owning the memoized latest.
--
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]
