diff --git a/fe/fe-core/src/main/java/org/apache/doris/catalog/RefreshManager.java b/fe/fe-core/src/main/java/org/apache/doris/catalog/RefreshManager.java index 86b664eaf36078..836a86b6c4d430 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/catalog/RefreshManager.java +++ b/fe/fe-core/src/main/java/org/apache/doris/catalog/RefreshManager.java @@ -109,6 +109,10 @@ public void replayRefreshDb(ExternalObjectLog log) { } if (!db.isPresent()) { + // The database object cache can be cold while row-count entries from an earlier + // generation are still resident. Retire the catalog scope because replay cannot + // recover a canonical database id without loading remote metadata. + Env.getCurrentEnv().getExtMetaCacheMgr().invalidateRowCountCache(catalog.getId()); LOG.warn("failed to find db when replaying refresh db: {}", log.debugForRefreshDb()); } else { refreshDbInternal(db.get()); @@ -168,6 +172,7 @@ public void replayRefreshTable(ExternalObjectLog log) { } // See comment in refreshDbInternal for why db and table may be null. if (!db.isPresent()) { + Env.getCurrentEnv().getExtMetaCacheMgr().invalidateRowCountCache(catalog.getId()); LOG.warn("failed to find db when replaying refresh table: {}", log.debugForRefreshTable()); return; } @@ -178,6 +183,7 @@ public void replayRefreshTable(ExternalObjectLog log) { table = db.get().getTableForReplay(log.getTableId()); } if (!table.isPresent()) { + Env.getCurrentEnv().getExtMetaCacheMgr().invalidateRowCountCache(catalog.getId(), db.get().getId()); LOG.warn("failed to find table when replaying refresh table: {}", log.debugForRefreshTable()); return; } @@ -195,6 +201,7 @@ public void replayRefreshTable(ExternalObjectLog log) { HiveExternalMetaCache cache = Env.getCurrentEnv().getExtMetaCacheMgr() .hive(catalog.getId()); cache.refreshAffectedPartitionsCache((HMSExternalTable) table.get(), modifiedPartNames, newPartNames); + Env.getCurrentEnv().getExtMetaCacheMgr().invalidateRowCountCache(table.get()); if (table.get() instanceof HMSExternalTable && log.getLastUpdateTime() > 0) { ((HMSExternalTable) table.get()).setUpdateTime(log.getLastUpdateTime()); } @@ -281,6 +288,7 @@ public void refreshPartitions(String catalogName, String dbName, String tableNam for (String partitionName : partitionNames) { cache.invalidatePartitionCache(externalTable, partitionName); } + Env.getCurrentEnv().getExtMetaCacheMgr().invalidateRowCountCache(externalTable); ((HMSExternalTable) table).setUpdateTime(updateTime); } diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/CatalogMgr.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/CatalogMgr.java index c83cb64e34847c..86a92cbdc6e6ee 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/datasource/CatalogMgr.java +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/CatalogMgr.java @@ -1041,6 +1041,10 @@ public void addExternalPartitions(String catalogName, String dbName, String tabl } HMSExternalTable hmsTable = (HMSExternalTable) table; + // The metastore mutation has already committed when this event is handled. Fence the + // independent row-count cache even when the local partition cache cannot represent the + // table and this method returns early. + Env.getCurrentEnv().getExtMetaCacheMgr().invalidateRowCountCache(hmsTable); List partitionColumnTypes; try { partitionColumnTypes = hmsTable.getPartitionColumnTypes(MvccUtil.getSnapshotFromContext(hmsTable)); @@ -1082,6 +1086,7 @@ public void dropExternalPartitions(String catalogName, String dbName, String tab HMSExternalTable hmsTable = (HMSExternalTable) table; Env.getCurrentEnv().getExtMetaCacheMgr().hive(catalog.getId()) .dropPartitionsCache(hmsTable, partitionNames, true); + Env.getCurrentEnv().getExtMetaCacheMgr().invalidateRowCountCache(hmsTable); hmsTable.setUpdateTime(updateTime); } diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/ExternalCatalog.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/ExternalCatalog.java index 2c31414a99fbdf..dab252cd1a47c6 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/datasource/ExternalCatalog.java +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/ExternalCatalog.java @@ -182,6 +182,7 @@ public abstract class ExternalCatalog protected ExternalMetadataOps metadataOps; protected TransactionManager transactionManager; protected MetaCache> metaCache; + private volatile boolean invalidatingAllMetaCache; protected ExecutionAuthenticator executionAuthenticator; protected ThreadPoolExecutor threadPoolWithPreAuth; // Map lowercase database names to actual remote database names for case-insensitive lookup @@ -424,7 +425,8 @@ private void buildMetaCache() { localDbName -> Optional.ofNullable( buildDbForInit(null, localDbName, Util.genIdByName(name, localDbName), logType, true)), - (key, value, cause) -> value.ifPresent(v -> v.resetMetaToUninitialized())); + (key, value, cause) -> value.ifPresent( + v -> v.resetMetaToUninitialized(!invalidatingAllMetaCache))); } } @@ -656,9 +658,15 @@ public void onRefreshCache(boolean invalidCache) { * Refresh meta cache only (database level cache), without invalidating catalog level cache. * This method is safe to call within synchronized block. */ - private void refreshMetaCacheOnly() { + private synchronized void refreshMetaCacheOnly() { if (metaCache != null) { - metaCache.invalidateAll(); + invalidatingAllMetaCache = true; + try { + metaCache.invalidateAll(); + } finally { + invalidatingAllMetaCache = false; + } + Env.getCurrentEnv().getExtMetaCacheMgr().getRowCountCache().invalidateCatalog(id); } } @@ -1163,10 +1171,24 @@ public void unregisterDatabase(String dbName) { if (LOG.isDebugEnabled()) { LOG.debug("unregister database [{}]", dbName); } - if (isInitialized()) { - metaCache.invalidate(dbName, Util.genIdByName(name, dbName)); + // Resolve the canonical database object before removing it from the local metadata cache. + // The row-count cache can outlive that object and must be invalidated by its numeric id. + boolean catalogInitialized = isInitialized(); + Optional> db = catalogInitialized + ? getDbForReplay(dbName) : Optional.empty(); + String localDbName = db.map(ExternalDatabase::getFullName).orElse(dbName); + long dbId = db.map(ExternalDatabase::getId).orElseGet(() -> Util.genIdByName(name, localDbName)); + try { + if (db.isPresent()) { + Env.getCurrentEnv().getExtMetaCacheMgr().invalidateDb(getId(), dbId, localDbName); + } else { + Env.getCurrentEnv().getExtMetaCacheMgr().invalidateDb(getId(), dbName); + } + } finally { + if (catalogInitialized) { + metaCache.invalidate(localDbName, dbId); + } } - Env.getCurrentEnv().getExtMetaCacheMgr().invalidateDb(getId(), dbName); } public void registerDatabase(long dbId, String dbName) { diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/ExternalDatabase.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/ExternalDatabase.java index 940fade40a84c2..41166ce8852e97 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/datasource/ExternalDatabase.java +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/ExternalDatabase.java @@ -117,6 +117,10 @@ public void setRemoteName(String remoteName) { } public void resetMetaToUninitialized() { + resetMetaToUninitialized(true); + } + + void resetMetaToUninitialized(boolean invalidateRowCountCache) { if (LOG.isDebugEnabled()) { LOG.debug("resetToUninitialized db name {}, id {}, isInitializing: {}, initialized: {}", this.name, this.id, isInitializing, initialized, new Exception()); @@ -128,7 +132,9 @@ public void resetMetaToUninitialized() { metaCache.invalidateAll(); } } - Env.getCurrentEnv().getExtMetaCacheMgr().invalidateDb(extCatalog.getId(), getFullName()); + if (invalidateRowCountCache) { + Env.getCurrentEnv().getExtMetaCacheMgr().invalidateDb(extCatalog.getId(), getId(), getFullName()); + } } public boolean isInitialized() { @@ -573,6 +579,9 @@ public void unregisterTable(String tableName) { // check if the table exists in cache, it not, does return ExternalTable dorisTable = getTableForReplay(tableName).orElse(null); if (dorisTable == null) { + // The table object cache is much smaller than the row-count cache. A drop or rename + // must still retire stale row counts when the table object has already been evicted. + Env.getCurrentEnv().getExtMetaCacheMgr().invalidateTable(extCatalog.getId(), getFullName(), tableName); return; } // clear the cache related to this table. diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/ExternalMetaCacheMgr.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/ExternalMetaCacheMgr.java index bb62d178f968a8..df8b5b567d7d3e 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/datasource/ExternalMetaCacheMgr.java +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/ExternalMetaCacheMgr.java @@ -50,6 +50,7 @@ import java.util.Map; import java.util.Objects; import java.util.Optional; +import java.util.OptionalLong; import java.util.Set; import java.util.concurrent.ExecutorService; import java.util.concurrent.locks.Lock; @@ -379,9 +380,13 @@ private Map runtimeEffectiveCacheProperties( } public void invalidateCatalog(long catalogId) { - routeCatalogEngines(catalogId, cache -> safeInvalidate( - cache, catalogId, "invalidateCatalog", - () -> cache.invalidateCatalogEntries(catalogId))); + try { + routeCatalogEngines(catalogId, cache -> safeInvalidate( + cache, catalogId, "invalidateCatalog", + () -> cache.invalidateCatalogEntries(catalogId))); + } finally { + rowCountCache.invalidateCatalog(catalogId); + } } public void invalidateCatalogByEngine(long catalogId, String engine) { @@ -413,6 +418,7 @@ public void removeCatalog(long catalogId) { cache, catalogId, "removeCatalog", () -> cache.invalidateCatalog(catalogId))); } finally { + rowCountCache.invalidateCatalog(catalogId); lifecycleLock.unlock(); } } @@ -452,6 +458,7 @@ public void removeCatalogPermanently(long catalogId) { } } } finally { + rowCountCache.invalidateCatalog(catalogId); lifecycleLock.unlock(); } } @@ -463,10 +470,12 @@ public void rollbackCatalogProperties(ExternalCatalog catalog, Map safeInvalidate( cache, catalogId, "rollbackCatalogProperties", () -> cache.invalidateCatalog(catalogId))); } finally { + rowCountCache.invalidateCatalog(catalogId); lifecycleLock.unlock(); } } @@ -484,27 +493,88 @@ public void removeCatalogByEngine(long catalogId, String engine) { } public void invalidateDb(long catalogId, String dbName) { - routeCatalogEngines(catalogId, cache -> safeInvalidate( - cache, catalogId, "invalidateDb", () -> cache.invalidateDb(catalogId, dbName))); + OptionalLong dbId = getCachedDbId(catalogId, dbName); + invalidateDb(catalogId, dbName, dbId); + } + + public void invalidateDb(long catalogId, long dbId, String dbName) { + invalidateDb(catalogId, dbName, OptionalLong.of(dbId)); + } + + private void invalidateDb(long catalogId, String dbName, OptionalLong dbId) { + try { + routeCatalogEngines(catalogId, cache -> safeInvalidate( + cache, catalogId, "invalidateDb", () -> cache.invalidateDb(catalogId, dbName))); + } finally { + if (dbId.isPresent()) { + rowCountCache.invalidateDb(catalogId, dbId.getAsLong()); + } else { + // The database object cache is smaller than the row-count cache. If the object has + // already been evicted, retire the catalog scope rather than hashing caller spelling. + rowCountCache.invalidateCatalog(catalogId); + } + } } public void invalidateTable(long catalogId, String dbName, String tableName) { - routeCatalogEngines(catalogId, cache -> safeInvalidate( - cache, catalogId, "invalidateTable", - () -> cache.invalidateTable(catalogId, dbName, tableName))); + Optional> db = getCachedDb(catalogId, dbName); + try { + routeCatalogEngines(catalogId, cache -> safeInvalidate( + cache, catalogId, "invalidateTable", + () -> cache.invalidateTable(catalogId, dbName, tableName))); + } finally { + invalidateTableRowCount(catalogId, db, tableName); + } } public void invalidateTableByEngine(long catalogId, String engine, String dbName, String tableName) { - routeSpecifiedEngine(engine, cache -> safeInvalidate( - cache, catalogId, "invalidateTableByEngine", - () -> cache.invalidateTable(catalogId, dbName, tableName))); + Optional> db = getCachedDb(catalogId, dbName); + try { + routeSpecifiedEngine(engine, cache -> safeInvalidate( + cache, catalogId, "invalidateTableByEngine", + () -> cache.invalidateTable(catalogId, dbName, tableName))); + } finally { + invalidateTableRowCount(catalogId, db, tableName); + } } public void invalidatePartitions(long catalogId, String dbName, String tableName, List partitions) { - routeCatalogEngines(catalogId, cache -> safeInvalidate( - cache, catalogId, "invalidatePartitions", - () -> cache.invalidatePartitions(catalogId, dbName, tableName, partitions))); + Optional> db = getCachedDb(catalogId, dbName); + try { + routeCatalogEngines(catalogId, cache -> safeInvalidate( + cache, catalogId, "invalidatePartitions", + () -> cache.invalidatePartitions(catalogId, dbName, tableName, partitions))); + } finally { + invalidateTableRowCount(catalogId, db, tableName); + } + } + + private void invalidateTableRowCount(long catalogId, + Optional> db, String tableName) { + if (db.isPresent()) { + Optional table = db.get().getTableForReplay(tableName); + if (table.isPresent()) { + invalidateRowCountCache(table.get()); + } else { + rowCountCache.invalidateDb(catalogId, db.get().getId()); + } + } else { + rowCountCache.invalidateCatalog(catalogId); + } + } + + private OptionalLong getCachedDbId(long catalogId, String dbName) { + Optional> db = getCachedDb(catalogId, dbName); + return db.isPresent() ? OptionalLong.of(db.get().getId()) : OptionalLong.empty(); + } + + private Optional> getCachedDb(long catalogId, String dbName) { + CatalogIf catalog = getCatalog(catalogId); + if (!(catalog instanceof ExternalCatalog)) { + return Optional.empty(); + } + return ((ExternalCatalog) catalog).getDbForReplay(dbName); } public List getCatalogCacheStats(long catalogId) { @@ -669,15 +739,32 @@ public ExternalRowCountCache getRowCountCache() { } public void invalidateTableCache(ExternalTable dorisTable) { - invalidateTable(dorisTable.getCatalog().getId(), - dorisTable.getDbName(), - dorisTable.getName()); + long catalogId = dorisTable.getCatalog().getId(); + try { + routeCatalogEngines(catalogId, cache -> safeInvalidate( + cache, catalogId, "invalidateTableCache", + () -> cache.invalidateTable(catalogId, dorisTable.getDbName(), dorisTable.getName()))); + } finally { + invalidateRowCountCache(dorisTable); + } if (LOG.isDebugEnabled()) { LOG.debug("invalid table cache for {}.{} in catalog {}", dorisTable.getRemoteDbName(), dorisTable.getRemoteName(), dorisTable.getCatalog().getName()); } } + public void invalidateRowCountCache(ExternalTable table) { + rowCountCache.invalidateTable(table.getCatalog().getId(), table.getDb().getId(), table.getId()); + } + + public void invalidateRowCountCache(long catalogId) { + rowCountCache.invalidateCatalog(catalogId); + } + + public void invalidateRowCountCache(long catalogId, long dbId) { + rowCountCache.invalidateDb(catalogId, dbId); + } + public LegacyMetaCacheFactory legacyMetaCacheFactory() { return legacyMetaCacheFactory; } @@ -698,6 +785,10 @@ void replaceEngineCachesForTest(List caches) { bindCatalogPreparers(); } + void replaceRowCountCacheForTest(ExternalRowCountCache cache) { + rowCountCache = cache; + } + /** * Fallback implementation of {@link AbstractExternalMetaCache} for engines that do not * provide dedicated cache entries. diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/ExternalRowCountCache.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/ExternalRowCountCache.java index f32ba5ae20cbc4..e68361139aa60d 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/datasource/ExternalRowCountCache.java +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/ExternalRowCountCache.java @@ -24,22 +24,42 @@ import org.apache.doris.statistics.BasicAsyncCacheLoader; import org.apache.doris.statistics.util.StatisticsUtil; +import com.github.benmanes.caffeine.cache.AsyncCacheLoader; import com.github.benmanes.caffeine.cache.AsyncLoadingCache; +import com.github.benmanes.caffeine.cache.Ticker; import lombok.Getter; import org.apache.logging.log4j.LogManager; import org.apache.logging.log4j.Logger; +import java.util.Objects; import java.util.Optional; import java.util.OptionalLong; +import java.util.Set; import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.Executor; import java.util.concurrent.ExecutorService; +import java.util.concurrent.locks.ReentrantReadWriteLock; +import java.util.function.Supplier; public class ExternalRowCountCache { private static final Logger LOG = LogManager.getLogger(ExternalRowCountCache.class); private final AsyncLoadingCache> rowCountCache; + private final ConcurrentHashMap> inFlightLoads = new ConcurrentHashMap<>(); + // Serialize future publication with explicit invalidation. Invalidation marks matching in-flight loads + // before removing their cache entries, so a refresh that finishes later cannot republish stale data. + private final ReentrantReadWriteLock publicationLock = new ReentrantReadWriteLock(); public ExternalRowCountCache(ExecutorService executor) { + this(executor, null); + } + + ExternalRowCountCache(ExecutorService executor, Ticker ticker) { + this(executor, ticker, new RowCountCacheLoader()); + } + + ExternalRowCountCache(ExecutorService executor, Ticker ticker, RowCountCacheLoader loader) { // 1. set expireAfterWrite to 1 day, avoid too many entries // 2. set refreshAfterWrite to 10min(default), so that the cache will be refreshed after 10min CacheFactory rowCountCacheFactory = new CacheFactory( @@ -47,8 +67,9 @@ public ExternalRowCountCache(ExecutorService executor) { OptionalLong.of(Config.external_cache_refresh_time_minutes * 60), Config.max_external_table_row_count_cache_num, false, - null); - rowCountCache = rowCountCacheFactory.buildAsyncCache(new RowCountCacheLoader(), executor); + ticker); + rowCountCache = rowCountCacheFactory.buildAsyncCache( + new InvalidationAwareLoader(loader), executor); } @Getter @@ -87,6 +108,116 @@ protected Optional doLoad(RowCountKey rowCountKey) { } } + private final class InvalidationAwareLoader implements AsyncCacheLoader> { + private final RowCountCacheLoader delegate; + + private InvalidationAwareLoader(RowCountCacheLoader delegate) { + this.delegate = delegate; + } + + @Override + public CompletableFuture> asyncLoad(RowCountKey key, Executor executor) { + return loadWithInvalidationFence(key, executor, () -> delegate.doLoad(key)); + } + } + + private static final class LoadFence { + private boolean invalidated; + } + + // RowCountKey intentionally uses tableId as the Caffeine cache identity. In-flight loads need + // the complete scope so catalog/database invalidation can fence a same-tableId replacement. + private static final class LoadKey { + private final long catalogId; + private final long dbId; + private final long tableId; + + private LoadKey(RowCountKey key) { + catalogId = key.catalogId; + dbId = key.dbId; + tableId = key.tableId; + } + + @Override + public boolean equals(Object obj) { + if (this == obj) { + return true; + } + if (!(obj instanceof LoadKey)) { + return false; + } + LoadKey other = (LoadKey) obj; + return catalogId == other.catalogId && dbId == other.dbId && tableId == other.tableId; + } + + @Override + public int hashCode() { + return Objects.hash(catalogId, dbId, tableId); + } + } + + private CompletableFuture> loadWithInvalidationFence( + RowCountKey key, Executor executor, Supplier> loader) { + LoadFence fence = new LoadFence(); + LoadKey loadKey = new LoadKey(key); + publicationLock.readLock().lock(); + try { + inFlightLoads.compute(loadKey, (ignored, fences) -> { + Set currentFences = fences == null ? ConcurrentHashMap.newKeySet() : fences; + currentFences.add(fence); + return currentFences; + }); + } finally { + publicationLock.readLock().unlock(); + } + + CompletableFuture> publishedFuture = new CompletableFuture<>(); + CompletableFuture> loadFuture; + try { + loadFuture = CompletableFuture.supplyAsync(loader, executor); + } catch (RuntimeException e) { + publicationLock.readLock().lock(); + try { + removeInFlightLoad(loadKey, fence); + } finally { + publicationLock.readLock().unlock(); + } + throw e; + } + loadFuture.whenComplete((value, throwable) -> { + publicationLock.readLock().lock(); + try { + if (throwable != null) { + publishedFuture.completeExceptionally(throwable); + } else if (fence.invalidated) { + publishedFuture.complete(null); + } else { + publishedFuture.complete(value); + } + } finally { + removeInFlightLoad(loadKey, fence); + publicationLock.readLock().unlock(); + } + }); + return publishedFuture; + } + + private void removeInFlightLoad(LoadKey key, LoadFence fence) { + inFlightLoads.computeIfPresent(key, (ignored, fences) -> { + fences.remove(fence); + return fences.isEmpty() ? null : fences; + }); + } + + int getInFlightLoadCountForTest() { + publicationLock.readLock().lock(); + try { + return inFlightLoads.values().stream().mapToInt(Set::size).sum(); + } finally { + publicationLock.readLock().unlock(); + } + } + static Optional loadRowCount(RowCountKey rowCountKey, boolean fillMetaCache) { try { ExternalTable table = (ExternalTable) StatisticsUtil.findTable( @@ -122,10 +253,16 @@ static Optional loadRowCount(RowCountKey rowCountKey, boolean fillMetaCach public long getCachedRowCount(long catalogId, long dbId, long tableId, boolean fillMetaCache) { RowCountKey key = new RowCountKey(catalogId, dbId, tableId); try { - CompletableFuture> f = fillMetaCache - ? rowCountCache.get(key, (rowCountKey, executor) -> CompletableFuture.supplyAsync( - () -> loadRowCount(rowCountKey, true), executor)) - : rowCountCache.get(key); + CompletableFuture> f; + publicationLock.readLock().lock(); + try { + f = fillMetaCache + ? rowCountCache.get(key, (rowCountKey, executor) -> loadWithInvalidationFence( + rowCountKey, executor, () -> loadRowCount(rowCountKey, true))) + : rowCountCache.get(key); + } finally { + publicationLock.readLock().unlock(); + } // Get row count synchronously by default. if (ConnectContext.get() == null || ConnectContext.get().getSessionVariable().fetchHiveRowCountSync) { @@ -150,7 +287,13 @@ public long getCachedRowCount(long catalogId, long dbId, long tableId, boolean f public long getCachedRowCountIfPresent(long catalogId, long dbId, long tableId) { RowCountKey key = new RowCountKey(catalogId, dbId, tableId); try { - CompletableFuture> f = rowCountCache.getIfPresent(key); + CompletableFuture> f; + publicationLock.readLock().lock(); + try { + f = rowCountCache.getIfPresent(key); + } finally { + publicationLock.readLock().unlock(); + } if (f == null) { return -1; } else if (f.isDone()) { @@ -162,4 +305,48 @@ public long getCachedRowCountIfPresent(long catalogId, long dbId, long tableId) return -1; } + // Catalog/db invalidation is O(N): row-count keys are numeric ids, and Caffeine + // does not support prefix invalidation by catalog or database id. + void invalidateCatalog(long catalogId) { + publicationLock.writeLock().lock(); + try { + inFlightLoads.forEach((key, fences) -> { + if (key.catalogId == catalogId) { + fences.forEach(fence -> fence.invalidated = true); + } + }); + rowCountCache.asMap().keySet().removeIf(key -> key.catalogId == catalogId); + } finally { + publicationLock.writeLock().unlock(); + } + } + + void invalidateDb(long catalogId, long dbId) { + publicationLock.writeLock().lock(); + try { + inFlightLoads.forEach((key, fences) -> { + if (key.catalogId == catalogId && key.dbId == dbId) { + fences.forEach(fence -> fence.invalidated = true); + } + }); + rowCountCache.asMap().keySet().removeIf(key -> key.catalogId == catalogId && key.dbId == dbId); + } finally { + publicationLock.writeLock().unlock(); + } + } + + void invalidateTable(long catalogId, long dbId, long tableId) { + publicationLock.writeLock().lock(); + try { + RowCountKey key = new RowCountKey(catalogId, dbId, tableId); + Set fences = inFlightLoads.get(new LoadKey(key)); + if (fences != null) { + fences.forEach(fence -> fence.invalidated = true); + } + rowCountCache.synchronous().invalidate(key); + } finally { + publicationLock.writeLock().unlock(); + } + } + } diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/hive/HiveMetadataOps.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/hive/HiveMetadataOps.java index 7d8501f449fb50..94d7ee04ed222b 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/datasource/hive/HiveMetadataOps.java +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/hive/HiveMetadataOps.java @@ -346,8 +346,10 @@ public void afterTruncateTable(String dbName, String tblName, long updateTime) { if (tbl.isPresent()) { Env.getCurrentEnv().getRefreshManager() .refreshTableInternal(db.get(), (ExternalTable) tbl.get(), updateTime); + return; } } + Env.getCurrentEnv().getExtMetaCacheMgr().invalidateTable(catalog.getId(), dbName, tblName); } catch (Exception e) { LOG.warn("exception when calling afterTruncateTable for db: {}, table: {}, error: {}", dbName, tblName, e.getMessage(), e); diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/HiveInsertExecutor.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/HiveInsertExecutor.java index 760a2c0551d5a0..fac2dca5c91d61 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/HiveInsertExecutor.java +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/HiveInsertExecutor.java @@ -91,6 +91,7 @@ protected void doAfterCommit() throws DdlException { HiveExternalMetaCache cache = Env.getCurrentEnv().getExtMetaCacheMgr() .hive(hmsTable.getCatalog().getId()); cache.refreshAffectedPartitions(hmsTable, partitionUpdates, modifiedPartNames, newPartNames); + Env.getCurrentEnv().getExtMetaCacheMgr().invalidateRowCountCache(hmsTable); } else { // Non-partitioned table or no partition updates, do full table refresh Env.getCurrentEnv().getExtMetaCacheMgr().invalidateTableCache(hmsTable); diff --git a/fe/fe-core/src/test/java/org/apache/doris/catalog/RefreshManagerTest.java b/fe/fe-core/src/test/java/org/apache/doris/catalog/RefreshManagerTest.java new file mode 100644 index 00000000000000..1a5c82349e15e0 --- /dev/null +++ b/fe/fe-core/src/test/java/org/apache/doris/catalog/RefreshManagerTest.java @@ -0,0 +1,53 @@ +// 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.catalog; + +import org.apache.doris.datasource.CatalogMgr; +import org.apache.doris.datasource.ExternalCatalog; +import org.apache.doris.datasource.ExternalMetaCacheMgr; +import org.apache.doris.datasource.ExternalObjectLog; + +import org.junit.jupiter.api.Test; +import org.mockito.MockedStatic; +import org.mockito.Mockito; + +import java.util.Optional; + +public class RefreshManagerTest { + + @Test + void testColdDatabaseReplayInvalidatesCatalogRowCount() { + long catalogId = 51L; + ExternalCatalog catalog = Mockito.mock(ExternalCatalog.class); + Mockito.when(catalog.getId()).thenReturn(catalogId); + Mockito.when(catalog.getDbForReplay("db1")).thenReturn(Optional.empty()); + CatalogMgr catalogMgr = Mockito.mock(CatalogMgr.class); + Mockito.doReturn(catalog).when(catalogMgr).getCatalog(catalogId); + ExternalMetaCacheMgr cacheMgr = Mockito.mock(ExternalMetaCacheMgr.class); + Env env = Mockito.mock(Env.class); + Mockito.when(env.getCatalogMgr()).thenReturn(catalogMgr); + Mockito.when(env.getExtMetaCacheMgr()).thenReturn(cacheMgr); + + try (MockedStatic mockedEnv = Mockito.mockStatic(Env.class)) { + mockedEnv.when(Env::getCurrentEnv).thenReturn(env); + new RefreshManager().replayRefreshDb(ExternalObjectLog.createForRefreshDb(catalogId, "db1")); + } + + Mockito.verify(cacheMgr).invalidateRowCountCache(catalogId); + } +} diff --git a/fe/fe-core/src/test/java/org/apache/doris/datasource/CatalogMgrTest.java b/fe/fe-core/src/test/java/org/apache/doris/datasource/CatalogMgrTest.java index e17153e10f6593..23e6af0173c191 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/datasource/CatalogMgrTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/datasource/CatalogMgrTest.java @@ -21,8 +21,12 @@ import org.apache.doris.catalog.Env; import org.apache.doris.catalog.TableIf; import org.apache.doris.common.DdlException; +import org.apache.doris.datasource.hive.HMSExternalCatalog; +import org.apache.doris.datasource.hive.HMSExternalTable; +import org.apache.doris.datasource.metacache.MetaCache; import org.apache.doris.datasource.paimon.PaimonExternalCatalog; import org.apache.doris.datasource.property.metastore.AbstractPaimonProperties; +import org.apache.doris.nereids.exceptions.NotSupportedException; import com.google.common.collect.ImmutableMap; import org.junit.jupiter.api.Assertions; @@ -35,6 +39,7 @@ import java.util.HashMap; import java.util.List; import java.util.Map; +import java.util.Optional; import java.util.concurrent.ConcurrentMap; import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutorService; @@ -54,6 +59,17 @@ private static void addCatalog(CatalogMgr catalogMgr, ExternalCatalog catalog) t idToCatalog.put(catalog.getId(), catalog); } + private static void addNamedCatalog(CatalogMgr catalogMgr, ExternalCatalog catalog) throws Exception { + addCatalog(catalogMgr, catalog); + Field nameToCatalogField = CatalogMgr.class.getDeclaredField("nameToCatalog"); + nameToCatalogField.setAccessible(true); + @SuppressWarnings("unchecked") + ConcurrentMap>> nameToCatalog = + (ConcurrentMap>>) + nameToCatalogField.get(catalogMgr); + nameToCatalog.put(catalog.getName(), catalog); + } + @Test void testAlterCatalogRollsBackUncheckedValidationFailure() throws Exception { CatalogMgr catalogMgr = new CatalogMgr(); @@ -192,6 +208,61 @@ void testReplayKeepsPersistedLegacyPaimonOptionLoadableButInactive() throws Exce Assertions.assertTrue(restoredProperties.getTableOptionsMap().isEmpty()); } + @Test + void testUnsupportedAddPartitionEventStillInvalidatesRowCount() throws Exception { + CatalogMgr catalogMgr = new CatalogMgr(); + long catalogId = 46L; + HMSExternalCatalog catalog = Mockito.mock(HMSExternalCatalog.class); + ExternalDatabase db = Mockito.mock(ExternalDatabase.class); + HMSExternalTable table = Mockito.mock(HMSExternalTable.class); + Mockito.when(catalog.getId()).thenReturn(catalogId); + Mockito.when(catalog.getName()).thenReturn("hms"); + Mockito.doReturn(db).when(catalog).getDbNullable("db1"); + Mockito.when(db.getTableNullable("tbl1")).thenReturn(table); + Mockito.when(table.getPartitionColumnTypes(Mockito.any())) + .thenThrow(new NotSupportedException("unsupported table")); + addNamedCatalog(catalogMgr, catalog); + + Env env = Mockito.mock(Env.class); + ExternalMetaCacheMgr cacheMgr = Mockito.mock(ExternalMetaCacheMgr.class); + Mockito.when(env.getExtMetaCacheMgr()).thenReturn(cacheMgr); + try (MockedStatic mockedEnv = Mockito.mockStatic(Env.class)) { + mockedEnv.when(Env::getCurrentEnv).thenReturn(env); + catalogMgr.addExternalPartitions( + "hms", "db1", "tbl1", Collections.singletonList("p=1"), 1L, false); + } + + Mockito.verify(cacheMgr).invalidateRowCountCache(table); + Mockito.verify(cacheMgr, Mockito.never()).hive(catalogId); + } + + @Test + void testUnregisterDatabaseRemovesLocalEntryWhenEngineInvalidationFails() { + long catalogId = 47L; + long dbId = 48L; + TestingUnregisterCatalog catalog = new TestingUnregisterCatalog(catalogId); + @SuppressWarnings("unchecked") + MetaCache> metaCache = Mockito.mock(MetaCache.class); + ExternalDatabase db = Mockito.mock(ExternalDatabase.class); + Mockito.when(db.getId()).thenReturn(dbId); + Mockito.when(db.getFullName()).thenReturn("CanonicalDb"); + Mockito.when(metaCache.tryGetMetaObj("CanonicalDb")).thenReturn(Optional.of(db)); + catalog.installMetaCache(metaCache); + + Env env = Mockito.mock(Env.class); + ExternalMetaCacheMgr cacheMgr = Mockito.mock(ExternalMetaCacheMgr.class); + Mockito.when(env.getExtMetaCacheMgr()).thenReturn(cacheMgr); + Mockito.doThrow(new IllegalStateException("engine invalidation failed")) + .when(cacheMgr).invalidateDb(catalogId, dbId, "CanonicalDb"); + try (MockedStatic mockedEnv = Mockito.mockStatic(Env.class)) { + mockedEnv.when(Env::getCurrentEnv).thenReturn(env); + Assertions.assertThrows(IllegalStateException.class, + () -> catalog.unregisterDatabase("CanonicalDb")); + } + + Mockito.verify(metaCache).invalidate("CanonicalDb", dbId); + } + private static class LatchingValidationCatalog extends ExternalCatalog { private final CountDownLatch validationStarted = new CountDownLatch(1); private final CountDownLatch initializationReadProperties = new CountDownLatch(1); @@ -243,4 +314,30 @@ public void notifyPropertiesUpdated(Map updatedProps) { // This test isolates edit-log property restoration from environment-owned cache services. } } + + private static class TestingUnregisterCatalog extends ExternalCatalog { + TestingUnregisterCatalog(long id) { + super(id, "testing_catalog", InitCatalogLog.Type.TEST, ""); + catalogProperty = new CatalogProperty(null, Collections.emptyMap()); + } + + void installMetaCache(MetaCache> cache) { + metaCache = cache; + initialized = true; + } + + @Override + protected List listTableNamesFromRemote(SessionContext ctx, String dbName) { + return Collections.emptyList(); + } + + @Override + public boolean tableExist(SessionContext ctx, String dbName, String tblName) { + return false; + } + + @Override + protected void initLocalObjectsImpl() { + } + } } diff --git a/fe/fe-core/src/test/java/org/apache/doris/datasource/ExternalMetaCacheRouteResolverTest.java b/fe/fe-core/src/test/java/org/apache/doris/datasource/ExternalMetaCacheRouteResolverTest.java index 111a8173257c3a..8f1a734caac2d7 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/datasource/ExternalMetaCacheRouteResolverTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/datasource/ExternalMetaCacheRouteResolverTest.java @@ -29,6 +29,7 @@ import org.apache.doris.datasource.metacache.MetaCacheEntryStats; import org.apache.doris.datasource.paimon.PaimonExternalCatalog; +import com.google.common.util.concurrent.Uninterruptibles; import mockit.Mock; import mockit.MockUp; import org.junit.Assert; @@ -39,6 +40,7 @@ import java.util.HashMap; import java.util.List; import java.util.Map; +import java.util.Optional; import java.util.Set; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.CountDownLatch; @@ -304,12 +306,40 @@ public void testRollbackRetiresGroupInitializedFromRejectedCandidate() throws Ex mockCurrentCatalog(catalogId, catalog); hive.initializedCatalogIds.add(catalogId); Map oldProperties = Collections.singletonMap("generation", "old"); - - metaCacheMgr.rollbackCatalogProperties(catalog, oldProperties); - - Mockito.verify(catalog).rollBackCatalogProps(oldProperties); - Assert.assertFalse(hive.isCatalogInitialized(catalogId)); - Assert.assertEquals(1, hive.invalidateCatalogCalls); + CountDownLatch loadStarted = new CountDownLatch(1); + CountDownLatch allowLoadToFinish = new CountDownLatch(1); + ExternalRowCountCache.RowCountCacheLoader loader = new ExternalRowCountCache.RowCountCacheLoader() { + @Override + protected Optional doLoad(ExternalRowCountCache.RowCountKey rowCountKey) { + loadStarted.countDown(); + Assert.assertTrue(Uninterruptibles.awaitUninterruptibly( + allowLoadToFinish, 30, TimeUnit.SECONDS)); + return Optional.of(100L); + } + }; + ExecutorService loaderExecutor = Executors.newSingleThreadExecutor(); + metaCacheMgr.replaceRowCountCacheForTest(new ExternalRowCountCache(loaderExecutor, null, loader)); + ExecutorService caller = Executors.newSingleThreadExecutor(); + try { + Future load = caller.submit( + () -> metaCacheMgr.getRowCountCache().getCachedRowCount(catalogId, 2L, 3L, false)); + Assert.assertTrue(loadStarted.await(30, TimeUnit.SECONDS)); + + metaCacheMgr.rollbackCatalogProperties(catalog, oldProperties); + allowLoadToFinish.countDown(); + + Assert.assertEquals(TableIf.UNKNOWN_ROW_COUNT, (long) load.get(30, TimeUnit.SECONDS)); + Assert.assertEquals(TableIf.UNKNOWN_ROW_COUNT, + metaCacheMgr.getRowCountCache().getCachedRowCountIfPresent(catalogId, 2L, 3L)); + Mockito.verify(catalog).rollBackCatalogProps(oldProperties); + Mockito.verify(catalog).resetToUninitialized(false); + Assert.assertFalse(hive.isCatalogInitialized(catalogId)); + Assert.assertEquals(1, hive.invalidateCatalogCalls); + } finally { + allowLoadToFinish.countDown(); + caller.shutdownNow(); + loaderExecutor.shutdownNow(); + } } @Test @@ -410,6 +440,35 @@ public void testMissingCatalogLifecycleOnlyTouchesInitializedEngine() throws Exc Assert.assertEquals(0, paimon.invalidateCatalogCalls); } + @Test + public void testEngineSpecificTableInvalidationAlsoFencesRowCount() throws Exception { + RecordingExternalMetaCache hive = new RecordingExternalMetaCache( + "hive", Collections.singletonList("hms"), catalog -> catalog instanceof HMSExternalCatalog); + ExternalMetaCacheMgr metaCacheMgr = newManagerWithCaches(hive); + ExternalRowCountCache rowCountCache = Mockito.mock(ExternalRowCountCache.class); + metaCacheMgr.replaceRowCountCacheForTest(rowCountCache); + long catalogId = 17L; + long dbId = 18L; + long tableId = 19L; + HMSExternalCatalog catalog = Mockito.mock(HMSExternalCatalog.class); + ExternalDatabase db = Mockito.mock(ExternalDatabase.class); + ExternalTable table = Mockito.mock(ExternalTable.class); + Mockito.when(catalog.getId()).thenReturn(catalogId); + Mockito.when(catalog.getDbForReplay("db1")).thenReturn(Optional.of(db)); + Mockito.when(db.getId()).thenReturn(dbId); + Mockito.doReturn(Optional.of(table)).when(db).getTableForReplay("tbl1"); + Mockito.when(table.getCatalog()).thenReturn(catalog); + Mockito.when(table.getDb()).thenReturn(db); + Mockito.when(table.getId()).thenReturn(tableId); + mockCurrentCatalog(catalogId, catalog); + hive.initializedCatalogIds.add(catalogId); + + metaCacheMgr.invalidateTableByEngine(catalogId, "hive", "db1", "tbl1"); + + Assert.assertEquals(1, hive.invalidateTableCalls); + Mockito.verify(rowCountCache).invalidateTable(catalogId, dbId, tableId); + } + @SuppressWarnings("unchecked") private ExternalMetaCacheMgr newManagerWithCaches(RecordingExternalMetaCache... caches) throws Exception { ExternalMetaCacheMgr metaCacheMgr = new ExternalMetaCacheMgr(true); diff --git a/fe/fe-core/src/test/java/org/apache/doris/datasource/ExternalRowCountCacheTest.java b/fe/fe-core/src/test/java/org/apache/doris/datasource/ExternalRowCountCacheTest.java index 075806aa82bc49..32bb80e4bf0aae 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/datasource/ExternalRowCountCacheTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/datasource/ExternalRowCountCacheTest.java @@ -17,11 +17,15 @@ package org.apache.doris.datasource; +import org.apache.doris.catalog.Env; import org.apache.doris.catalog.TableIf; +import org.apache.doris.common.Config; import org.apache.doris.common.ThreadPoolManager; import org.apache.doris.statistics.util.StatisticsUtil; +import com.google.common.testing.FakeTicker; import com.google.common.util.concurrent.MoreExecutors; +import com.google.common.util.concurrent.Uninterruptibles; import mockit.Mock; import mockit.MockUp; import org.junit.jupiter.api.Assertions; @@ -29,10 +33,45 @@ import org.mockito.Mockito; import java.util.Optional; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.RejectedExecutionException; import java.util.concurrent.ThreadPoolExecutor; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; public class ExternalRowCountCacheTest { + @Test + public void testCatalogCacheResetSuppressesPerDatabaseRowCountScan() { + ExternalCatalog catalog = Mockito.mock(ExternalCatalog.class); + Mockito.when(catalog.getId()).thenReturn(1L); + ExternalMetaCacheMgr metaCacheMgr = Mockito.mock(ExternalMetaCacheMgr.class); + Env env = Mockito.mock(Env.class); + Mockito.when(env.getExtMetaCacheMgr()).thenReturn(metaCacheMgr); + new MockUp() { + @Mock + Env getCurrentEnv() { + return env; + } + }; + ExternalDatabase db = new ExternalDatabase( + catalog, 2L, "db", "db", InitDatabaseLog.Type.TEST) { + @Override + protected ExternalTable buildTableInternal(String remoteTableName, String localTableName, long tblId, + ExternalCatalog externalCatalog, ExternalDatabase externalDatabase) { + return null; + } + }; + + db.resetMetaToUninitialized(false); + Mockito.verifyNoInteractions(metaCacheMgr); + + db.resetMetaToUninitialized(); + Mockito.verify(metaCacheMgr).invalidateDb(1L, 2L, "db"); + } + @Test public void testRowCountKeyUsesTableIdAsCacheIdentity() { ExternalRowCountCache.RowCountKey key1 = new ExternalRowCountCache.RowCountKey(1, 2, 3); @@ -84,20 +123,155 @@ public TableIf findTable(long catalogId, long dbId, long tblId) { Mockito.verify(table).fetchRowCountWithMetaCache(false); } + @Test + public void testInvalidationScopes() { + ExternalRowCountCache.RowCountCacheLoader loader = new ExternalRowCountCache.RowCountCacheLoader() { + @Override + protected Optional doLoad(ExternalRowCountCache.RowCountKey rowCountKey) { + return Optional.of(rowCountKey.getTableId()); + } + }; + + ExternalRowCountCache cache = new ExternalRowCountCache( + MoreExecutors.newDirectExecutorService(), null, loader); + Assertions.assertEquals(100L, cache.getCachedRowCount(1, 10, 100, false)); + Assertions.assertEquals(101L, cache.getCachedRowCount(1, 10, 101, false)); + Assertions.assertEquals(102L, cache.getCachedRowCount(1, 11, 102, false)); + Assertions.assertEquals(200L, cache.getCachedRowCount(2, 20, 200, false)); + + cache.invalidateTable(1, 10, 100); + Assertions.assertEquals(TableIf.UNKNOWN_ROW_COUNT, cache.getCachedRowCountIfPresent(1, 10, 100)); + Assertions.assertEquals(101L, cache.getCachedRowCountIfPresent(1, 10, 101)); + + cache.invalidateDb(1, 10); + Assertions.assertEquals(TableIf.UNKNOWN_ROW_COUNT, cache.getCachedRowCountIfPresent(1, 10, 101)); + Assertions.assertEquals(102L, cache.getCachedRowCountIfPresent(1, 11, 102)); + + cache.invalidateCatalog(1); + Assertions.assertEquals(TableIf.UNKNOWN_ROW_COUNT, cache.getCachedRowCountIfPresent(1, 11, 102)); + Assertions.assertEquals(200L, cache.getCachedRowCountIfPresent(2, 20, 200)); + } + + @Test + public void testInvalidateWhileRefreshIsRunningDoesNotRepublishStaleValue() throws Exception { + AtomicInteger loadCount = new AtomicInteger(); + CountDownLatch refreshStarted = new CountDownLatch(1); + CountDownLatch allowRefreshToFinish = new CountDownLatch(1); + ExternalRowCountCache.RowCountCacheLoader loader = new ExternalRowCountCache.RowCountCacheLoader() { + @Override + protected Optional doLoad(ExternalRowCountCache.RowCountKey rowCountKey) { + int currentLoad = loadCount.incrementAndGet(); + if (currentLoad == 2) { + refreshStarted.countDown(); + Assertions.assertTrue(Uninterruptibles.awaitUninterruptibly( + allowRefreshToFinish, 30, TimeUnit.SECONDS)); + } + return Optional.of(currentLoad * 100L); + } + }; + + ExecutorService executor = Executors.newSingleThreadExecutor(); + try { + FakeTicker ticker = new FakeTicker(); + ExternalRowCountCache cache = new ExternalRowCountCache(executor, ticker::read, loader); + Assertions.assertEquals(100L, cache.getCachedRowCount(1, 10, 100, false)); + + ticker.advance(Config.external_cache_refresh_time_minutes + 1, TimeUnit.MINUTES); + Assertions.assertEquals(100L, cache.getCachedRowCount(1, 10, 100, false)); + Assertions.assertTrue(refreshStarted.await(30, TimeUnit.SECONDS)); + + cache.invalidateTable(1, 10, 100); + allowRefreshToFinish.countDown(); + executor.submit(() -> { }).get(30, TimeUnit.SECONDS); + + Assertions.assertEquals(TableIf.UNKNOWN_ROW_COUNT, + cache.getCachedRowCountIfPresent(1, 10, 100)); + Assertions.assertEquals(300L, cache.getCachedRowCount(1, 10, 100, false)); + } finally { + allowRefreshToFinish.countDown(); + executor.shutdownNow(); + } + } + + @Test + public void testCatalogInvalidationUsesFullInFlightLoadIdentity() throws Exception { + CountDownLatch oldLoadStarted = new CountDownLatch(1); + CountDownLatch newLoadStarted = new CountDownLatch(1); + CountDownLatch allowLoadsToFinish = new CountDownLatch(1); + ExternalRowCountCache.RowCountCacheLoader loader = new ExternalRowCountCache.RowCountCacheLoader() { + @Override + protected Optional doLoad(ExternalRowCountCache.RowCountKey rowCountKey) { + if (rowCountKey.getCatalogId() == 1L) { + oldLoadStarted.countDown(); + } else { + newLoadStarted.countDown(); + } + Assertions.assertTrue(Uninterruptibles.awaitUninterruptibly( + allowLoadsToFinish, 30, TimeUnit.SECONDS)); + return Optional.of(rowCountKey.getCatalogId() * 100L); + } + }; + + ExecutorService loaderExecutor = Executors.newFixedThreadPool(2); + ExecutorService callers = Executors.newFixedThreadPool(2); + try { + ExternalRowCountCache cache = new ExternalRowCountCache(loaderExecutor, null, loader); + Future oldLoad = callers.submit(() -> cache.getCachedRowCount(1, 10, 100, false)); + Assertions.assertTrue(oldLoadStarted.await(30, TimeUnit.SECONDS)); + cache.invalidateCatalog(1L); + + Future newLoad = callers.submit(() -> cache.getCachedRowCount(2, 20, 100, false)); + Assertions.assertTrue(newLoadStarted.await(30, TimeUnit.SECONDS)); + cache.invalidateCatalog(2L); + allowLoadsToFinish.countDown(); + + Assertions.assertEquals(TableIf.UNKNOWN_ROW_COUNT, oldLoad.get(30, TimeUnit.SECONDS)); + Assertions.assertEquals(TableIf.UNKNOWN_ROW_COUNT, newLoad.get(30, TimeUnit.SECONDS)); + Assertions.assertEquals(TableIf.UNKNOWN_ROW_COUNT, + cache.getCachedRowCountIfPresent(2, 20, 100)); + } finally { + allowLoadsToFinish.countDown(); + callers.shutdownNow(); + loaderExecutor.shutdownNow(); + } + } + + @Test + public void testRejectedSubmissionRemovesInFlightFence() { + ExecutorService rejectingExecutor = Mockito.mock(ExecutorService.class); + Mockito.doThrow(new RejectedExecutionException("rejected")) + .when(rejectingExecutor).execute(Mockito.any(Runnable.class)); + ExternalRowCountCache cache = new ExternalRowCountCache(rejectingExecutor); + + Assertions.assertEquals(TableIf.UNKNOWN_ROW_COUNT, + cache.getCachedRowCount(1, 10, 100, false)); + Assertions.assertEquals(0, cache.getInFlightLoadCountForTest()); + } + @Test public void testLoadWithException() throws Exception { ThreadPoolExecutor executor = ThreadPoolManager.newDaemonFixedThreadPool( 1, Integer.MAX_VALUE, "TEST", true); AtomicInteger counter = new AtomicInteger(0); - new MockUp() { - @Mock + ExternalRowCountCache.RowCountCacheLoader loader = new ExternalRowCountCache.RowCountCacheLoader() { + @Override protected Optional doLoad(ExternalRowCountCache.RowCountKey rowCountKey) { - counter.incrementAndGet(); - return null; + int currentLoad = counter.incrementAndGet(); + if (currentLoad == 1) { + return null; + } + if (currentLoad == 3) { + try { + Thread.sleep(2000); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + } + return Optional.of(100L); } }; - ExternalRowCountCache cache = new ExternalRowCountCache(executor); + ExternalRowCountCache cache = new ExternalRowCountCache(executor, null, loader); long cachedRowCount = cache.getCachedRowCount(1, 1, 1, false); Assertions.assertEquals(TableIf.UNKNOWN_ROW_COUNT, cachedRowCount); for (int i = 0; i < 60; i++) { @@ -108,13 +282,6 @@ protected Optional doLoad(ExternalRowCountCache.RowCountKey rowCountKey) { } Assertions.assertEquals(1, counter.get()); - new MockUp() { - @Mock - protected Optional doLoad(ExternalRowCountCache.RowCountKey rowCountKey) { - counter.incrementAndGet(); - return Optional.of(100L); - } - }; cache.getCachedRowCount(1, 1, 1, false); for (int i = 0; i < 60; i++) { cachedRowCount = cache.getCachedRowCount(1, 1, 1, false); @@ -128,18 +295,6 @@ protected Optional doLoad(ExternalRowCountCache.RowCountKey rowCountKey) { Assertions.assertEquals(100, cachedRowCount); Assertions.assertEquals(2, counter.get()); - new MockUp() { - @Mock - protected Optional doLoad(ExternalRowCountCache.RowCountKey rowCountKey) { - counter.incrementAndGet(); - try { - Thread.sleep(2000); - } catch (InterruptedException e) { - e.printStackTrace(); - } - return Optional.of(100L); - } - }; cachedRowCount = cache.getCachedRowCount(2, 2, 2, false); Assertions.assertEquals(100, cachedRowCount); Thread.sleep(1000); diff --git a/fe/fe-core/src/test/java/org/apache/doris/datasource/hive/HiveMetadataOpsTest.java b/fe/fe-core/src/test/java/org/apache/doris/datasource/hive/HiveMetadataOpsTest.java new file mode 100644 index 00000000000000..89da06271dcbfe --- /dev/null +++ b/fe/fe-core/src/test/java/org/apache/doris/datasource/hive/HiveMetadataOpsTest.java @@ -0,0 +1,56 @@ +// 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.datasource.hive; + +import org.apache.doris.catalog.Env; +import org.apache.doris.datasource.ExternalDatabase; +import org.apache.doris.datasource.ExternalMetaCacheMgr; + +import mockit.Mock; +import mockit.MockUp; +import org.junit.jupiter.api.Test; +import org.mockito.Mockito; + +import java.util.Optional; + +public class HiveMetadataOpsTest { + @Test + @SuppressWarnings("unchecked") + public void testColdTruncateReplayInvalidatesByName() { + HMSExternalCatalog catalog = Mockito.mock(HMSExternalCatalog.class); + Mockito.when(catalog.getId()).thenReturn(1L); + ExternalDatabase db = Mockito.mock(ExternalDatabase.class); + Mockito.when(db.getTableForReplay("tbl")).thenReturn(Optional.empty()); + Mockito.when(catalog.getDbForReplay("db")).thenReturn((Optional) Optional.of(db)); + + ExternalMetaCacheMgr metaCacheMgr = Mockito.mock(ExternalMetaCacheMgr.class); + Env env = Mockito.mock(Env.class); + Mockito.when(env.getExtMetaCacheMgr()).thenReturn(metaCacheMgr); + new MockUp() { + @Mock + Env getCurrentEnv() { + return env; + } + }; + + HiveMetadataOps metadataOps = new HiveMetadataOps(catalog, Mockito.mock(HMSCachedClient.class)); + metadataOps.afterTruncateTable("db", "tbl", 100L); + + Mockito.verify(metaCacheMgr).invalidateTable(1L, "db", "tbl"); + } +}