[fix](fe) Publish metadata IDs with cache objects

### What problem does this PR solve?

Issue Number: close #64159

Related PR: #64160

Problem Summary: Incremental metadata events published the ID-to-name mapping before the matching database or table object was committed to the shared cache. A concurrent lookup by ID could therefore load and publish a different object in that gap, which was then replaced and cleaned up by the event update. Publish the ID mapping and object inside one guarded cache commit, fence overlapping loads, and make the object-and-ID step run only once when the outer name-cache CAS retries so a newer object is never replaced by replaying an already completed event update.

### Release note

None

### Check List (For Author)

- Test: Unit Test
    - ScopedMetaCacheConcurrencyTest
    - FeMetaCacheEntryTest
    - ExternalCatalogTest
    - ExternalDatabaseTest
    - ./build.sh --fe
- Behavior changed: No
- Does this need documentation: No

Author: 924060929

fe/fe-connector/fe-connector-cache/src/main/java/org/apache/doris/connector/cache/MetaCache.java

@@ -72,8 +72,21 @@ public void put(K key, V value) {
     }
 
     public boolean compareAndSet(K key, V expectedValue, V updatedValue) {
+        return compareAndSet(key, expectedValue, updatedValue, () -> {
+        });
+    }
+
+    /**
+     * Updates a value only if it is still current, and runs an auxiliary action inside the same guarded commit.
+     * Concurrent cache loads can not publish between the action and the value update.
+     * The action runs while the publication guards and key lock are held, so it must be short, non-blocking, and
+     * must not re-enter this cache. If it throws, the value is not updated, but completed external side effects are
+     * not rolled back.
+     */
+    public boolean compareAndSet(K key, V expectedValue, V updatedValue, Runnable commitAction) {
         K nonNullKey = Objects.requireNonNull(key, "key can not be null");
-        return delegate.compareAndSet(nonNullKey, definition.scope(nonNullKey), expectedValue, updatedValue);
+        return delegate.compareAndSet(nonNullKey, definition.scope(nonNullKey), expectedValue, updatedValue,
+                Objects.requireNonNull(commitAction, "commitAction can not be null"));
     }
 
     public void invalidateKey(K key) {

fe/fe-connector/fe-connector-cache/src/main/java/org/apache/doris/connector/cache/ScopedMetaCache.java

@@ -292,10 +292,17 @@ public void put(K key, ScopePath path, V value) {
     }
 
     public boolean compareAndSet(K key, ScopePath path, V expectedValue, V updatedValue) {
+        return compareAndSet(key, path, expectedValue, updatedValue, NO_OP);
+    }
+
+    public boolean compareAndSet(
+            K key, ScopePath path, V expectedValue, V updatedValue, Runnable commitAction) {
         Objects.requireNonNull(key, "key can not be null");
         Objects.requireNonNull(path, "path can not be null");
+        Runnable action = Objects.requireNonNull(commitAction, "commitAction can not be null");
         checkOpen();
         if (!effectiveEnabled) {
+            action.run();
             return true;
         }
         try (PublicationLease<K, V> lease = acquirePublicationLease(key, path, false)) {
@@ -306,7 +313,9 @@ public boolean compareAndSet(K key, ScopePath path, V expectedValue, V updatedVa
                     return false;
                 }
                 lease.keyNode.loadPublicationState.set(new Object());
+                action.run();
                 if (updatedValue == currentValue) {
+                    lease.keyNode.loadPublicationState.set(new Object());
                     return true;
                 }
                 if (updatedValue == null) {
@@ -317,6 +326,7 @@ public boolean compareAndSet(K key, ScopePath path, V expectedValue, V updatedVa
                 } else {
                     publishCommitted(lease, key, updatedValue);
                 }
+                lease.keyNode.loadPublicationState.set(new Object());
                 return true;
             });
         }

fe/fe-connector/fe-connector-cache/src/test/java/org/apache/doris/connector/cache/ScopedMetaCacheConcurrencyTest.java

@@ -71,6 +71,37 @@ public void concurrentMissesForTheSameKeyRunOneLoader() throws Exception {
         }
     }
 
+    @Test
+    public void commitActionPreventsConcurrentMissFromPublishingBetweenIndexAndValue() throws Exception {
+        ExecutorService executor = Executors.newFixedThreadPool(2);
+        CountDownLatch actionStarted = new CountDownLatch(1);
+        CountDownLatch concurrentLoadElected = new CountDownLatch(1);
+        AtomicInteger loads = new AtomicInteger();
+        try (ScopedMetaCacheRegistry registry = new ScopedMetaCacheRegistry()) {
+            ScopedMetaCache<String, String> cache = registry.createCache(
+                    "test", ENABLED, null, null, concurrentLoadElected::countDown, () -> {
+                    });
+
+            Future<Boolean> update = executor.submit(() -> cache.compareAndSet(
+                    "key", TABLE, null, "event-value", () -> {
+                        actionStarted.countDown();
+                        await(concurrentLoadElected);
+                    }));
+            await(actionStarted);
+            Future<String> lookup = executor.submit(() -> cache.get("key", TABLE, ignored -> {
+                loads.incrementAndGet();
+                return "lookup-value";
+            }));
+
+            Assertions.assertTrue(update.get(TIMEOUT_SECONDS, TimeUnit.SECONDS));
+            Assertions.assertEquals("event-value", lookup.get(TIMEOUT_SECONDS, TimeUnit.SECONDS));
+            Assertions.assertEquals("event-value", cache.getIfPresent("key", TABLE));
+            Assertions.assertEquals(0, loads.get());
+        } finally {
+            executor.shutdownNow();
+        }
+    }
+
     @Test
     public void closeDuringRefreshAdmissionDoesNotRetainRefreshMarker() throws Exception {
         ExecutorService executor = Executors.newSingleThreadExecutor();

fe/fe-core/src/main/java/org/apache/doris/datasource/ExternalCatalog.java

@@ -80,6 +80,7 @@
 import java.util.Optional;
 import java.util.Set;
 import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.function.Function;
 import java.util.stream.Collectors;
 
@@ -1506,6 +1507,7 @@ private void updateDatabaseCache(String remoteDbName, String localDbName,
         // lightweight lookup index and must always track registered objects so normal by-ID lookup can load on demand.
         // By default, incremental updates keep cold names/object cache entries cold.
         // forceUpdateCacheState is only for paths that intentionally populate those cold entries.
+        AtomicBoolean objectEntryUpdated = new AtomicBoolean();
         databaseNames.computeAfterValidation(
                 "",
                 (ignored, current) -> {
@@ -1517,10 +1519,18 @@ private void updateDatabaseCache(String remoteDbName, String localDbName,
                     // can not publish a stale snapshot after this incremental update.
                     return current == null ? null : current.withName(remoteDbName, localDbName);
                 },
-                () -> databases.computeAfterValidation(
-                        localDbName,
-                        (ignored, current) -> (forceUpdateCacheState || current != null) ? db : null,
-                        () -> dbIdNameIndex.put(dbId, localDbName)));
+                () -> {
+                    // The outer names CAS may retry after this object step succeeds. Do not replay the object
+                    // ownership change: a later lookup may already have published and returned a newer object.
+                    if (objectEntryUpdated.get()) {
+                        return;
+                    }
+                    databases.computeWithCommitAction(
+                            localDbName,
+                            (ignored, current) -> (forceUpdateCacheState || current != null) ? db : null,
+                            () -> dbIdNameIndex.put(dbId, localDbName));
+                    objectEntryUpdated.set(true);
+                });
     }
 
     protected void invalidateDatabaseCache(String localDbName) {

fe/fe-core/src/main/java/org/apache/doris/datasource/ExternalDatabase.java

@@ -59,6 +59,7 @@
 import java.util.Optional;
 import java.util.Set;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.function.Function;
 import java.util.stream.Collectors;
 
@@ -701,6 +702,7 @@ protected void updateTableCache(T table, String remoteTableName, String localTab
         // lightweight lookup index and must always track registered objects so normal by-ID lookup can load on demand.
         // By default, incremental updates keep cold names/object cache entries cold.
         // forceUpdateCacheState is only for paths that intentionally populate those cold entries.
+        AtomicBoolean objectEntryUpdated = new AtomicBoolean();
         tableNames.computeAfterValidation(
                 "",
                 (ignored, current) -> {
@@ -712,10 +714,18 @@ protected void updateTableCache(T table, String remoteTableName, String localTab
                     // can not publish a stale snapshot after this incremental update.
                     return current == null ? null : current.withName(remoteTableName, localTableName);
                 },
-                () -> tables.computeAfterValidation(
-                        localTableName,
-                        (ignored, current) -> (forceUpdateCacheState || current != null) ? table : null,
-                        () -> tableIdNameIndex.put(table.getId(), localTableName)));
+                () -> {
+                    // The outer names CAS may retry after this object step succeeds. Do not replay the object
+                    // ownership change: a later lookup may already have published a newer object.
+                    if (objectEntryUpdated.get()) {
+                        return;
+                    }
+                    tables.computeWithCommitAction(
+                            localTableName,
+                            (ignored, current) -> (forceUpdateCacheState || current != null) ? table : null,
+                            () -> tableIdNameIndex.put(table.getId(), localTableName));
+                    objectEntryUpdated.set(true);
+                });
     }
 
     protected void invalidateTableCache(String localTableName) {

fe/fe-core/src/main/java/org/apache/doris/datasource/metacache/FeMetaCacheEntry.java

@@ -320,6 +320,29 @@ public V computeAfterValidation(K key, BiFunction<K, V, V> remappingFunction, Ru
         }
     }
 
+    /**
+     * Runs the action only after the expected cache value is confirmed, while concurrent loads are prevented from
+     * publishing until both the action and the value update finish. The action runs while this entry's stripe and
+     * the shared-cache publication locks are held, so it must be short, non-blocking, and must not re-enter this
+     * entry. If it throws, the value is not updated, but completed external side effects are not rolled back.
+     */
+    public V computeWithCommitAction(K key, BiFunction<K, V, V> remappingFunction, Runnable commitAction) {
+        BiFunction<K, V, V> remapper = Objects.requireNonNull(remappingFunction, "remappingFunction can not be null");
+        Runnable action = Objects.requireNonNull(commitAction, "commitAction can not be null");
+        StripeState<K> stripe = stripeState(key);
+        synchronized (stripe) {
+            while (true) {
+                V current = data.getIfPresent(key);
+                V updated = effectiveEnabled ? remapper.apply(key, current) : null;
+                bumpAction(stripe, key);
+                beforePublicMutationWriteForTest(key);
+                if (data.compareAndSet(key, current, updated, action)) {
+                    return updated;
+                }
+            }
+        }
+    }
+
     public void invalidateKey(K key) {
         invalidateKeyAndRun(key, () -> {
         });