branch-3.0: [fix](catalog) do cache load when cache value is not present #50188 (#50450)

Cherry-picked from #50188

Co-authored-by: Mingyu Chen (Rayner) <morningman@163.com>

Author: github-actions[bot]

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

@@ -23,6 +23,7 @@
 import com.github.benmanes.caffeine.cache.CacheLoader;
 import com.github.benmanes.caffeine.cache.LoadingCache;
 import com.github.benmanes.caffeine.cache.RemovalListener;
+import com.google.common.annotations.VisibleForTesting;
 import com.google.common.collect.Lists;
 import com.google.common.collect.Maps;
 
@@ -88,8 +89,13 @@ public String getRemoteName(String localName) {
 
     public Optional<T> getMetaObj(String name, long id) {
         Optional<T> val = metaObjCache.getIfPresent(name);
-        if (val == null) {
+        if (val == null || !val.isPresent()) {
             synchronized (metaObjCache) {
+                val = metaObjCache.getIfPresent(name);
+                if (val != null && val.isPresent()) {
+                    return val;
+                }
+                metaObjCache.invalidate(name);
                 val = metaObjCache.get(name);
                 idToName.put(id, name);
             }
@@ -133,4 +139,9 @@ public void invalidateAll() {
         metaObjCache.invalidateAll();
         idToName.clear();
     }
+
+    @VisibleForTesting
+    public LoadingCache<String, Optional<T>> getMetaObjCache() {
+        return metaObjCache;
+    }
 }

fe/fe-core/src/test/java/org/apache/doris/datasource/MetaCacheTest.java

@@ -34,6 +34,7 @@
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
 
 public class MetaCacheTest {
 
@@ -216,4 +217,160 @@ public void testMetaObjCacheLoader() throws InterruptedException {
         latch.await();
 
     }
+
+    @Test
+    public void testGetMetaObjCacheLoading() throws InterruptedException {
+        // Create a CountDownLatch to track cache loading invocations
+        CountDownLatch loadLatch = new CountDownLatch(2);
+
+        // Create a custom cache loader that counts invocations
+        CacheLoader<String, Optional<String>> metaObjCacheLoader = key -> {
+            loadLatch.countDown();
+            return Optional.of("loaded_" + key);
+        };
+
+        // Create a new MetaCache instance with our custom loader
+        MetaCache<String> testCache = new MetaCache<>(
+                "testCache",
+                Executors.newCachedThreadPool(),
+                OptionalLong.of(1),
+                OptionalLong.of(1),
+                100,
+                key -> Lists.newArrayList(),
+                metaObjCacheLoader,
+                (key, value, cause) -> {
+                }
+        );
+
+        // Case 1: Test when key does not exist in cache (val == null)
+        Optional<String> result1 = testCache.getMetaObj("non_existent_key", 1L);
+        Assert.assertTrue(result1.isPresent());
+        Assert.assertEquals("loaded_non_existent_key", result1.get());
+
+        // Case 2: Test when key exists but value is empty Optional
+        // First, manually put an empty Optional into cache
+        testCache.getMetaObjCache().put("empty_key", Optional.empty());
+        Optional<String> result2 = testCache.getMetaObj("empty_key", 2L);
+        Assert.assertTrue(result2.isPresent());
+        Assert.assertEquals("loaded_empty_key", result2.get());
+
+        // Verify that cache loader was invoked exactly twice
+        Assert.assertTrue(loadLatch.await(1, TimeUnit.SECONDS));
+    }
+
+    @Test
+    public void testGetMetaObjConcurrent() throws InterruptedException {
+        // Create a CountDownLatch to track cache loading invocations
+        CountDownLatch loadLatch = new CountDownLatch(1);
+        AtomicInteger loadCount = new AtomicInteger(0);
+
+        // Create a custom cache loader that counts invocations and simulates slow loading
+        CacheLoader<String, Optional<String>> metaObjCacheLoader = key -> {
+            loadCount.incrementAndGet();
+            Thread.sleep(100); // Simulate slow loading
+            loadLatch.countDown();
+            return Optional.of("loaded_" + key);
+        };
+
+        // Create a new MetaCache instance with our custom loader
+        MetaCache<String> testCache = new MetaCache<>(
+                "testCache",
+                Executors.newCachedThreadPool(),
+                OptionalLong.of(1),
+                OptionalLong.of(1),
+                100,
+                key -> Lists.newArrayList(),
+                metaObjCacheLoader,
+                (key, value, cause) -> {
+                }
+        );
+
+        // Test concurrent access to non-existent key
+        ExecutorService executor = Executors.newFixedThreadPool(10);
+        final CountDownLatch startLatch = new CountDownLatch(1);
+        final CountDownLatch finishLatch = new CountDownLatch(10);
+
+        for (int i = 0; i < 10; i++) {
+            executor.submit(() -> {
+                try {
+                    startLatch.await();
+                    Optional<String> result = testCache.getMetaObj("concurrent_key", 1L);
+                    Assert.assertTrue(result.isPresent());
+                    Assert.assertEquals("loaded_concurrent_key", result.get());
+                } catch (Exception e) {
+                    Assert.fail("Exception occurred: " + e.getMessage());
+                } finally {
+                    finishLatch.countDown();
+                }
+            });
+        }
+
+        // Start all threads
+        startLatch.countDown();
+        // Wait for all threads to complete
+        finishLatch.await(5, TimeUnit.SECONDS);
+        // Wait for cache loading to complete
+        loadLatch.await(5, TimeUnit.SECONDS);
+
+        // Verify that cache loader was invoked exactly once
+        Assert.assertEquals(1, loadCount.get());
+
+        // Test concurrent access to existing but empty key
+        loadCount.set(0);
+        CountDownLatch loadLatch2 = new CountDownLatch(1);
+        CacheLoader<String, Optional<String>> metaObjCacheLoader2 = key -> {
+            loadCount.incrementAndGet();
+            Thread.sleep(100); // Simulate slow loading
+            loadLatch2.countDown();
+            return Optional.of("loaded_" + key);
+        };
+
+        // Create another MetaCache instance
+        MetaCache<String> testCache2 = new MetaCache<>(
+                "testCache2",
+                Executors.newCachedThreadPool(),
+                OptionalLong.of(1),
+                OptionalLong.of(1),
+                100,
+                key -> Lists.newArrayList(),
+                metaObjCacheLoader2,
+                (key, value, cause) -> {
+                }
+        );
+
+        // Manually put an empty Optional into cache
+        testCache2.getMetaObjCache().put("empty_concurrent_key", Optional.empty());
+
+        // Reset latches for second test
+        final CountDownLatch startLatch2 = new CountDownLatch(1);
+        final CountDownLatch finishLatch2 = new CountDownLatch(10);
+
+        for (int i = 0; i < 10; i++) {
+            executor.submit(() -> {
+                try {
+                    startLatch2.await();
+                    Optional<String> result = testCache2.getMetaObj("empty_concurrent_key", 2L);
+                    Assert.assertTrue(result.isPresent());
+                    Assert.assertEquals("loaded_empty_concurrent_key", result.get());
+                } catch (Exception e) {
+                    Assert.fail("Exception occurred: " + e.getMessage());
+                } finally {
+                    finishLatch2.countDown();
+                }
+            });
+        }
+
+        // Start all threads
+        startLatch2.countDown();
+        // Wait for all threads to complete
+        finishLatch2.await(5, TimeUnit.SECONDS);
+        // Wait for cache loading to complete
+        loadLatch2.await(5, TimeUnit.SECONDS);
+
+        // Verify that cache loader was invoked exactly once
+        Assert.assertEquals(1, loadCount.get());
+
+        executor.shutdown();
+        executor.awaitTermination(1, TimeUnit.SECONDS);
+    }
 }

regression-test/data/external_table_p0/hive/test_hive_use_meta_cache.out

@@ -27,6 +27,8 @@ test_use_meta_cache_db_hive
 -- !sql07 --
 test_use_meta_cache_tbl_hive
 
+-- !aother_test_sql --
+
 -- !sql08 --
 
 -- !sql09 --
@@ -118,6 +120,8 @@ test_use_meta_cache_db_hive
 -- !sql07 --
 test_use_meta_cache_tbl_hive
 
+-- !aother_test_sql --
+
 -- !sql08 --
 
 -- !sql09 --

regression-test/suites/external_table_p0/hive/test_hive_use_meta_cache.groovy

@@ -49,10 +49,10 @@ suite("test_hive_use_meta_cache", "p0,external,hive,external_docker,external_doc
                 String partitioned_table_hive = "test_use_meta_cache_partitioned_tbl_hive"
 
                 sql "switch ${catalog}"
-                sql "drop database if exists ${database}"
-                sql "drop database if exists ${database_hive}"
+                sql "drop database if exists ${database} force"
+                sql "drop database if exists ${database_hive} force"
                 order_qt_sql01 "show databases like '%${database}%'";
-                sql "drop database if exists ${database}"
+                sql "drop database if exists ${database} force"
                 sql "create database ${database}"
                 order_qt_sql02 "show databases like '%${database}%'";
                 sql "use ${database}"
@@ -100,6 +100,24 @@ suite("test_hive_use_meta_cache", "p0,external,hive,external_docker,external_doc
                 }
                 // can see
                 order_qt_sql07 "show tables"
+
+                // another table creation test only for use_meta_cache=true
+                // the main point is to select the table first before creation.
+                if (use_meta_cache) {
+                    // 0. create env
+                    hive_docker "drop table if exists ${database_hive}.another_table_creation_test"
+                    // 1. select a non exist table
+                    test {
+                        sql "select * from another_table_creation_test";
+                        exception "does not exist in database"
+                    }
+                    // 2. use hive to create this table
+                    hive_docker "create table ${database_hive}.another_table_creation_test (k1 int)"
+                    // 3. use doris to select, can see
+                    qt_aother_test_sql "select * from another_table_creation_test";
+                    // 4. drop table
+                    sql "drop table another_table_creation_test";
+                }
 
                 // test Hive Metastore table partition file listing
                 hive_docker "create table ${database_hive}.${partitioned_table_hive} (k1 int) partitioned by (p1 string)"