This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.2 by this push:
new 37671319cde branch-4.2: [fix](metacache) Complete guarded external
table lookup cleanup (#69806) (#70158)
37671319cde is described below
commit 37671319cdefc44dd8e5b85a0584a4826460f03d
Author: 924060929 <[email protected]>
AuthorDate: Wed Nov 4 14:22:24 2026 +0800
branch-4.2: [fix](metacache) Complete guarded external table lookup cleanup
(#69806) (#70158)
cherry pick from #69806
---
.../org/apache/doris/catalog/RefreshManager.java | 369 +++++++++++----------
.../apache/doris/datasource/ExternalCatalog.java | 96 ++++--
.../apache/doris/datasource/ExternalDatabase.java | 9 +-
.../datasource/ExternalLookupRefreshTest.java | 206 ++++++++++++
4 files changed, 482 insertions(+), 198 deletions(-)
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 76f6e207b98..70ac400cfe8 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
@@ -95,13 +95,18 @@ public class RefreshManager {
if (!(catalog instanceof ExternalCatalog)) {
throw new DdlException("Only support refresh database in external
catalog");
}
- DatabaseIf db = catalog.getDbOrDdlException(dbName);
- // Local DB-object eviction also resets metadata; only an explicit
refresh retires access.
- invalidateLanceTableAccess((ExternalDatabase) db);
- refreshDbInternal((ExternalDatabase) db);
+ ((ExternalCatalog) catalog).beginTableLookupRefresh();
+ try {
+ DatabaseIf db = catalog.getDbOrDdlException(dbName);
+ // Local DB-object eviction also resets metadata; only an explicit
refresh retires access.
+ invalidateLanceTableAccess((ExternalDatabase) db);
+ refreshDbInternal((ExternalDatabase) db);
- ExternalObjectLog log =
ExternalObjectLog.createForRefreshDb(catalog.getId(), db.getFullName());
- Env.getCurrentEnv().getEditLog().logRefreshExternalDb(log);
+ ExternalObjectLog log =
ExternalObjectLog.createForRefreshDb(catalog.getId(), db.getFullName());
+ Env.getCurrentEnv().getEditLog().logRefreshExternalDb(log);
+ } finally {
+ ((ExternalCatalog) catalog).endTableLookupRefresh();
+ }
}
public void replayRefreshDb(ExternalObjectLog log) {
@@ -112,18 +117,23 @@ public class RefreshManager {
LOG.warn("failed to find catalog when replaying refresh db:
{}", log.debugForRefreshDb());
return;
}
- invalidateLanceTableAccess(catalog, log);
- Optional<ExternalDatabase<? extends ExternalTable>> db;
- if (!Strings.isNullOrEmpty(log.getDbName())) {
- db = catalog.getDbForReplay(log.getDbName());
- } else {
- db = catalog.getDbForReplay(log.getDbId());
- }
+ catalog.beginTableLookupRefresh();
+ try {
+ invalidateLanceTableAccess(catalog, log);
+ Optional<ExternalDatabase<? extends ExternalTable>> db;
+ if (!Strings.isNullOrEmpty(log.getDbName())) {
+ db = catalog.getDbForReplay(log.getDbName());
+ } else {
+ db = catalog.getDbForReplay(log.getDbId());
+ }
- if (!db.isPresent()) {
- invalidateColdReplayDatabase(catalog, log);
- } else {
- refreshDbInternal(db.get());
+ if (!db.isPresent()) {
+ invalidateColdReplayDatabase(catalog, log);
+ } else {
+ refreshDbInternal(db.get());
+ }
+ } finally {
+ catalog.endTableLookupRefresh();
}
});
}
@@ -173,42 +183,52 @@ public class RefreshManager {
throw new DdlException("Only support refresh ExternalCatalog
Tables");
}
- DatabaseIf db = catalog.getDbNullable(dbName);
- if (db == null) {
- if (!ignoreIfNotExists) {
- throw new DdlException("Database " + dbName + " does not exist
in catalog " + catalog.getName());
+ ((ExternalCatalog) catalog).beginTableLookupRefresh();
+ try {
+ DatabaseIf db = catalog.getDbNullable(dbName);
+ if (db == null) {
+ if (!ignoreIfNotExists) {
+ throw new DdlException("Database " + dbName + " does not
exist in catalog " + catalog.getName());
+ }
+ return;
}
- return;
- }
- TableIf table = db.getTableNullable(tableName);
- if (table == null) {
- if (!ignoreIfNotExists) {
- throw new DdlException("Table " + tableName + " does not exist
in db " + dbName);
+ TableIf table = db.getTableNullable(tableName);
+ if (table == null) {
+ if (!ignoreIfNotExists) {
+ throw new DdlException("Table " + tableName + " does not
exist in db " + dbName);
+ }
+ return;
}
- return;
+ long updateTime = System.currentTimeMillis();
+ refreshTableInternal((ExternalDatabase) db, (ExternalTable) table,
updateTime, true);
+ ExternalObjectLog log =
ExternalObjectLog.createForRefreshTable(catalog.getId(), db.getFullName(),
+ table.getName(), updateTime, true);
+ Env.getCurrentEnv().getEditLog().logRefreshExternalTable(log);
+ } finally {
+ ((ExternalCatalog) catalog).endTableLookupRefresh();
}
- long updateTime = System.currentTimeMillis();
- refreshTableInternal((ExternalDatabase) db, (ExternalTable) table,
updateTime, true);
- ExternalObjectLog log =
ExternalObjectLog.createForRefreshTable(catalog.getId(), db.getFullName(),
- table.getName(), updateTime, true);
- Env.getCurrentEnv().getEditLog().logRefreshExternalTable(log);
}
public void refreshTableAfterCommit(ExternalTable table) {
- long updateTime = System.currentTimeMillis();
+ ((ExternalCatalog) table.getCatalog()).beginTableLookupRefresh();
try {
-
Env.getCurrentEnv().getExtMetaCacheMgr().invalidateRowCountCache(table);
- refreshTableInternal((ExternalDatabase) table.getDatabase(),
table, updateTime);
- } catch (RuntimeException e) {
- // The external transaction is already committed. Still notify
follower FEs even if a
- // cache layer failed, so peers do not keep serving the pre-commit
state.
- LOG.warn("Failed to refresh table cache after committing external
insert for {}",
- table.getNameWithFullQualifiers(), e);
+ long updateTime = System.currentTimeMillis();
+ try {
+
Env.getCurrentEnv().getExtMetaCacheMgr().invalidateRowCountCache(table);
+ refreshTableInternal((ExternalDatabase) table.getDatabase(),
table, updateTime);
+ } catch (RuntimeException e) {
+ // The external transaction is already committed. Still notify
follower FEs even if a
+ // cache layer failed, so peers do not keep serving the
pre-commit state.
+ LOG.warn("Failed to refresh table cache after committing
external insert for {}",
+ table.getNameWithFullQualifiers(), e);
+ }
+ ExternalObjectLog log = ExternalObjectLog.createForRefreshTable(
+ table.getCatalog().getId(),
table.getDatabase().getFullName(), table.getName(), updateTime);
+ Env.getCurrentEnv().getEditLog().logRefreshExternalTable(log);
+ } finally {
+ ((ExternalCatalog) table.getCatalog()).endTableLookupRefresh();
}
- ExternalObjectLog log = ExternalObjectLog.createForRefreshTable(
- table.getCatalog().getId(), table.getDatabase().getFullName(),
table.getName(), updateTime);
- Env.getCurrentEnv().getEditLog().logRefreshExternalTable(log);
}
public void replayRefreshTable(ExternalObjectLog log) {
@@ -219,118 +239,123 @@ public class RefreshManager {
LOG.warn("failed to find catalog when replaying refresh table:
{}", log.debugForRefreshTable());
return;
}
- invalidateLanceTableAccess(catalog, log);
- Optional<ExternalDatabase<? extends ExternalTable>> db;
- if (!Strings.isNullOrEmpty(log.getDbName())) {
- db = catalog.getDbForReplay(log.getDbName());
- } else {
- db = catalog.getDbForReplay(log.getDbId());
- }
- // See comment in refreshDbInternal for why db and table may be
null.
- if (!db.isPresent()) {
- invalidateColdReplayDatabase(catalog, log);
- return;
- }
- if (!Strings.isNullOrEmpty(log.getNewTableName())) {
- // this is a rename table op
- db.get().unregisterTableForReplay(log.getTableName());
- if
(db.get().getTableForReplay(log.getNewTableName()).isPresent()
- || db.get().hasLocalTableName(log.getNewTableName())) {
- db.get().unregisterTableForReplay(log.getNewTableName());
- }
- db.get().resetMetaCacheNames();
- return;
- }
- Optional<? extends ExternalTable> table;
- if (!Strings.isNullOrEmpty(log.getTableName())) {
- table = db.get().getTableForReplay(log.getTableName());
- } else {
- table = db.get().getTableForReplay(log.getTableId());
- }
- if (log.isInvalidateTableLookup()) {
- if (table.isPresent()) {
- refreshTableInternal(db.get(), table.get(),
log.getLastUpdateTime(), true);
+ catalog.beginTableLookupRefresh();
+ try {
+ invalidateLanceTableAccess(catalog, log);
+ Optional<ExternalDatabase<? extends ExternalTable>> db;
+ if (!Strings.isNullOrEmpty(log.getDbName())) {
+ db = catalog.getDbForReplay(log.getDbName());
} else {
- try {
- db.get().invalidateTableLookup(log.getTableName());
- } finally {
- if (catalog.getLowerCaseTableNames() != 0) {
- db.get().invalidateTableLookupDependencies();
- } else {
-
Env.getCurrentEnv().getExtMetaCacheMgr().invalidateTableByNameOrWider(
- catalog.getId(), db.get().getFullName(),
log.getTableName());
- }
- }
+ db = catalog.getDbForReplay(log.getDbId());
}
- return;
- }
- if (!table.isPresent()) {
- LOG.warn("failed to find table when replaying refresh table:
{}", log.debugForRefreshTable());
- // Only a genuinely unresolved name (for example a lost
case-insensitive mapping)
- // needs the conservative database-wide retirement; an
ordinary cold miss for a
- // known name must not evict unrelated cached siblings.
- if (!Strings.isNullOrEmpty(log.getTableName())
- && !db.get().hasLocalTableName(log.getTableName())) {
- db.get().retireAllTableObjectsWithoutEngineInvalidation();
+ // See comment in refreshDbInternal for why db and table may
be null.
+ if (!db.isPresent()) {
+ invalidateColdReplayDatabase(catalog, log);
+ return;
}
- // The independent row-count cache can outlive the table
object, so fence the
- // canonical database scope before acknowledging the committed
refresh.
- Env.getCurrentEnv().getExtMetaCacheMgr()
- .invalidateRowCountCache(catalog.getId(),
db.get().getId());
+ if (!Strings.isNullOrEmpty(log.getNewTableName())) {
+ // this is a rename table op
+ db.get().unregisterTableForReplay(log.getTableName());
+ if
(db.get().getTableForReplay(log.getNewTableName()).isPresent()
+ ||
db.get().hasLocalTableName(log.getNewTableName())) {
+
db.get().unregisterTableForReplay(log.getNewTableName());
+ }
+ db.get().resetMetaCacheNames();
+ return;
+ }
+ Optional<? extends ExternalTable> table;
if (!Strings.isNullOrEmpty(log.getTableName())) {
-
Env.getCurrentEnv().getExtMetaCacheMgr().invalidateTableByNameOrWider(
- catalog.getId(), db.get().getFullName(),
log.getTableName());
+ table = db.get().getTableForReplay(log.getTableName());
} else {
- // Legacy ID-only records cannot recover a cold table's
engine key.
- Env.getCurrentEnv().getExtMetaCacheMgr().invalidateDb(
- catalog.getId(), db.get().getId(),
db.get().getFullName());
+ table = db.get().getTableForReplay(log.getTableId());
}
- invalidatePaimonCatalogForReplay(catalog);
- return;
- }
- List<String> modifiedPartNames = log.getPartitionNames();
- List<String> newPartNames = log.getNewPartitionNames();
- if (catalog instanceof HMSExternalCatalog
- && ((modifiedPartNames != null &&
!modifiedPartNames.isEmpty())
- || (newPartNames != null && !newPartNames.isEmpty()))) {
- // Partition-level cache invalidation, only for hive catalog.
Fence the held table
- // before the hive(...) lookup can lazily initialize the group
and throw; otherwise
- // a failed acquisition skips the row-count fence and the
full-table fallback.
-
Env.getCurrentEnv().getExtMetaCacheMgr().invalidateRowCountCache(table.get());
- try {
- HiveExternalMetaCache cache =
Env.getCurrentEnv().getExtMetaCacheMgr()
- .hive(catalog.getId());
- cache.refreshAffectedPartitionsCache((HMSExternalTable)
table.get(), modifiedPartNames,
- newPartNames);
- // The held HMS table also caches the pre-insert table
parameters.
- table.get().unsetObjectCreated();
- // Close the admission window the opening fence left open:
a load admitted after
- // it can publish the pre-insert value from the
still-resident file list.
+ if (log.isInvalidateTableLookup()) {
+ if (table.isPresent()) {
+ refreshTableInternal(db.get(), table.get(),
log.getLastUpdateTime(), true);
+ } else {
+ try {
+ db.get().invalidateTableLookup(log.getTableName());
+ } finally {
+ if (catalog.getLowerCaseTableNames() != 0) {
+ db.get().invalidateTableLookupDependencies();
+ } else {
+
Env.getCurrentEnv().getExtMetaCacheMgr().invalidateTableByNameOrWider(
+ catalog.getId(),
db.get().getFullName(), log.getTableName());
+ }
+ }
+ }
+ return;
+ }
+ if (!table.isPresent()) {
+ LOG.warn("failed to find table when replaying refresh
table: {}", log.debugForRefreshTable());
+ // Only a genuinely unresolved name (for example a lost
case-insensitive mapping)
+ // needs the conservative database-wide retirement; an
ordinary cold miss for a
+ // known name must not evict unrelated cached siblings.
+ if (!Strings.isNullOrEmpty(log.getTableName())
+ &&
!db.get().hasLocalTableName(log.getTableName())) {
+
db.get().retireAllTableObjectsWithoutEngineInvalidation();
+ }
+ // The independent row-count cache can outlive the table
object, so fence the
+ // canonical database scope before acknowledging the
committed refresh.
+ Env.getCurrentEnv().getExtMetaCacheMgr()
+ .invalidateRowCountCache(catalog.getId(),
db.get().getId());
+ if (!Strings.isNullOrEmpty(log.getTableName())) {
+
Env.getCurrentEnv().getExtMetaCacheMgr().invalidateTableByNameOrWider(
+ catalog.getId(), db.get().getFullName(),
log.getTableName());
+ } else {
+ // Legacy ID-only records cannot recover a cold
table's engine key.
+ Env.getCurrentEnv().getExtMetaCacheMgr().invalidateDb(
+ catalog.getId(), db.get().getId(),
db.get().getFullName());
+ }
+ invalidatePaimonCatalogForReplay(catalog);
+ return;
+ }
+ List<String> modifiedPartNames = log.getPartitionNames();
+ List<String> newPartNames = log.getNewPartitionNames();
+ if (catalog instanceof HMSExternalCatalog
+ && ((modifiedPartNames != null &&
!modifiedPartNames.isEmpty())
+ || (newPartNames != null && !newPartNames.isEmpty())))
{
+ // Partition-level cache invalidation, only for hive
catalog. Fence the held table
+ // before the hive(...) lookup can lazily initialize the
group and throw; otherwise
+ // a failed acquisition skips the row-count fence and the
full-table fallback.
Env.getCurrentEnv().getExtMetaCacheMgr().invalidateRowCountCache(table.get());
- } catch (RuntimeException e) {
- // The insert is already committed. A partial or failed
selective refresh must not
- // leave stale Hive partition/file entries, so fall back
to the same conservative
- // full-table invalidation the leader uses.
- LOG.warn("failed to refresh affected partitions when
replaying refresh table for {}",
- table.get().getNameWithFullQualifiers(), e);
try {
-
Env.getCurrentEnv().getExtMetaCacheMgr().invalidateTableCache(table.get());
- } finally {
+ HiveExternalMetaCache cache =
Env.getCurrentEnv().getExtMetaCacheMgr()
+ .hive(catalog.getId());
+
cache.refreshAffectedPartitionsCache((HMSExternalTable) table.get(),
modifiedPartNames,
+ newPartNames);
+ // The held HMS table also caches the pre-insert table
parameters.
table.get().unsetObjectCreated();
+ // Close the admission window the opening fence left
open: a load admitted after
+ // it can publish the pre-insert value from the
still-resident file list.
Env.getCurrentEnv().getExtMetaCacheMgr().invalidateRowCountCache(table.get());
+ } catch (RuntimeException e) {
+ // The insert is already committed. A partial or
failed selective refresh must not
+ // leave stale Hive partition/file entries, so fall
back to the same conservative
+ // full-table invalidation the leader uses.
+ LOG.warn("failed to refresh affected partitions when
replaying refresh table for {}",
+ table.get().getNameWithFullQualifiers(), e);
+ try {
+
Env.getCurrentEnv().getExtMetaCacheMgr().invalidateTableCache(table.get());
+ } finally {
+ table.get().unsetObjectCreated();
+
Env.getCurrentEnv().getExtMetaCacheMgr().invalidateRowCountCache(table.get());
+ }
}
+ if (table.get() instanceof HMSExternalTable &&
log.getLastUpdateTime() > 0) {
+ ((HMSExternalTable)
table.get()).setUpdateTime(log.getLastUpdateTime());
+ }
+ LOG.info("replay refresh partitions for table {}, "
+ + "modified partitions count: {}, "
+ + "new partitions count: {}",
+ table.get().getName(), modifiedPartNames == null ?
0 : modifiedPartNames.size(),
+ newPartNames == null ? 0 : newPartNames.size());
+ } else {
+ // Full table cache invalidation
+ refreshTableInternal(db.get(), table.get(),
log.getLastUpdateTime());
}
- if (table.get() instanceof HMSExternalTable &&
log.getLastUpdateTime() > 0) {
- ((HMSExternalTable)
table.get()).setUpdateTime(log.getLastUpdateTime());
- }
- LOG.info("replay refresh partitions for table {}, "
- + "modified partitions count: {}, "
- + "new partitions count: {}",
- table.get().getName(), modifiedPartNames == null ? 0 :
modifiedPartNames.size(),
- newPartNames == null ? 0 : newPartNames.size());
- } else {
- // Full table cache invalidation
- refreshTableInternal(db.get(), table.get(),
log.getLastUpdateTime());
+ } finally {
+ catalog.endTableLookupRefresh();
}
});
}
@@ -348,25 +373,30 @@ public class RefreshManager {
&& ((HMSExternalCatalog)
catalog).isPartitionEventTargetExcluded(dbName, tableName)) {
return;
}
- // Whole-table events are already committed remotely. Fence the row
count by cached identity
- // before any database/table reload can fail and make the not-found
path return.
- Env.getCurrentEnv().getExtMetaCacheMgr()
- .invalidateRowCountCache(catalog.getId(), dbName, tableName);
- DatabaseIf db = catalog.getDbNullable(dbName);
- if (db == null) {
- // Cold database: widen so independently resident engine entries
are retired too.
+ ((ExternalCatalog) catalog).beginTableLookupRefresh();
+ try {
+ // Whole-table events are already committed remotely. Fence the
row count by cached identity
+ // before any database/table reload can fail and make the
not-found path return.
Env.getCurrentEnv().getExtMetaCacheMgr()
- .invalidateTableByNameOrWider(catalog.getId(), dbName,
tableName);
- return;
- }
+ .invalidateRowCountCache(catalog.getId(), dbName,
tableName);
+ DatabaseIf db = catalog.getDbNullable(dbName);
+ if (db == null) {
+ // Cold database: widen so independently resident engine
entries are retired too.
+ Env.getCurrentEnv().getExtMetaCacheMgr()
+ .invalidateTableByNameOrWider(catalog.getId(), dbName,
tableName);
+ return;
+ }
- TableIf table = db.getTableNullable(tableName);
- if (table == null) {
- Env.getCurrentEnv().getExtMetaCacheMgr()
- .invalidateTableByNameOrWider(catalog.getId(), dbName,
tableName);
- return;
+ TableIf table = db.getTableNullable(tableName);
+ if (table == null) {
+ Env.getCurrentEnv().getExtMetaCacheMgr()
+ .invalidateTableByNameOrWider(catalog.getId(), dbName,
tableName);
+ return;
+ }
+ refreshTableInternal((ExternalDatabase) db, (ExternalTable) table,
updateTime);
+ } finally {
+ ((ExternalCatalog) catalog).endTableLookupRefresh();
}
- refreshTableInternal((ExternalDatabase) db, (ExternalTable) table,
updateTime);
}
public void refreshTableInternal(ExternalDatabase db, ExternalTable table,
long updateTime) {
@@ -484,16 +514,21 @@ public class RefreshManager {
&& ((HMSExternalCatalog)
catalog).isPartitionEventTargetExcluded(dbName, tableName)) {
return;
}
- // Partition events are already committed remotely. Fence the row
count by cached identity
- // before any database/table reload can fail and make the
ignored-not-found path return.
- Env.getCurrentEnv().getExtMetaCacheMgr()
- .invalidateRowCountCache(catalog.getId(), dbName, tableName);
+ ((ExternalCatalog) catalog).beginTableLookupRefresh();
try {
- refreshPartitionsAfterFence(catalog, dbName, tableName,
partitionNames, updateTime,
- ignoreIfNotExists);
- } finally {
+ // Partition events are already committed remotely. Fence the row
count by cached identity
+ // before any database/table reload can fail and make the
ignored-not-found path return.
Env.getCurrentEnv().getExtMetaCacheMgr()
.invalidateRowCountCache(catalog.getId(), dbName,
tableName);
+ try {
+ refreshPartitionsAfterFence(catalog, dbName, tableName,
partitionNames, updateTime,
+ ignoreIfNotExists);
+ } finally {
+ Env.getCurrentEnv().getExtMetaCacheMgr()
+ .invalidateRowCountCache(catalog.getId(), dbName,
tableName);
+ }
+ } finally {
+ ((ExternalCatalog) catalog).endTableLookupRefresh();
}
}
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 342e0ca56d3..204574d3595 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
@@ -169,6 +169,10 @@ public abstract class ExternalCatalog
@SerializedName(value = "initialized")
protected boolean initialized = false;
private AtomicLong metadataLoadEpoch = new AtomicLong();
+ // Catalog-owned so refresh protection survives normal database/table
object eviction.
+ private transient volatile long tableLookupRefreshGeneration;
+ private transient int activeTableLookupRefreshes;
+
@SerializedName(value = "lastUpdateTime")
protected long lastUpdateTime;
// <db name, table name> to tableAutoAnalyzePolicy
@@ -455,6 +459,25 @@ public abstract class ExternalCatalog
}
}
+ public synchronized void beginTableLookupRefresh() {
+ activeTableLookupRefreshes++;
+ tableLookupRefreshGeneration++;
+ }
+
+ public synchronized void endTableLookupRefresh() {
+ activeTableLookupRefreshes--;
+ tableLookupRefreshGeneration++;
+ }
+
+ synchronized OptionalLong observeTableLookupRefreshGeneration() {
+ return activeTableLookupRefreshes == 0
+ ? OptionalLong.of(tableLookupRefreshGeneration) :
OptionalLong.empty();
+ }
+
+ long getTableLookupRefreshGeneration() {
+ return tableLookupRefreshGeneration;
+ }
+
protected long acquireMetadataLoadEpoch() {
synchronized (this) {
makeSureInitialized();
@@ -708,32 +731,37 @@ public abstract class ExternalCatalog
* and reloaded during the refresh process.
*/
public void resetToUninitialized(boolean invalidCache) {
- MetaCache<ExternalDatabase<? extends ExternalTable>> cacheToInvalidate
= null;
- Runnable objectInvalidation = null;
+ beginTableLookupRefresh();
try {
- synchronized (this) {
- metadataLoadEpoch.incrementAndGet();
- this.objectCreated = false;
- this.initialized = false;
- synchronized (this.confLock) {
- this.cachedConf = null;
+ MetaCache<ExternalDatabase<? extends ExternalTable>>
cacheToInvalidate = null;
+ Runnable objectInvalidation = null;
+ try {
+ synchronized (this) {
+ metadataLoadEpoch.incrementAndGet();
+ this.objectCreated = false;
+ this.initialized = false;
+ synchronized (this.confLock) {
+ this.cachedConf = null;
+ }
+ this.lowerCaseToDatabaseName.clear();
+ cacheToInvalidate = metaCache;
+ if (cacheToInvalidate != null) {
+ cacheToInvalidate.invalidateNames();
+ objectInvalidation = cacheToInvalidate.retireObjects();
+ }
+ onClose();
}
- this.lowerCaseToDatabaseName.clear();
- cacheToInvalidate = metaCache;
- if (cacheToInvalidate != null) {
- cacheToInvalidate.invalidateNames();
- objectInvalidation = cacheToInvalidate.retireObjects();
+ } finally {
+ if (objectInvalidation != null) {
+ objectInvalidation.run();
}
- onClose();
}
- } finally {
- if (objectInvalidation != null) {
- objectInvalidation.run();
+ setLastUpdateTime(System.currentTimeMillis());
+ if (invalidCache) {
+ Env.getCurrentEnv().getExtMetaCacheMgr().invalidateCatalog(id);
}
- }
- setLastUpdateTime(System.currentTimeMillis());
- if (invalidCache) {
- Env.getCurrentEnv().getExtMetaCacheMgr().invalidateCatalog(id);
+ } finally {
+ endTableLookupRefresh();
}
}
@@ -743,13 +771,18 @@ public abstract class ExternalCatalog
* @param invalidCache
*/
public void onRefreshCache(boolean invalidCache) {
- setLastUpdateTime(System.currentTimeMillis());
+ beginTableLookupRefresh();
try {
- refreshMetaCacheOnly(invalidCache);
- } finally {
- if (invalidCache) {
- Env.getCurrentEnv().getExtMetaCacheMgr().invalidateCatalog(id);
+ setLastUpdateTime(System.currentTimeMillis());
+ try {
+ refreshMetaCacheOnly(invalidCache);
+ } finally {
+ if (invalidCache) {
+
Env.getCurrentEnv().getExtMetaCacheMgr().invalidateCatalog(id);
+ }
}
+ } finally {
+ endTableLookupRefresh();
}
}
@@ -1139,6 +1172,8 @@ public abstract class ExternalCatalog
public void gsonPostProcess() throws IOException {
objectCreated = false;
metadataLoadEpoch = new AtomicLong();
+ tableLookupRefreshGeneration = 0;
+ activeTableLookupRefreshes = 0;
invalidateEngineCacheOnDatabaseRemoval = ThreadLocal.withInitial(() ->
true);
invalidatingAllMetaCache = ThreadLocal.withInitial(() -> false);
// TODO: This code is to compatible with older version of metadata.
@@ -1677,8 +1712,13 @@ public abstract class ExternalCatalog
}
public void replayTruncateTable(TruncateTableInfo info) {
- if (metadataOps != null) {
- metadataOps.afterTruncateTable(info.getDb(), info.getTable(),
info.getUpdateTime());
+ beginTableLookupRefresh();
+ try {
+ if (metadataOps != null) {
+ metadataOps.afterTruncateTable(info.getDb(), info.getTable(),
info.getUpdateTime());
+ }
+ } finally {
+ endTableLookupRefresh();
}
}
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 5d69dee0bb2..1d9a23e0db8 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
@@ -732,14 +732,16 @@ public abstract class ExternalDatabase<T extends
ExternalTable>
}
public Optional<Runnable> beginTableMetadataRefresh() {
- return metaCache == null ? Optional.empty() :
Optional.of(metaCache.beginMetadataRefresh());
+ extCatalog.beginTableLookupRefresh();
+ return Optional.of(extCatalog::endTableLookupRefresh);
}
/** A missing result is authorized only against the retained DB, names and
object generations. */
@SuppressWarnings("unchecked")
public Optional<Runnable> observeTableLookup(ExternalTable table) {
+ OptionalLong refreshGeneration =
extCatalog.observeTableLookupRefreshGeneration();
MetaCache<T> cache = metaCache;
- if (!initialized || cache == null
+ if (!refreshGeneration.isPresent() || !initialized || cache == null
||
extCatalog.shouldBypassTableNameCache(SessionContext.current())) {
return Optional.empty();
}
@@ -764,7 +766,8 @@ public abstract class ExternalDatabase<T extends
ExternalTable>
extCatalog.executeIfDatabaseCurrent(this, () -> {
synchronized (extCatalog) {
synchronized (this) {
- if (!initialized || metaCache != cache ||
!isTableMetadataLoadEpochCurrent(epoch)
+ if (extCatalog.getTableLookupRefreshGeneration() !=
refreshGeneration.getAsLong()
+ || !initialized || metaCache != cache ||
!isTableMetadataLoadEpochCurrent(epoch)
||
Env.getCurrentEnv().getCatalogMgr().getCatalog(extCatalog.getId()) !=
extCatalog) {
return false;
}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/datasource/ExternalLookupRefreshTest.java
b/fe/fe-core/src/test/java/org/apache/doris/datasource/ExternalLookupRefreshTest.java
index a115765bce4..778f433765d 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/datasource/ExternalLookupRefreshTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/datasource/ExternalLookupRefreshTest.java
@@ -33,6 +33,7 @@ import org.mockito.MockedStatic;
import org.mockito.Mockito;
import java.util.Optional;
+import java.util.OptionalLong;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicBoolean;
@@ -40,9 +41,213 @@ import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
class ExternalLookupRefreshTest {
+ @Test
+ void
missingReadStartedDuringRefreshCannotRetireReplacementAfterRefreshFinishes() {
+ ExternalCatalog catalog = lookupRefreshCatalog();
+ Runnable finished =
lookupDatabase(catalog).beginTableMetadataRefresh().get();
+ ExternalDatabase<?> replacement = lookupDatabase(catalog);
+ ExternalTable table = new ExternalTable();
+ table.name = "t";
+ table.setDb(replacement);
+ table.setRemoteName("t");
+ @SuppressWarnings("unchecked")
+ MetaCache<ExternalTable> cache = Deencapsulation.getField(replacement,
"metaCache");
+ AtomicInteger repairs = new AtomicInteger();
+ Mockito.doReturn(Optional.of((Runnable)
repairs::incrementAndGet)).when(cache).observeLookup(
+ Mockito.eq("t"), Mockito.eq("t"), Mockito.same(table),
Mockito.eq(false), Mockito.any());
+ Mockito.doAnswer(call -> ((java.util.function.BooleanSupplier)
call.getArgument(1)).getAsBoolean())
+
.when(catalog).executeIfDatabaseCurrent(Mockito.same(replacement),
Mockito.any());
+ Env env = Mockito.mock(Env.class);
+ CatalogMgr manager = Mockito.mock(CatalogMgr.class);
+ Mockito.when(env.getCatalogMgr()).thenReturn(manager);
+ Mockito.doReturn(catalog).when(manager).getCatalog(0L);
+ RuntimeException missing = new
org.apache.iceberg.exceptions.NoSuchTableException("old query missed t");
+ try (MockedStatic<Env> mocked = Mockito.mockStatic(Env.class)) {
+ mocked.when(Env::getCurrentEnv).thenReturn(env);
+ Assertions.assertSame(missing,
Assertions.assertThrows(RuntimeException.class,
+ () -> table.withMissingTableRepair(() -> {
+ // The remote response arrives after the refresh has
finished. This query
+ // began on an object admitted while the old database
was being refreshed.
+ finished.run();
+ throw missing;
+ })));
+ Assertions.assertEquals(0, repairs.get());
+ Assertions.assertSame(missing,
Assertions.assertThrows(RuntimeException.class,
+ () -> table.withMissingTableRepair(() -> {
+ throw missing;
+ })));
+ Assertions.assertEquals(1, repairs.get());
+ }
+ }
+
+ @Test
+ void refreshCommandsEventsAndReplayFenceColdDatabaseResolution() throws
Exception {
+ for (int route = 0; route < 6; route++) {
+ ExternalCatalog catalog = lookupRefreshCatalog();
+ CatalogMgr manager = Mockito.mock(CatalogMgr.class);
+ Mockito.doReturn(catalog).when(manager).getCatalog("catalog");
+ Mockito.doReturn(catalog).when(manager).getCatalog(0L);
+ Env env = Mockito.mock(Env.class);
+ Mockito.when(env.getCatalogMgr()).thenReturn(manager);
+ ExternalMetaCacheMgr caches =
Mockito.mock(ExternalMetaCacheMgr.class);
+ Mockito.when(env.getExtMetaCacheMgr()).thenReturn(caches);
+ org.mockito.stubbing.Answer<Object> duringResolution = call -> {
+ Assertions.assertFalse(Thread.holdsLock(catalog));
+
Assertions.assertFalse(catalog.observeTableLookupRefreshGeneration().isPresent());
+ // Neither normal cache eviction nor the cold-cache path can
reset the root fence.
+
Assertions.assertFalse(lookupDatabase(catalog).observeTableLookup(new
ExternalTable()).isPresent());
+ throw new IllegalStateException("database resolution failed");
+ };
+
Mockito.doAnswer(duringResolution).when(catalog).getDbNullable("db");
+
Mockito.doAnswer(duringResolution).when(catalog).getDbOrDdlException("db");
+
Mockito.doAnswer(duringResolution).when(catalog).getDbForReplay("db");
+ RefreshManager refresh = new RefreshManager();
+ final int selected = route;
+ org.junit.jupiter.api.function.Executable run = () -> {
+ switch (selected) {
+ case 0:
+ refresh.handleRefreshDb("catalog", "db");
+ break;
+ case 1:
+ refresh.handleRefreshTable("catalog", "db", "t",
false);
+ break;
+ case 2:
+ refresh.refreshExternalTableFromEvent("catalog", "db",
"t", 1L);
+ break;
+ case 3:
+ refresh.refreshPartitions("catalog", "db", "t",
java.util.Collections.emptyList(), 1L, true);
+ break;
+ case 4:
+
refresh.replayRefreshDb(ExternalObjectLog.createForRefreshDb(catalog.getId(),
"db"));
+ break;
+ default:
+
refresh.replayRefreshTable(ExternalObjectLog.createForRefreshTable(
+ catalog.getId(), "db", "t", 1L, true));
+ break;
+ }
+ };
+ try (MockedStatic<Env> mocked = Mockito.mockStatic(Env.class)) {
+ mocked.when(Env::getCurrentEnv).thenReturn(env);
+ if (route < 4) {
+ Assertions.assertThrows(IllegalStateException.class, run);
+ } else {
+ Assertions.assertDoesNotThrow(run);
+ }
+
Assertions.assertTrue(catalog.observeTableLookupRefreshGeneration().isPresent());
+ Assertions.assertEquals(2,
catalog.getTableLookupRefreshGeneration());
+ }
+ }
+ }
+
+ @Test
+ void catalogRefreshProtectsReplacementDatabaseAndOverlappingIntervals() {
+ ExternalCatalog catalog = lookupRefreshCatalog();
+ ExternalDatabase<?> oldDatabase = lookupDatabase(catalog);
+ Runnable finishFirst = oldDatabase.beginTableMetadataRefresh().get();
+ // Normal object-cache eviction admits a replacement with a new
MetaCache. It must still
+ // inherit the catalog's active refresh interval instead of
authorizing a delayed absence.
+ ExternalDatabase<?> replacement = lookupDatabase(catalog);
+ ExternalTable table = Mockito.mock(ExternalTable.class);
+
Assertions.assertFalse(replacement.observeTableLookup(table).isPresent());
+ Runnable finishSecond = replacement.beginTableMetadataRefresh().get();
+ finishFirst.run();
+
Assertions.assertFalse(catalog.observeTableLookupRefreshGeneration().isPresent());
+
Assertions.assertFalse(replacement.observeTableLookup(table).isPresent());
+ finishSecond.run();
+
Assertions.assertTrue(catalog.observeTableLookupRefreshGeneration().isPresent());
+ Assertions.assertEquals(4, catalog.getTableLookupRefreshGeneration());
+ }
+
+ @Test
+ void oldMissingResultCannotCleanRecreatedTableAfterRefresh() {
+ ExternalCatalog catalog = lookupRefreshCatalog();
+ ExternalDatabase<?> database = lookupDatabase(catalog);
+ ExternalTable table = Mockito.mock(ExternalTable.class);
+ Mockito.when(table.getName()).thenReturn("t");
+ Mockito.when(table.getRemoteName()).thenReturn("t");
+ @SuppressWarnings("unchecked")
+ MetaCache<ExternalTable> cache = Deencapsulation.getField(database,
"metaCache");
+ AtomicInteger repairs = new AtomicInteger();
+ Mockito.doReturn(Optional.of((Runnable)
repairs::incrementAndGet)).when(cache).observeLookup(
+ Mockito.eq("t"), Mockito.eq("t"), Mockito.same(table),
Mockito.eq(false), Mockito.any());
+ Mockito.doAnswer(call -> ((java.util.function.BooleanSupplier)
call.getArgument(1)).getAsBoolean())
+
.when(catalog).executeIfDatabaseCurrent(Mockito.same(database), Mockito.any());
+ Env env = Mockito.mock(Env.class);
+ CatalogMgr manager = Mockito.mock(CatalogMgr.class);
+ Mockito.when(env.getCatalogMgr()).thenReturn(manager);
+ Mockito.doReturn(catalog).when(manager).getCatalog(0L);
+ try (MockedStatic<Env> mocked = Mockito.mockStatic(Env.class)) {
+ mocked.when(Env::getCurrentEnv).thenReturn(env);
+ Runnable oldMissing = database.observeTableLookup(table).get();
+ catalog.beginTableLookupRefresh();
+
Assertions.assertFalse(database.observeTableLookup(table).isPresent());
+ catalog.endTableLookupRefresh();
+ oldMissing.run();
+ Assertions.assertEquals(0, repairs.get());
+ // A new observation remains eligible, so refresh does not
permanently disable repair.
+ database.observeTableLookup(table).get().run();
+ Assertions.assertEquals(1, repairs.get());
+ }
+ }
+
+ @Test
+ void catalogRefreshAndTruncateReplayReleaseGuardOnFailure() {
+ ExternalCatalog catalog = lookupRefreshCatalog();
+ ExternalMetaCacheMgr caches = Mockito.mock(ExternalMetaCacheMgr.class);
+ Env env = Mockito.mock(Env.class);
+ Mockito.when(env.getExtMetaCacheMgr()).thenReturn(caches);
+
Mockito.doCallRealMethod().when(catalog).onRefreshCache(Mockito.anyBoolean());
+ Mockito.doAnswer(call -> {
+ Assertions.assertFalse(Thread.holdsLock(catalog));
+
Assertions.assertFalse(catalog.observeTableLookupRefreshGeneration().isPresent());
+
Assertions.assertFalse(lookupDatabase(catalog).observeTableLookup(new
ExternalTable()).isPresent());
+ throw new IllegalStateException("invalidation failed");
+ }).when(caches).invalidateCatalog(0L);
+ try (MockedStatic<Env> mocked = Mockito.mockStatic(Env.class)) {
+ mocked.when(Env::getCurrentEnv).thenReturn(env);
+ Assertions.assertThrows(IllegalStateException.class, () ->
catalog.onRefreshCache(true));
+
Assertions.assertTrue(catalog.observeTableLookupRefreshGeneration().isPresent());
+ catalog.onRefreshCache(false);
+
Assertions.assertTrue(catalog.observeTableLookupRefreshGeneration().isPresent());
+ }
+ org.apache.doris.datasource.operations.ExternalMetadataOps ops =
+
Mockito.mock(org.apache.doris.datasource.operations.ExternalMetadataOps.class);
+ Deencapsulation.setField(catalog, "metadataOps", ops);
+
Mockito.doCallRealMethod().when(catalog).replayTruncateTable(Mockito.any());
+ Mockito.doAnswer(call -> {
+
Assertions.assertFalse(catalog.observeTableLookupRefreshGeneration().isPresent());
+ throw new IllegalStateException("truncate invalidation failed");
+ }).when(ops).afterTruncateTable(Mockito.anyString(),
Mockito.anyString(), Mockito.anyLong());
+ org.apache.doris.persist.TruncateTableInfo info = Mockito.mock(
+ org.apache.doris.persist.TruncateTableInfo.class);
+ Mockito.when(info.getDb()).thenReturn("db");
+ Mockito.when(info.getTable()).thenReturn("t");
+ Assertions.assertThrows(IllegalStateException.class, () ->
catalog.replayTruncateTable(info));
+
Assertions.assertTrue(catalog.observeTableLookupRefreshGeneration().isPresent());
+ }
+
+ private ExternalCatalog lookupRefreshCatalog() {
+ ExternalCatalog catalog = Mockito.mock(ExternalCatalog.class);
+ Mockito.doCallRealMethod().when(catalog).beginTableLookupRefresh();
+ Mockito.doCallRealMethod().when(catalog).endTableLookupRefresh();
+
Mockito.when(catalog.observeTableLookupRefreshGeneration()).thenCallRealMethod();
+
Mockito.when(catalog.getTableLookupRefreshGeneration()).thenCallRealMethod();
+
Mockito.when(catalog.isMetadataLoadEpochCurrent(Mockito.anyLong())).thenReturn(true);
+ return catalog;
+ }
+
+ private ExternalDatabase<?> lookupDatabase(ExternalCatalog catalog) {
+ ExternalDatabase<?> database = new
org.apache.doris.datasource.test.TestExternalDatabase(
+ catalog, 1L, "db", "db");
+ Deencapsulation.setField(database, "metaCache",
Mockito.mock(MetaCache.class));
+ Deencapsulation.setField(database, "initialized", true);
+ return database;
+ }
+
@Test
void sessionBypassDoesNotObserveSharedLookup() {
ExternalCatalog catalog = Mockito.mock(ExternalCatalog.class);
+
Mockito.when(catalog.observeTableLookupRefreshGeneration()).thenReturn(OptionalLong.of(0));
Mockito.when(catalog.shouldBypassTableNameCache(Mockito.any())).thenReturn(true);
ExternalDatabase<?> database = Mockito.mock(ExternalDatabase.class,
Mockito.CALLS_REAL_METHODS);
MetaCache<?> cache = Mockito.mock(MetaCache.class);
@@ -108,6 +313,7 @@ class ExternalLookupRefreshTest {
Deencapsulation.setField(database, "metaCache", cache);
Deencapsulation.setField(database, "initialized", true);
Mockito.when(catalog.isMetadataLoadEpochCurrent(Mockito.anyLong())).thenReturn(true);
+
Mockito.when(catalog.observeTableLookupRefreshGeneration()).thenReturn(OptionalLong.of(0));
Mockito.doReturn(Optional.of((Runnable) () -> {
})).when(cache).observeLookup(
Mockito.eq("t"), Mockito.eq("t"), Mockito.same(table),
Mockito.eq(false), Mockito.any());
Mockito.doAnswer(call -> {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
