branch-4.1: [fix](meta cache) Invalidate external row count cache after metadata changes #64160 #68196
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -182,6 +182,7 @@ public abstract class ExternalCatalog | |
| protected ExternalMetadataOps metadataOps; | ||
| protected TransactionManager transactionManager; | ||
| protected MetaCache<ExternalDatabase<? extends ExternalTable>> 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(!invalidatingAllMetaCache))); | ||
|
Contributor
There was a problem hiding this comment. [P2] Avoid scanning the global row-count cache twice for |
||
| } | ||
| } | ||
|
|
||
|
|
@@ -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 synchronized void refreshMetaCacheOnly() { | ||
| if (metaCache != null) { | ||
| 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); | ||
| } | ||
| // Resolve the canonical database object before removing it from the local metadata cache. | ||
|
Contributor
There was a problem hiding this comment. [P2] Keep local DB removal independent of routed cache failures. This new invalidateDb call runs before metaCache.invalidate so it can resolve the numeric ID, but safeInvalidate does not catch a runtime failure from an engine cache. If that happens, control never reaches local removal and the dropped DB remains visible with stale metadata/replay state. Please capture the canonical ID/object first and make metadata removal and row-count fencing finally-safe around engine invalidation.
Contributor
Author
There was a problem hiding this comment. Fixed in 8b4749c. unregisterDatabase captures the canonical database name and numeric ID first, then performs local MetaCache removal in finally so an engine invalidation exception cannot leave the dropped DB visible. The manager-level invalidateDb finally block still fences row counts. Added testUnregisterDatabaseRemovesLocalEntryWhenEngineInvalidationFails. |
||
| // The row-count cache can outlive that object and must be invalidated by its numeric id. | ||
| boolean catalogInitialized = isInitialized(); | ||
| Optional<ExternalDatabase<? extends ExternalTable>> 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); | ||
| } | ||
| } | ||
| } | ||
|
|
||
| public void registerDatabase(long dbId, String dbName) { | ||
|
|
||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -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<String, String> runtimeEffectiveCacheProperties( | |
| } | ||
|
|
||
| public void invalidateCatalog(long catalogId) { | ||
| try { | ||
| routeCatalogEngines(catalogId, cache -> safeInvalidate( | ||
| cache, catalogId, "invalidateCatalog", | ||
| () -> cache.invalidateCatalogEntries(catalogId))); | ||
| } finally { | ||
| rowCountCache.invalidateCatalog(catalogId); | ||
|
Contributor
There was a problem hiding this comment. [P1] Invalidate row counts on every whole-catalog retirement. This call is only reached by |
||
| } | ||
| } | ||
|
|
||
| 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); | ||
|
Contributor
There was a problem hiding this comment. [P1] Apply this fence to tentative-property rollback as well. Legacy validators publish |
||
| lifecycleLock.unlock(); | ||
| } | ||
| } | ||
|
|
@@ -463,10 +470,12 @@ public void rollbackCatalogProperties(ExternalCatalog catalog, Map<String, Strin | |
| lifecycleLock.lock(); | ||
| try { | ||
| catalog.rollBackCatalogProps(oldProperties); | ||
| catalog.resetToUninitialized(false); | ||
| routeCatalogEngines(catalogId, cache -> 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) { | ||
| 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) { | ||
| Optional<ExternalDatabase<? extends ExternalTable>> 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) { | ||
| Optional<ExternalDatabase<? extends ExternalTable>> 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<String> partitions) { | ||
| Optional<ExternalDatabase<? extends ExternalTable>> db = getCachedDb(catalogId, dbName); | ||
| try { | ||
| routeCatalogEngines(catalogId, cache -> safeInvalidate( | ||
| cache, catalogId, "invalidatePartitions", | ||
| () -> cache.invalidatePartitions(catalogId, dbName, tableName, partitions))); | ||
|
Contributor
There was a problem hiding this comment. [P1] Fence row counts on the engine-only retry. When Hive catches ERR_CACHE_INCONSISTENCY, getHivePartitionValues calls invalidateTableByEngine and retries, but this helper only clears the selected metadata engine. The independent rowCountCache remains publishable, while Hive's fallback row-count estimate reads the same partition/file metadata, so a subsequent getRowCount can return the pre-retry value until TTL. Please fence the table row-count identity here (or in the retry) and add a regression test.
Contributor
Author
There was a problem hiding this comment. Fixed in 8b4749c. invalidateTableByEngine now captures the cached DB/table identity and fences rowCountCache in a finally block, sharing the same conservative table/DB/catalog fallback as the normal invalidation path. Added testEngineSpecificTableInvalidationAlsoFencesRowCount. |
||
| } finally { | ||
| invalidateTableRowCount(catalogId, db, tableName); | ||
| } | ||
| } | ||
|
|
||
| private void invalidateTableRowCount(long catalogId, | ||
| Optional<ExternalDatabase<? extends ExternalTable>> db, String tableName) { | ||
| if (db.isPresent()) { | ||
| Optional<? extends ExternalTable> 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<ExternalDatabase<? extends ExternalTable>> db = getCachedDb(catalogId, dbName); | ||
| return db.isPresent() ? OptionalLong.of(db.get().getId()) : OptionalLong.empty(); | ||
| } | ||
|
|
||
| private Optional<ExternalDatabase<? extends ExternalTable>> getCachedDb(long catalogId, String dbName) { | ||
| CatalogIf<?> catalog = getCatalog(catalogId); | ||
| if (!(catalog instanceof ExternalCatalog)) { | ||
| return Optional.empty(); | ||
| } | ||
| return ((ExternalCatalog) catalog).getDbForReplay(dbName); | ||
| } | ||
|
|
||
| public List<CatalogMetaCacheStats> getCatalogCacheStats(long catalogId) { | ||
|
|
@@ -669,15 +739,32 @@ public ExternalRowCountCache getRowCountCache() { | |
| } | ||
|
|
||
| public void invalidateTableCache(ExternalTable dorisTable) { | ||
| long catalogId = dorisTable.getCatalog().getId(); | ||
| try { | ||
| routeCatalogEngines(catalogId, cache -> safeInvalidate( | ||
| cache, catalogId, "invalidateTableCache", | ||
| () -> cache.invalidateTable(catalogId, dorisTable.getDbName(), dorisTable.getName()))); | ||
| } finally { | ||
| invalidateRowCountCache(dorisTable); | ||
|
Contributor
There was a problem hiding this comment. [P1] Make this row-count fence reachable from cold TRUNCATE replay. A follower dispatches |
||
| } | ||
| 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<? extends ExternalMetaCache> caches) { | |
| bindCatalogPreparers(); | ||
| } | ||
|
|
||
| void replaceRowCountCacheForTest(ExternalRowCountCache cache) { | ||
| rowCountCache = cache; | ||
| } | ||
|
|
||
| /** | ||
| * Fallback implementation of {@link AbstractExternalMetaCache} for engines that do not | ||
| * provide dedicated cache entries. | ||
|
|
||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
[P1] Fence cold REFRESH DATABASE replay too. This new table-replay miss branch retires catalog/DB row counts, but replayRefreshDb still cache-lookups the database and only logs/returns when it is absent. Because the DB metadata cache can be evicted while row-count entries remain, a replayed REFRESH DATABASE can leave a stale estimate to be served later. Please invalidate the catalog (or carry/use a canonical DB ID) on the missing-DB branch and add a cold database-cache replay test.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Fixed in 8b4749c. When replayRefreshDb cannot recover a cached database object, it now retires the catalog row-count scope without loading remote metadata. Added testColdDatabaseReplayInvalidatesCatalogRowCount.