924060929 · Oct 23, 2026 · Oct 23, 2026 · Oct 23, 2026 · Oct 23, 2026 · Oct 23, 2026
@@ -151,6 +151,11 @@ public void invalidateCatalog() {
         registry.invalidate(ScopePath.catalog());
     }
 
+    /** Runs a short auxiliary-index update at the same publication boundary as catalog invalidation. */
+    public void invalidateCatalogAndRun(Runnable commitAction) {
+        registry.invalidateAndRunUnderPublicationGate(ScopePath.catalog(), commitAction);
+    }
+
     public void invalidateDatabase(String database) {
         registry.invalidate(ScopePath.database(database));
     }

@@ -18,7 +18,10 @@
 package org.apache.doris.connector.cache;
 
 import java.util.Objects;
+import java.util.Optional;
 import java.util.function.BiConsumer;
+import java.util.function.BiFunction;
+import java.util.function.BooleanSupplier;
 import java.util.function.Consumer;
 import java.util.function.Function;
 
@@ -89,6 +92,34 @@ public boolean compareAndSet(K key, V expectedValue, V updatedValue, Runnable co
                 Objects.requireNonNull(commitAction, "commitAction can not be null"));
     }
 
+    /** Captures an exact retained registration without loading or holding locks across a remote read. */
+    public Optional<Observation> observe(K key, Object expectedValue) {
+        K nonNullKey = Objects.requireNonNull(key, "key");
+        return delegate.observe(nonNullKey, definition.scope(nonNullKey),
+                Objects.requireNonNull(expectedValue, "expectedValue"));
+    }
+
+    /**
+     * A short guarded action may nest observations in parent-to-child order. It must not perform I/O,
+     * re-enter this cache, or invalidate this cache's owner. Returning false leaves this entry unchanged.
+     * Conditional invalidation still removes the observed value if its dependency cleanup throws.
+     */
+    public static final class Observation {
+        private final BiFunction<Boolean, BooleanSupplier, Boolean> operation;
+
+        Observation(BiFunction<Boolean, BooleanSupplier, Boolean> operation) {
+            this.operation = operation;
+        }
+
+        public boolean runIfCurrent(BooleanSupplier action) {
+            return operation.apply(false, Objects.requireNonNull(action, "action"));
+        }
+
+        public boolean invalidateIfCurrent(BooleanSupplier action) {
+            return operation.apply(true, Objects.requireNonNull(action, "action"));
+        }
+    }
+
     public void invalidateKey(K key) {
         delegate.invalidateKey(Objects.requireNonNull(key, "key can not be null"));
     }

@@ -0,0 +1,79 @@
+// 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.connector.cache;
+
+import java.util.ArrayDeque;
+import java.util.Deque;
+
+/** Defers physical removal and resource release until an enclosing guarded cache mutation has finished. */
+public final class MetaCacheCleanup {
+    private static final ThreadLocal<Deque<Runnable>> PENDING = new ThreadLocal<>();
+
+    private MetaCacheCleanup() {
+    }
+
+    /**
+     * Logical invalidation still completes inside the action. Detached registrations and removal callbacks
+     * are cleaned synchronously after the outermost action returns, including when that action fails.
+     */
+    public static void runDeferred(Runnable action) {
+        if (PENDING.get() != null) {
+            action.run();
+            return;
+        }
+        Deque<Runnable> pending = new ArrayDeque<>();
+        PENDING.set(pending);
+        Throwable failure = null;
+        try {
+            action.run();
+        } catch (RuntimeException | Error e) {
+            failure = e;
+        }
+        try {
+            Runnable cleanup;
+            while ((cleanup = pending.pollFirst()) != null) {
+                try {
+                    cleanup.run();
+                } catch (RuntimeException | Error e) {
+                    if (failure == null) {
+                        failure = e;
+                    } else if (failure != e) {
+                        failure.addSuppressed(e);
+                    }
+                }
+            }
+        } finally {
+            PENDING.remove();
+        }
+        if (failure instanceof RuntimeException) {
+            throw (RuntimeException) failure;
+        }
+        if (failure != null) {
+            throw (Error) failure;
+        }
+    }
+
+    static void execute(Runnable cleanup) {
+        Deque<Runnable> pending = PENDING.get();
+        if (pending == null) {
+            cleanup.run();
+        } else {
+            pending.addLast(cleanup);
+        }
+    }
+}
@@ -270,6 +270,44 @@ public V getWithPublicationAction(K key, ScopePath path, Function<K, V> loader,
         }
     }
 
+    Optional<MetaCache.Observation> observe(K key, ScopePath path, Object expectedValue) {
+        VersionedValue<K, V> value = currentVersionedValue(key, path);
+        if (value == null || value.value != expectedValue) {
+            return Optional.empty();
+        }
+        ScopedMetaCacheRegistry.ScopeObservation scope = registry.observe(value.scopeSnapshot);
+        return Optional.of(new MetaCache.Observation((invalidate, action) -> deferRemovals(() -> {
+            BooleanSupplier commit = () -> scope.runIfCurrent(() -> {
+                synchronized (value.keyNode) {
+                    if (closed.get() || data.getIfPresent(key) != value || !value.isCurrent(registry, keyNodes)) {
+                        return false;
+                    }
+                    boolean invalidateValue = invalidate;
+                    try {
+                        if (!action.getAsBoolean()) {
+                            invalidateValue = false;
+                            return false;
+                        }
+                        return true;
+                    } finally {
+                        // Once authorized, a throwing dependency cleanup must not retain this stale value.
+                        if (invalidateValue) {
+                            exactInvalidationSequence = exactInvalidationSequence.add(BigInteger.ONE);
+                            if (!activeBulkStarts.isEmpty()) {
+                                exactInvalidations.put(key, exactInvalidationSequence);
+                            }
+                            replaceKeyState(value.keyNode);
+                            value.keyNode.loadPublicationState.set(new Object());
+                            removeCurrentVersion(key, value);
+                        }
+                    }
+                }
+            });
+            return invalidate ? bulkInvalidationGate.write(commit::getAsBoolean)
+                    : bulkInvalidationGate.readBoolean(commit);
+        })));
+    }
+
     public V getIfPresent(K key, ScopePath path) {
         Objects.requireNonNull(key, "key can not be null");
         Objects.requireNonNull(path, "path can not be null");
@@ -1118,7 +1156,9 @@ private void drainDeferredRemovals(RemovalDeferral<K, V> removalDeferral) {
             DeferredRemoval<K, V> removal;
             while ((removal = removalDeferral.removals.pollFirst()) != null) {
                 try {
-                    completeRemoval(removal.versioned, removal.cause, removal.budgetEviction);
+                    DeferredRemoval<K, V> detached = removal;
+                    MetaCacheCleanup.execute(() ->
+                            completeRemoval(detached.versioned, detached.cause, detached.budgetEviction));
                 } catch (RuntimeException | Error e) {
                     if (failure == null) {
                         failure = e;

@@ -212,6 +212,32 @@ void invalidate(ScopePath path, Runnable afterStateReplacement) {
         tryPrune(invalidated.existing);
     }
 
+    /** Keeps a short auxiliary-index mutation ordered with value publication, then cleans values outside the gate. */
+    void invalidateAndRunUnderPublicationGate(ScopePath path, Runnable commitAction) {
+        Objects.requireNonNull(path, "path can not be null");
+        Runnable action = Objects.requireNonNull(commitAction, "commitAction can not be null");
+        checkOpen();
+        AtomicReference<InvalidatedScope> invalidated = new AtomicReference<>();
+        try {
+            publicationGate.write(() -> {
+                List<ScopeNode> existing = resolveExisting(path);
+                bumpPublicationStates(existing);
+                if (existing.size() == path.level().ordinal() + 1) {
+                    ScopeNode target = existing.get(existing.size() - 1);
+                    invalidated.set(new InvalidatedScope(existing, replaceState(target)));
+                }
+                action.run();
+                return null;
+            });
+        } finally {
+            InvalidatedScope scope = invalidated.get();
+            if (scope != null) {
+                cleanDetachedState(scope.oldState);
+                tryPrune(scope.existing);
+            }
+        }
+    }
+
     void invalidate(List<ScopePath> paths, Runnable afterStateReplacement) {
         Objects.requireNonNull(paths, "paths can not be null");
         Objects.requireNonNull(afterStateReplacement, "afterStateReplacement can not be null");
@@ -242,6 +268,26 @@ void invalidate(List<ScopePath> paths, Runnable afterStateReplacement) {
         invalidated.forEach(scope -> tryPrune(scope.existing));
     }
 
+    ScopeObservation observe(ScopeSnapshot snapshot) {
+        return new ScopeObservation(snapshot);
+    }
+
+    final class ScopeObservation {
+        private final ScopeSnapshot snapshot;
+        private final PublicationState publicationState;
+
+        private ScopeObservation(ScopeSnapshot snapshot) {
+            this.snapshot = snapshot;
+            this.publicationState = snapshot.leafNode().descendantPublication.get();
+        }
+
+        boolean runIfCurrent(BooleanSupplier action) {
+            return publicationGate.readBoolean(() -> snapshot.isCurrent(ScopedMetaCacheRegistry.this)
+                    && snapshot.leafNode().descendantPublication.get() == publicationState
+                    && action.getAsBoolean());
+        }
+    }
+
     public ScopeMetrics metrics() {
         ScopeMetricsAccumulator accumulator = new ScopeMetricsAccumulator();
         collectMetrics(root, accumulator);
@@ -433,12 +479,16 @@ private void bumpPublicationStates(List<ScopeNode> nodes) {
     }
 
     private void cleanDetachedState(ScopeState state) {
+        MetaCacheCleanup.execute(() -> cleanDetachedStateNow(state));
+    }
+
+    private void cleanDetachedStateNow(ScopeState state) {
         state.entries.forEach((address, ownershipToken) -> {
             if (state.entries.remove(address, ownershipToken)) {
                 address.removeExpected(ownershipToken);
             }
         });
-        state.children.values().forEach(child -> cleanDetachedState(child.current.get()));
+        state.children.values().forEach(child -> cleanDetachedStateNow(child.current.get()));
         state.children.clear();
     }