From 8f7d4d61cae09e14a636dd16d90eafe9bb24687b Mon Sep 17 00:00:00 2001 From: 924060929 Date: Fri, 23 Oct 2026 11:59:58 +0800 Subject: [PATCH 1/8] [fix](fe) Refresh external table lookup on explicit refresh ### What problem does this PR solve? Issue Number: None Problem Summary: Explicit REFRESH TABLE invalidated the connector metadata but kept the external database's cached table-name snapshot, table object, and ID lookup. After a remote table was dropped or recreated, SHOW TABLES and later lookups could keep using the old identity until cache expiry. Invalidate that lookup state for explicit refresh on the leader and during follower replay, including cold replay paths. In case-insensitive mode, invalidate the database scope so an old spelling cannot retain a stale object. Keep content-only refreshes unchanged and treat old edit logs without the new flag as content-only refreshes. ### Release note Explicit REFRESH TABLE now reloads external table names and table lookup identity. ### Check List (For Author) - Test: Unit Test - ExternalDatabaseTest and RefreshManagerTest: 52 tests passed. - FE build and Checkstyle passed. - Behavior changed: Yes. Explicit REFRESH TABLE reloads external table lookup state. - Does this need documentation: No --- .../apache/doris/catalog/RefreshManager.java | 47 +++++-- .../doris/datasource/ExternalDatabase.java | 19 +++ .../datasource/log/ExternalObjectLog.java | 10 ++ .../doris/catalog/RefreshManagerTest.java | 120 ++++++++++++++++++ .../datasource/ExternalDatabaseTest.java | 47 +++++++ 5 files changed, 235 insertions(+), 8 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 8e5586b344f79c..7964310f1c16c5 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 @@ -170,8 +170,8 @@ public void handleRefreshTable(String catalogName, String dbName, String tableNa return; } long updateTime = System.currentTimeMillis(); - refreshTableInternal((ExternalDatabase) db, (ExternalTable) table, updateTime); - ExternalObjectLog log = ExternalObjectLog.createForRefreshTable( + refreshTableInternal((ExternalDatabase) db, (ExternalTable) table, updateTime, true); + ExternalObjectLog log = ExternalObjectLog.createForExplicitRefreshTable( catalog.getId(), db.getFullName(), table.getName(), updateTime); env.getEditLog().logRefreshExternalTable(log); } @@ -209,8 +209,12 @@ public void replayRefreshTable(ExternalObjectLog log) { try { invalidateAllConnectorCachesIfPresent(catalog); } finally { - Env.getCurrentEnv().getExtMetaCacheMgr().invalidateTable( - log.getCatalogId(), dbId, localDbName, tableId, localTableName); + if (log.isInvalidateTableLookup() && catalog.getLowerCaseTableNames() == 2) { + Env.getCurrentEnv().getExtMetaCacheMgr().invalidateDb(log.getCatalogId(), dbId, localDbName); + } else { + Env.getCurrentEnv().getExtMetaCacheMgr().invalidateTable( + log.getCatalogId(), dbId, localDbName, tableId, localTableName); + } } LOG.info("database object cache is cold when replaying refresh table; " + "invalidated caches by local names {}.{}: {}", @@ -225,14 +229,25 @@ public void replayRefreshTable(ExternalObjectLog log) { ((PluginDrivenExternalCatalog) catalog).getConnector().invalidateDb(db.get().getRemoteName()); } } finally { - Env.getCurrentEnv().getExtMetaCacheMgr().invalidateTable( - log.getCatalogId(), dbId, db.get().getFullName(), tableId, localTableName); + try { + if (log.isInvalidateTableLookup()) { + db.get().invalidateTableLookup(localTableName); + } + } finally { + if (log.isInvalidateTableLookup() && catalog.getLowerCaseTableNames() == 2) { + Env.getCurrentEnv().getExtMetaCacheMgr().invalidateDb( + log.getCatalogId(), dbId, db.get().getFullName()); + } else { + Env.getCurrentEnv().getExtMetaCacheMgr().invalidateTable( + log.getCatalogId(), dbId, db.get().getFullName(), tableId, localTableName); + } + } } LOG.info("table object cache is cold when replaying refresh table; invalidated caches by local name {}: {}", localTableName, log.debugForRefreshTable()); return; } - refreshTableInternal(db.get(), table.get(), log.getLastUpdateTime()); + refreshTableInternal(db.get(), table.get(), log.getLastUpdateTime(), log.isInvalidateTableLookup()); } private void replayRenameTable(ExternalObjectLog log, ExternalCatalog catalog, @@ -301,6 +316,11 @@ public void refreshExternalTableFromEvent(String catalogName, String localDbName } public void refreshTableInternal(ExternalDatabase db, ExternalTable table, long updateTime) { + refreshTableInternal(db, table, updateTime, false); + } + + private void refreshTableInternal(ExternalDatabase db, ExternalTable table, long updateTime, + boolean invalidateTableLookup) { table.unsetObjectCreated(); if (updateTime > 0) { table.setUpdateTime(updateTime); @@ -312,7 +332,18 @@ public void refreshTableInternal(ExternalDatabase db, ExternalTable table, long .invalidateTable(db.getRemoteName(), table.getRemoteName()); } } finally { - Env.getCurrentEnv().getExtMetaCacheMgr().invalidateTable(table); + try { + if (invalidateTableLookup) { + db.invalidateTableLookup(table.getName()); + } + } finally { + if (invalidateTableLookup && table.getCatalog().getLowerCaseTableNames() == 2) { + Env.getCurrentEnv().getExtMetaCacheMgr().invalidateDb( + table.getCatalog().getId(), db.getId(), db.getFullName()); + } else { + Env.getCurrentEnv().getExtMetaCacheMgr().invalidateTable(table); + } + } } LOG.info("refresh table {}, id {} from db {} in catalog {}, update time: {}", table.getName(), table.getId(), db.getFullName(), db.getCatalog().getName(), updateTime); 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 3d24d0c3b78ce9..62ee38684e88f0 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 @@ -743,6 +743,25 @@ protected void invalidateTableCache(String localTableName) { } } + /** Refreshes table membership and lookup identity without loading remote metadata. */ + public void invalidateTableLookup(String tableName) { + if (!isInitialized()) { + return; + } + // In case-insensitive lookup mode an older spelling may still be loading while a refreshed + // names snapshot already contains the new spelling. Fence every object/ID load in this database. + if (isTableNamesCaseInsensitive()) { + invalidateAllTableCache(); + return; + } + String localTableName = resolveTableNameForInvalidation(tableName); + try { + invalidateTableCache(localTableName); + } finally { + resetMetaCacheNames(); + } + } + private void invalidateAllTableCache() { if (tableNames != null) { tableNames.invalidateAll(); diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/log/ExternalObjectLog.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/log/ExternalObjectLog.java index 30464457e60a80..308b3cd5184100 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/datasource/log/ExternalObjectLog.java +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/log/ExternalObjectLog.java @@ -61,6 +61,9 @@ public class ExternalObjectLog implements Writable { @SerializedName(value = "newPartitionNames") private List newPartitionNames; + @SerializedName(value = "invalidateTableLookup") + private boolean invalidateTableLookup; + @SerializedName(value = "lastUpdateTime") private long lastUpdateTime; @@ -85,6 +88,13 @@ public static ExternalObjectLog createForRefreshTable(long catalogId, String dbN return externalObjectLog; } + public static ExternalObjectLog createForExplicitRefreshTable(long catalogId, String dbName, String tblName, + long updateTime) { + ExternalObjectLog log = createForRefreshTable(catalogId, dbName, tblName, updateTime); + log.setInvalidateTableLookup(true); + return log; + } + public static ExternalObjectLog createForRefreshPartitions(long catalogId, String dbName, String tblName, List modifiedPartNames, List newPartNames, long updateTime) { ExternalObjectLog externalObjectLog = new ExternalObjectLog(); 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 index 44059931620e82..e0e79fdbfc268f 100644 --- 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 @@ -40,6 +40,7 @@ import org.apache.doris.datasource.test.TestExternalDatabase; import org.apache.doris.datasource.test.TestExternalTable; import org.apache.doris.persist.EditLog; +import org.apache.doris.persist.gson.GsonUtils; import com.google.common.util.concurrent.MoreExecutors; import org.junit.jupiter.api.AfterEach; @@ -53,6 +54,7 @@ import org.mockito.Mockito; import java.util.Collections; +import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.Optional; @@ -116,6 +118,7 @@ public void tearDown() { mockedEnv.close(); mockedEnv = null; } + MutableCatalogProvider.reset(); } @Test @@ -167,6 +170,79 @@ public void testReplayNameBasedRefreshInvalidatesColdTableByLogName() { CATALOG_ID, DATABASE_ID, DATABASE_NAME, TABLE_ID, TABLE_NAME); } + @Test + public void testExplicitRefreshReloadsTableNamesAndObject() throws Exception { + useMutableCatalog(); + TestExternalTable oldTable = database.getTableNullable(TABLE_NAME); + Assertions.assertNotNull(oldTable); + + MutableCatalogProvider.putTable("new_column"); + new RefreshManager().handleRefreshTable(CATALOG_NAME, DATABASE_NAME, TABLE_NAME, false); + + Assertions.assertNull(database.getCachedTableNamesForTest()); + Assertions.assertNull(database.getCachedTableForTest(TABLE_NAME)); + TestExternalTable refreshed = database.getTableNullable(TABLE_NAME); + Assertions.assertNotSame(oldTable, refreshed); + ArgumentCaptor log = ArgumentCaptor.forClass(ExternalObjectLog.class); + Mockito.verify(editLog).logRefreshExternalTable(log.capture()); + Assertions.assertTrue(log.getValue().isInvalidateTableLookup()); + } + + @Test + public void testContentRefreshRetainsTableNamesAndObject() { + useMutableCatalog(); + TestExternalTable table = database.getTableNullable(TABLE_NAME); + Assertions.assertNotNull(table); + + new RefreshManager().refreshTableAfterExternalMutation(table); + + Assertions.assertNotNull(database.getCachedTableNamesForTest()); + Assertions.assertSame(table, database.getCachedTableForTest(TABLE_NAME)); + ArgumentCaptor log = ArgumentCaptor.forClass(ExternalObjectLog.class); + Mockito.verify(editLog).logRefreshExternalTable(log.capture()); + Assertions.assertFalse(log.getValue().isInvalidateTableLookup()); + } + + @Test + public void testExplicitRefreshReplayClearsNamesWhenTableObjectIsCold() { + useMutableCatalog(); + TestExternalTable table = database.getTableNullable(TABLE_NAME); + Assertions.assertNotNull(table); + database.evictTableObjectForTest(TABLE_NAME); + MutableCatalogProvider.removeTable(); + + new RefreshManager().replayRefreshTable(ExternalObjectLog.createForExplicitRefreshTable( + CATALOG_ID, DATABASE_NAME, TABLE_NAME, 123L)); + + Assertions.assertNull(database.getCachedTableNamesForTest()); + Assertions.assertTrue(database.getTableNamesWithLock().isEmpty()); + } + + @Test + public void testExplicitRefreshReplayInvalidatesDatabaseInCaseInsensitiveMode() { + useMutableCatalog(2); + Assertions.assertNotNull(database.getTableNullable(TABLE_NAME)); + ExternalObjectLog log = ExternalObjectLog.createForExplicitRefreshTable( + CATALOG_ID, DATABASE_NAME, TABLE_NAME, 123L); + + new RefreshManager().replayRefreshTable( + GsonUtils.GSON.fromJson(GsonUtils.GSON.toJson(log), ExternalObjectLog.class)); + + Assertions.assertNull(database.getCachedTableNamesForTest()); + Mockito.verify(metaCacheMgr).invalidateDb(CATALOG_ID, DATABASE_ID, DATABASE_NAME); + } + + @Test + public void testExplicitRefreshReplayInvalidatesColdDatabaseInCaseInsensitiveMode() { + useMutableCatalog(2); + catalog.setInitializedForTest(false); + + new RefreshManager().replayRefreshTable(ExternalObjectLog.createForExplicitRefreshTable( + CATALOG_ID, DATABASE_NAME, TABLE_NAME, 123L)); + + Mockito.verify(metaCacheMgr).invalidateDb(CATALOG_ID, DATABASE_ID, DATABASE_NAME); + } + @Test public void testRefreshTableRemovesRowCountLoadedDuringConnectorInvalidation() throws Exception { PluginCatalogFixture fixture = usePluginCatalog(); @@ -405,6 +481,45 @@ private void assertColdRenameMigrated() { Assertions.assertEquals(NEW_TABLE_NAME, constraintManager.newTableName.getTbl()); } + private void useMutableCatalog() { + useMutableCatalog(0); + } + + private void useMutableCatalog(int lowerCaseTableNames) { + MutableCatalogProvider.putTable("old_column"); + Map properties = new HashMap<>(); + properties.put("catalog_provider.class", MutableCatalogProvider.class.getName()); + properties.put(ExternalCatalog.LOWER_CASE_TABLE_NAMES, Integer.toString(lowerCaseTableNames)); + catalog = new TestExternalCatalog(CATALOG_ID, CATALOG_NAME, "", properties, ""); + catalog.setInitializedForTest(true); + database = new CountingDatabase(catalog, DATABASE_ID, DATABASE_NAME, DATABASE_NAME); + database.setInitializedForTest(true); + catalog.addDatabaseForTest(database); + testingCatalogMgr.setCatalog(catalog); + } + + public static class MutableCatalogProvider implements TestExternalCatalog.TestCatalogProvider { + private static final Map>> METADATA = new HashMap<>(); + + static void putTable(String columnName) { + METADATA.computeIfAbsent(DATABASE_NAME, ignored -> new HashMap<>()) + .put(TABLE_NAME, Collections.singletonList(new Column(columnName, PrimitiveType.INT))); + } + + static void removeTable() { + METADATA.get(DATABASE_NAME).remove(TABLE_NAME); + } + + static void reset() { + METADATA.clear(); + } + + @Override + public Map>> getMetadata() { + return METADATA; + } + } + public static class EmptyCatalogProvider implements TestExternalCatalog.TestCatalogProvider { @Override public Map>> getMetadata() { @@ -471,6 +586,11 @@ void setCatalog(CatalogIf> catalog) { public CatalogIf> getCatalog(long id) { return catalog != null && id == catalog.getId() ? catalog : null; } + + @Override + public CatalogIf> getCatalog(String name) { + return catalog != null && name.equals(catalog.getName()) ? catalog : null; + } } private static class TestingEnv extends Env { diff --git a/fe/fe-core/src/test/java/org/apache/doris/datasource/ExternalDatabaseTest.java b/fe/fe-core/src/test/java/org/apache/doris/datasource/ExternalDatabaseTest.java index d2d9a91e77209c..7d16e07e96142a 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/datasource/ExternalDatabaseTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/datasource/ExternalDatabaseTest.java @@ -63,6 +63,53 @@ protected void runBeforeAll() throws Exception { createDefaultCtx().setThreadLocalInfo(); } + @Test + public void testExplicitRefreshInvalidatesNamesObjectAndId() { + NameMissTableCatalogProvider.reset(); + try { + NameMissTableCatalogProvider.putTable("db1", "Foo"); + NameMissTableCatalog catalog = new NameMissTableCatalog(0); + InspectableDatabase db = new InspectableDatabase(catalog, 200L, "db1", "db1"); + db.setInitializedForTest(true); + TestExternalTable oldTable = db.getTableNullable("Foo"); + Assertions.assertNotNull(oldTable); + + NameMissTableCatalogProvider.MOCKED_META.get("db1").remove("Foo"); + db.invalidateTableLookup("Foo"); + + Assertions.assertNull(db.getCachedTableForTest("Foo")); + Assertions.assertNull(db.getCachedTableNameByIdForTest(oldTable.getId())); + Assertions.assertTrue(db.getTableNamesWithLock().isEmpty()); + } finally { + NameMissTableCatalogProvider.reset(); + } + } + + @Test + public void testExplicitRefreshFencesCaseEquivalentOldObject() { + NameMissTableCatalogProvider.reset(); + try { + NameMissTableCatalogProvider.putTable("db_ci", "Dest"); + CaseInsensitiveCatalog catalog = new CaseInsensitiveCatalog(); + InspectableDatabase db = new InspectableDatabase(catalog, 210L, "db_ci", "db_ci"); + db.setInitializedForTest(true); + TestExternalTable oldTable = db.getTableNullable("Dest"); + Assertions.assertNotNull(oldTable); + + NameMissTableCatalogProvider.MOCKED_META.get("db_ci").remove("Dest"); + NameMissTableCatalogProvider.putTable("db_ci", "DEST"); + db.resetMetaCacheNames(); + Assertions.assertTrue(db.getTableNamesWithLock().contains("DEST")); + db.invalidateTableLookup("DEST"); + + Assertions.assertNull(db.getCachedTableForTest("Dest")); + Assertions.assertNull(db.getCachedTableNameByIdForTest(oldTable.getId())); + Assertions.assertNotSame(oldTable, db.getTableNullable("DEST")); + } finally { + NameMissTableCatalogProvider.reset(); + } + } + @Test public void testResetMetaCacheNamesKeepsObjectAndIdCache() { InspectableCatalog catalog = new InspectableCatalog(); From 0da4c6d3143e1adf6a4d28bd76342fb8c66d5fde Mon Sep 17 00:00:00 2001 From: 924060929 Date: Fri, 23 Oct 2026 12:13:22 +0800 Subject: [PATCH 2/8] [fix](fe) Invalidate connector database for case-insensitive table refresh ### What problem does this PR solve? Issue Number: None Problem Summary: Explicit REFRESH TABLE with case-insensitive table lookup invalidates all FE table identities in the database, but a hot-table replay could invalidate only the selected connector table spelling. If another spelling already had a cached connector snapshot, queries could continue reading that snapshot after refresh. Invalidate the connector database before clearing FE lookup and metadata caches in this mode, while preserving single-table invalidation for other refreshes. ### Release note None ### Check List (For Author) - Test: Unit Test - RefreshManagerTest: 16 tests passed. - FE build and Checkstyle passed. - Behavior changed: Yes. Explicit case-insensitive REFRESH TABLE invalidates database-scoped connector caches. - Does this need documentation: No --- .../apache/doris/catalog/RefreshManager.java | 13 +++++++-- .../doris/catalog/RefreshManagerTest.java | 29 ++++++++++++++++++- 2 files changed, 38 insertions(+), 4 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 7964310f1c16c5..6a92bf51ab3d8a 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 @@ -321,6 +321,7 @@ public void refreshTableInternal(ExternalDatabase db, ExternalTable table, long private void refreshTableInternal(ExternalDatabase db, ExternalTable table, long updateTime, boolean invalidateTableLookup) { + boolean invalidateDatabase = invalidateTableLookup && table.getCatalog().getLowerCaseTableNames() == 2; table.unsetObjectCreated(); if (updateTime > 0) { table.setUpdateTime(updateTime); @@ -328,8 +329,14 @@ private void refreshTableInternal(ExternalDatabase db, ExternalTable table, long // Connector metadata is the row-count source, so invalidate it before engine and row-count caches. try { if (table.getCatalog() instanceof PluginDrivenExternalCatalog) { - ((PluginDrivenExternalCatalog) table.getCatalog()).getConnector() - .invalidateTable(db.getRemoteName(), table.getRemoteName()); + if (invalidateDatabase) { + // A refreshed local name may resolve to a different remote spelling in this database. + ((PluginDrivenExternalCatalog) table.getCatalog()).getConnector() + .invalidateDb(db.getRemoteName()); + } else { + ((PluginDrivenExternalCatalog) table.getCatalog()).getConnector() + .invalidateTable(db.getRemoteName(), table.getRemoteName()); + } } } finally { try { @@ -337,7 +344,7 @@ private void refreshTableInternal(ExternalDatabase db, ExternalTable table, long db.invalidateTableLookup(table.getName()); } } finally { - if (invalidateTableLookup && table.getCatalog().getLowerCaseTableNames() == 2) { + if (invalidateDatabase) { Env.getCurrentEnv().getExtMetaCacheMgr().invalidateDb( table.getCatalog().getId(), db.getId(), db.getFullName()); } else { 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 index e0e79fdbfc268f..cbed01bb2aa075 100644 --- 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 @@ -243,6 +243,27 @@ public void testExplicitRefreshReplayInvalidatesColdDatabaseInCaseInsensitiveMod Mockito.verify(metaCacheMgr).invalidateDb(CATALOG_ID, DATABASE_ID, DATABASE_NAME); } + @Test + public void testExplicitRefreshReplayInvalidatesConnectorDatabaseInCaseInsensitiveMode() { + PluginCatalogFixture fixture = usePluginCatalog(2); + ExternalTable oldTable = new ExternalTable(TABLE_ID, TABLE_NAME, TABLE_NAME, + fixture.catalog, fixture.database, TableIf.TableType.PLUGIN_EXTERNAL_TABLE); + String otherSpelling = TABLE_NAME.toUpperCase(java.util.Locale.ROOT); + ExternalTable otherTable = new ExternalTable( + Util.genIdByName(CATALOG_NAME, DATABASE_NAME, otherSpelling), otherSpelling, otherSpelling, + fixture.catalog, fixture.database, TableIf.TableType.PLUGIN_EXTERNAL_TABLE); + fixture.database.addTableForTest(oldTable); + fixture.database.addTableForTest(otherTable); + + new RefreshManager().replayRefreshTable(ExternalObjectLog.createForExplicitRefreshTable( + CATALOG_ID, DATABASE_NAME, TABLE_NAME, 123L)); + + Mockito.verify(fixture.connector).invalidateDb(DATABASE_NAME); + Mockito.verify(fixture.connector, Mockito.never()).invalidateTable(DATABASE_NAME, TABLE_NAME); + Mockito.verify(metaCacheMgr).invalidateDb(CATALOG_ID, DATABASE_ID, DATABASE_NAME); + Assertions.assertFalse(fixture.database.getTableForReplay(otherSpelling).isPresent()); + } + @Test public void testRefreshTableRemovesRowCountLoadedDuringConnectorInvalidation() throws Exception { PluginCatalogFixture fixture = usePluginCatalog(); @@ -442,8 +463,14 @@ private Connector usePluginCatalogWithDisabledDatabaseObjectCache() { } private PluginCatalogFixture usePluginCatalog() { + return usePluginCatalog(0); + } + + private PluginCatalogFixture usePluginCatalog(int lowerCaseTableNames) { Connector connector = Mockito.mock(Connector.class); - Map properties = Collections.singletonMap("type", "test"); + Map properties = new HashMap<>(); + properties.put("type", "test"); + properties.put(ExternalCatalog.LOWER_CASE_TABLE_NAMES, Integer.toString(lowerCaseTableNames)); PluginDrivenExternalCatalog pluginCatalog = new PluginDrivenExternalCatalog( CATALOG_ID, CATALOG_NAME, null, properties, "", connector); pluginCatalog.setInitializedForTest(true); From 8bd1d68d65ac063430ba54c21e14c32e4288795c Mon Sep 17 00:00:00 2001 From: 924060929 Date: Fri, 23 Oct 2026 12:52:30 +0800 Subject: [PATCH 3/8] [fix](fe) Keep bulk table ID cleanup atomic with cache publication ### What problem does this PR solve? Issue Number: None Related PR: None Problem Summary: A catalog-wide table cache invalidation could clear the table ID index after a concurrent lookup published a fresh table. The table object would remain cached while its ID lookup disappeared. Run the index clear at the same publication boundary as cache invalidation, and route warm-value index repairs through that boundary. A concurrent lookup test covers the clear and republish ordering. ### Release note None ### Check List (For Author) - Test: Unit Test: FeMetaCacheEntryTest, ExternalDatabaseTest, RefreshManagerTest (72 passed); FE build and Checkstyle passed. - Behavior changed: Yes (table ID lookup remains consistent across concurrent bulk invalidation) - Does this need documentation: No --- .../connector/cache/CatalogMetaCache.java | 5 +++ .../cache/ScopedMetaCacheRegistry.java | 26 ++++++++++++ .../doris/datasource/ExternalDatabase.java | 5 ++- .../metacache/FeMetaCacheEntry.java | 11 ++++- .../metacache/FeMetaCacheEntryTest.java | 40 +++++++++++++++++++ 5 files changed, 83 insertions(+), 4 deletions(-) diff --git a/fe/fe-connector/fe-connector-cache/src/main/java/org/apache/doris/connector/cache/CatalogMetaCache.java b/fe/fe-connector/fe-connector-cache/src/main/java/org/apache/doris/connector/cache/CatalogMetaCache.java index d84c389b04b186..45b53b48fef542 100644 --- a/fe/fe-connector/fe-connector-cache/src/main/java/org/apache/doris/connector/cache/CatalogMetaCache.java +++ b/fe/fe-connector/fe-connector-cache/src/main/java/org/apache/doris/connector/cache/CatalogMetaCache.java @@ -151,6 +151,11 @@ public void invalidateCatalog() { registry.invalidate(ScopePath.catalog()); } + /** Runs a short auxiliary-index update at the same publication boundary as catalog invalidation. */ + public void invalidateCatalogAndRun(Runnable commitAction) { + registry.invalidateAndRunUnderPublicationGate(ScopePath.catalog(), commitAction); + } + public void invalidateDatabase(String database) { registry.invalidate(ScopePath.database(database)); } diff --git a/fe/fe-connector/fe-connector-cache/src/main/java/org/apache/doris/connector/cache/ScopedMetaCacheRegistry.java b/fe/fe-connector/fe-connector-cache/src/main/java/org/apache/doris/connector/cache/ScopedMetaCacheRegistry.java index 918b2829a7927b..575ffee012d50b 100644 --- a/fe/fe-connector/fe-connector-cache/src/main/java/org/apache/doris/connector/cache/ScopedMetaCacheRegistry.java +++ b/fe/fe-connector/fe-connector-cache/src/main/java/org/apache/doris/connector/cache/ScopedMetaCacheRegistry.java @@ -212,6 +212,32 @@ void invalidate(ScopePath path, Runnable afterStateReplacement) { tryPrune(invalidated.existing); } + /** Keeps a short auxiliary-index mutation ordered with value publication, then cleans values outside the gate. */ + void invalidateAndRunUnderPublicationGate(ScopePath path, Runnable commitAction) { + Objects.requireNonNull(path, "path can not be null"); + Runnable action = Objects.requireNonNull(commitAction, "commitAction can not be null"); + checkOpen(); + AtomicReference invalidated = new AtomicReference<>(); + try { + publicationGate.write(() -> { + List existing = resolveExisting(path); + bumpPublicationStates(existing); + if (existing.size() == path.level().ordinal() + 1) { + ScopeNode target = existing.get(existing.size() - 1); + invalidated.set(new InvalidatedScope(existing, replaceState(target))); + } + action.run(); + return null; + }); + } finally { + InvalidatedScope scope = invalidated.get(); + if (scope != null) { + cleanDetachedState(scope.oldState); + tryPrune(scope.existing); + } + } + } + void invalidate(List paths, Runnable afterStateReplacement) { Objects.requireNonNull(paths, "paths can not be null"); Objects.requireNonNull(afterStateReplacement, "afterStateReplacement can not be null"); 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 62ee38684e88f0..afdcb9e0d2df4c 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 @@ -767,9 +767,10 @@ private void invalidateAllTableCache() { tableNames.invalidateAll(); } if (tables != null) { - tables.invalidateAll(); + tables.invalidateAllAndRun(tableIdNameIndex::clear); + } else { + tableIdNameIndex.clear(); } - tableIdNameIndex.clear(); } @Override diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/metacache/FeMetaCacheEntry.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/metacache/FeMetaCacheEntry.java index 1cb7f0dcf9c444..4c140171110be4 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/datasource/metacache/FeMetaCacheEntry.java +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/metacache/FeMetaCacheEntry.java @@ -239,7 +239,8 @@ public V getAndRunIfCurrent(K key, BiPredicate actionRequired, && (!effectiveEnabled || data.getIfPresent(key) == value); if (current && required.test(key, value) && isCurrent(stripe, token)) { try { - action.accept(key, value); + // A warm-value index repair must share the publication barrier with bulk invalidation. + data.compareAndSet(key, value, value, () -> action.accept(key, value)); } catch (RuntimeException | Error throwable) { data.invalidateKey(key); throw throwable; @@ -360,6 +361,12 @@ public void invalidateKeyAndRun(K key, Runnable afterInvalidation) { } public void invalidateAll() { + invalidateAllAndRun(() -> { + }); + } + + public void invalidateAllAndRun(Runnable commitAction) { + Runnable action = Objects.requireNonNull(commitAction, "commitAction can not be null"); for (int i = 0; i < stripeCount; i++) { StripeState stripe = stripeStates.get(i); if (stripe != null) { @@ -370,7 +377,7 @@ public void invalidateAll() { } } } - owner.invalidateCatalog(); + owner.invalidateCatalogAndRun(action); } public void forEach(BiConsumer consumer) { diff --git a/fe/fe-core/src/test/java/org/apache/doris/datasource/metacache/FeMetaCacheEntryTest.java b/fe/fe-core/src/test/java/org/apache/doris/datasource/metacache/FeMetaCacheEntryTest.java index 22bbc80ad788f0..bcadcd3f7149f8 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/datasource/metacache/FeMetaCacheEntryTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/datasource/metacache/FeMetaCacheEntryTest.java @@ -25,7 +25,9 @@ import java.util.ArrayList; import java.util.HashSet; import java.util.List; +import java.util.Map; import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutionException; import java.util.concurrent.ExecutorService; @@ -271,6 +273,44 @@ protected void beforeCurrentValueActionForTest(String key, Integer value) { } } + @Test + public void testBulkInvalidationOrdersNewObjectAndIdPublication() throws Exception { + ExecutorService refreshExecutor = Executors.newSingleThreadExecutor(); + ExecutorService workers = Executors.newFixedThreadPool(2); + CountDownLatch indexClearStarted = new CountDownLatch(1); + CountDownLatch finishIndexClear = new CountDownLatch(1); + CountDownLatch newLookupStarted = new CountDownLatch(1); + Map ids = new ConcurrentHashMap<>(); + try { + FeMetaCacheEntry entry = new FeMetaCacheEntry<>( + "objects", String::length, ENABLED, refreshExecutor, false); + entry.getAndRunIfCurrent("old", (key, value) -> ids.put(key, value)); + + Future invalidate = workers.submit(() -> entry.invalidateAllAndRun(() -> { + indexClearStarted.countDown(); + await(finishIndexClear); + ids.clear(); + })); + await(indexClearStarted); + Future newLookup = workers.submit(() -> { + newLookupStarted.countDown(); + return entry.getAndRunIfCurrent("new", (key, value) -> ids.put(key, value)); + }); + await(newLookupStarted); + finishIndexClear.countDown(); + + invalidate.get(3L, TimeUnit.SECONDS); + Assertions.assertEquals(Integer.valueOf(3), newLookup.get(3L, TimeUnit.SECONDS)); + Assertions.assertEquals(Integer.valueOf(3), entry.getIfPresent("new")); + Assertions.assertEquals(Integer.valueOf(3), ids.get("new")); + Assertions.assertFalse(ids.containsKey("old")); + } finally { + finishIndexClear.countDown(); + workers.shutdownNow(); + refreshExecutor.shutdownNow(); + } + } + @Test public void testColdValueRemainsHiddenUntilAuxiliaryIndexPublication() throws Exception { ExecutorService refreshExecutor = Executors.newSingleThreadExecutor(); From de5afecd42b0d50022f19ad40e845dc3596ed24e Mon Sep 17 00:00:00 2001 From: 924060929 Date: Fri, 23 Oct 2026 14:32:02 +0800 Subject: [PATCH 4/8] [test](fe) Update refresh test for replaced external table identity ### What problem does this PR solve? Issue Number: None Related PR: None Problem Summary: Explicit external-table refresh now invalidates the cached table object. The existing test continued to read the old object after refresh and expected its schema load to advance the old update time, causing FE UT to fail. Resolve the table again and verify the replacement is reachable by both name and ID while the old object's update time stays unchanged. ### Release note None ### Check List (For Author) - Test: Unit Test: RefreshTableTest, FeMetaCacheEntryTest, ExternalDatabaseTest, RefreshManagerTest (74 passed); FE build and Checkstyle passed. - Behavior changed: No (test expectation only) - Does this need documentation: No --- .../apache/doris/catalog/RefreshTableTest.java | 15 ++++++++------- 1 file changed, 8 insertions(+), 7 deletions(-) diff --git a/fe/fe-core/src/test/java/org/apache/doris/catalog/RefreshTableTest.java b/fe/fe-core/src/test/java/org/apache/doris/catalog/RefreshTableTest.java index 36edaacdf0197c..b9935acbd342c6 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/catalog/RefreshTableTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/catalog/RefreshTableTest.java @@ -104,13 +104,14 @@ public void testRefreshTable() throws Exception { Assertions.assertFalse(table.isObjectCreated()); long l3 = table.getUpdateTime(); Assertions.assertTrue(l3 >= l2); - table.getFullSchema(); - // only table.getFullSchema() can change table.lastUpdateTime - long l4 = table.getUpdateTime(); - Assertions.assertTrue(l4 > l3); - // updateTime is equal to schema update time as default - long l5 = table.getUpdateTime(); - Assertions.assertTrue(l5 >= l4); + TestExternalTable refreshed = (TestExternalTable) test1.getDbNullable("db1").getTable("tbl11").get(); + Assertions.assertNotSame(table, refreshed); + Assertions.assertSame(refreshed, test1.getDbNullable("db1").getTableNullable(refreshed.getId())); + refreshed.getFullSchema(); + // Schema loading updates the replacement table, not the invalidated table object. + long l4 = refreshed.getUpdateTime(); + Assertions.assertTrue(l4 >= l3); + Assertions.assertEquals(l3, table.getUpdateTime()); // external info schema db ExternalInfoSchemaDatabase infoDb = (ExternalInfoSchemaDatabase) test1.getDbNullable(InfoSchemaDb.DATABASE_NAME); From a099bd2d6c400f7ca3c7a011e5bd164a8296e483 Mon Sep 17 00:00:00 2001 From: 924060929 Date: Fri, 23 Oct 2026 16:39:24 +0800 Subject: [PATCH 5/8] [fix](fe) Clean missing external table lookup state safely ### What problem does this PR solve? Issue Number: None Problem Summary: A live connector lookup can confirm a remotely dropped table while FE names, objects, IDs and metadata still retain its old identity. Capture exact retained cache observations before the remote read, and conditionally invalidate matching state only while the catalog, database, names and object registrations are still current. This prevents an old missing response from clearing a refreshed or recreated table. Defer physical removal and resource release until all enclosing guards are released. Keep ordinary write refreshes narrow. Also preserve numeric ID publication across refresh, cover both case-folding modes on explicit refresh/replay, and initialize the timestamp on successful object reload. ### Release note External table lookups automatically clear matching stale metadata after a live lookup confirms a remote drop. ### Check List (For Author) - Test: Unit Test: 313 cache/FE cases and 33 ES/Paimon cases; Hive metadata regression with generated output and comparison; ./build.sh --fe -j4 including Checkstyle - Behavior changed: Yes, definitive missing-table observations invalidate matching external lookup state; ordinary writes do not reload table names - Does this need documentation: No --- .../doris/connector/cache/MetaCache.java | 31 + .../connector/cache/MetaCacheCleanup.java | 79 +++ .../connector/cache/ScopedMetaCache.java | 42 +- .../cache/ScopedMetaCacheRegistry.java | 26 +- .../connector/cache/CatalogMetaCacheTest.java | 177 ++++++ .../connector/es/EsConnectorRestClient.java | 12 +- .../es/EsConnectorRestClientTest.java | 29 + .../paimon/PaimonConnectorMetadata.java | 3 +- .../paimon/PaimonConnectorMetadataTest.java | 9 + .../paimon/RecordingPaimonCatalogOps.java | 4 + .../spi/ConnectorTableMetadataOps.java | 5 +- .../apache/doris/catalog/RefreshManager.java | 6 +- .../doris/datasource/ExternalCatalog.java | 8 + .../doris/datasource/ExternalDatabase.java | 55 +- .../metacache/FeMetaCacheEntry.java | 45 ++ .../plugin/PluginDrivenExternalTable.java | 16 +- .../doris/catalog/RefreshManagerTest.java | 23 +- .../doris/catalog/RefreshTableTest.java | 2 + .../plugin/PluginDrivenMissingTableTest.java | 550 ++++++++++++++++++ .../test_hive_external_lookup_refresh.out | 35 ++ .../test_hive_external_lookup_refresh.groovy | 93 +++ 21 files changed, 1223 insertions(+), 27 deletions(-) create mode 100644 fe/fe-connector/fe-connector-cache/src/main/java/org/apache/doris/connector/cache/MetaCacheCleanup.java create mode 100644 fe/fe-core/src/test/java/org/apache/doris/datasource/plugin/PluginDrivenMissingTableTest.java create mode 100644 regression-test/data/external_table_p0/hive/test_hive_external_lookup_refresh.out create mode 100644 regression-test/suites/external_table_p0/hive/test_hive_external_lookup_refresh.groovy diff --git a/fe/fe-connector/fe-connector-cache/src/main/java/org/apache/doris/connector/cache/MetaCache.java b/fe/fe-connector/fe-connector-cache/src/main/java/org/apache/doris/connector/cache/MetaCache.java index 07afd00576d4bf..e30a24f72f9487 100644 --- a/fe/fe-connector/fe-connector-cache/src/main/java/org/apache/doris/connector/cache/MetaCache.java +++ b/fe/fe-connector/fe-connector-cache/src/main/java/org/apache/doris/connector/cache/MetaCache.java @@ -18,7 +18,10 @@ package org.apache.doris.connector.cache; import java.util.Objects; +import java.util.Optional; import java.util.function.BiConsumer; +import java.util.function.BiFunction; +import java.util.function.BooleanSupplier; import java.util.function.Consumer; import java.util.function.Function; @@ -89,6 +92,34 @@ public boolean compareAndSet(K key, V expectedValue, V updatedValue, Runnable co Objects.requireNonNull(commitAction, "commitAction can not be null")); } + /** Captures an exact retained registration without loading or holding locks across a remote read. */ + public Optional observe(K key, Object expectedValue) { + K nonNullKey = Objects.requireNonNull(key, "key"); + return delegate.observe(nonNullKey, definition.scope(nonNullKey), + Objects.requireNonNull(expectedValue, "expectedValue")); + } + + /** + * A short guarded action may nest observations in parent-to-child order. It must not perform I/O, + * re-enter this cache, or invalidate this cache's owner. Returning false leaves this entry unchanged. + * Conditional invalidation still removes the observed value if its dependency cleanup throws. + */ + public static final class Observation { + private final BiFunction operation; + + Observation(BiFunction operation) { + this.operation = operation; + } + + public boolean runIfCurrent(BooleanSupplier action) { + return operation.apply(false, Objects.requireNonNull(action, "action")); + } + + public boolean invalidateIfCurrent(BooleanSupplier action) { + return operation.apply(true, Objects.requireNonNull(action, "action")); + } + } + public void invalidateKey(K key) { delegate.invalidateKey(Objects.requireNonNull(key, "key can not be null")); } diff --git a/fe/fe-connector/fe-connector-cache/src/main/java/org/apache/doris/connector/cache/MetaCacheCleanup.java b/fe/fe-connector/fe-connector-cache/src/main/java/org/apache/doris/connector/cache/MetaCacheCleanup.java new file mode 100644 index 00000000000000..fb6f235aec3d97 --- /dev/null +++ b/fe/fe-connector/fe-connector-cache/src/main/java/org/apache/doris/connector/cache/MetaCacheCleanup.java @@ -0,0 +1,79 @@ +// 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.connector.cache; + +import java.util.ArrayDeque; +import java.util.Deque; + +/** Defers physical removal and resource release until an enclosing guarded cache mutation has finished. */ +public final class MetaCacheCleanup { + private static final ThreadLocal> PENDING = new ThreadLocal<>(); + + private MetaCacheCleanup() { + } + + /** + * Logical invalidation still completes inside the action. Detached registrations and removal callbacks + * are cleaned synchronously after the outermost action returns, including when that action fails. + */ + public static void runDeferred(Runnable action) { + if (PENDING.get() != null) { + action.run(); + return; + } + Deque pending = new ArrayDeque<>(); + PENDING.set(pending); + Throwable failure = null; + try { + action.run(); + } catch (RuntimeException | Error e) { + failure = e; + } + try { + Runnable cleanup; + while ((cleanup = pending.pollFirst()) != null) { + try { + cleanup.run(); + } catch (RuntimeException | Error e) { + if (failure == null) { + failure = e; + } else if (failure != e) { + failure.addSuppressed(e); + } + } + } + } finally { + PENDING.remove(); + } + if (failure instanceof RuntimeException) { + throw (RuntimeException) failure; + } + if (failure != null) { + throw (Error) failure; + } + } + + static void execute(Runnable cleanup) { + Deque pending = PENDING.get(); + if (pending == null) { + cleanup.run(); + } else { + pending.addLast(cleanup); + } + } +} diff --git a/fe/fe-connector/fe-connector-cache/src/main/java/org/apache/doris/connector/cache/ScopedMetaCache.java b/fe/fe-connector/fe-connector-cache/src/main/java/org/apache/doris/connector/cache/ScopedMetaCache.java index b58ad13bd9dfa5..960822e9c558f4 100644 --- a/fe/fe-connector/fe-connector-cache/src/main/java/org/apache/doris/connector/cache/ScopedMetaCache.java +++ b/fe/fe-connector/fe-connector-cache/src/main/java/org/apache/doris/connector/cache/ScopedMetaCache.java @@ -270,6 +270,44 @@ public V getWithPublicationAction(K key, ScopePath path, Function loader, } } + Optional observe(K key, ScopePath path, Object expectedValue) { + VersionedValue value = currentVersionedValue(key, path); + if (value == null || value.value != expectedValue) { + return Optional.empty(); + } + ScopedMetaCacheRegistry.ScopeObservation scope = registry.observe(value.scopeSnapshot); + return Optional.of(new MetaCache.Observation((invalidate, action) -> deferRemovals(() -> { + BooleanSupplier commit = () -> scope.runIfCurrent(() -> { + synchronized (value.keyNode) { + if (closed.get() || data.getIfPresent(key) != value || !value.isCurrent(registry, keyNodes)) { + return false; + } + boolean invalidateValue = invalidate; + try { + if (!action.getAsBoolean()) { + invalidateValue = false; + return false; + } + return true; + } finally { + // Once authorized, a throwing dependency cleanup must not retain this stale value. + if (invalidateValue) { + exactInvalidationSequence = exactInvalidationSequence.add(BigInteger.ONE); + if (!activeBulkStarts.isEmpty()) { + exactInvalidations.put(key, exactInvalidationSequence); + } + replaceKeyState(value.keyNode); + value.keyNode.loadPublicationState.set(new Object()); + removeCurrentVersion(key, value); + } + } + } + }); + return invalidate ? bulkInvalidationGate.write(commit::getAsBoolean) + : bulkInvalidationGate.readBoolean(commit); + }))); + } + public V getIfPresent(K key, ScopePath path) { Objects.requireNonNull(key, "key can not be null"); Objects.requireNonNull(path, "path can not be null"); @@ -1118,7 +1156,9 @@ private void drainDeferredRemovals(RemovalDeferral removalDeferral) { DeferredRemoval removal; while ((removal = removalDeferral.removals.pollFirst()) != null) { try { - completeRemoval(removal.versioned, removal.cause, removal.budgetEviction); + DeferredRemoval detached = removal; + MetaCacheCleanup.execute(() -> + completeRemoval(detached.versioned, detached.cause, detached.budgetEviction)); } catch (RuntimeException | Error e) { if (failure == null) { failure = e; diff --git a/fe/fe-connector/fe-connector-cache/src/main/java/org/apache/doris/connector/cache/ScopedMetaCacheRegistry.java b/fe/fe-connector/fe-connector-cache/src/main/java/org/apache/doris/connector/cache/ScopedMetaCacheRegistry.java index 575ffee012d50b..7f749acbab512b 100644 --- a/fe/fe-connector/fe-connector-cache/src/main/java/org/apache/doris/connector/cache/ScopedMetaCacheRegistry.java +++ b/fe/fe-connector/fe-connector-cache/src/main/java/org/apache/doris/connector/cache/ScopedMetaCacheRegistry.java @@ -268,6 +268,26 @@ void invalidate(List paths, Runnable afterStateReplacement) { invalidated.forEach(scope -> tryPrune(scope.existing)); } + ScopeObservation observe(ScopeSnapshot snapshot) { + return new ScopeObservation(snapshot); + } + + final class ScopeObservation { + private final ScopeSnapshot snapshot; + private final PublicationState publicationState; + + private ScopeObservation(ScopeSnapshot snapshot) { + this.snapshot = snapshot; + this.publicationState = snapshot.leafNode().descendantPublication.get(); + } + + boolean runIfCurrent(BooleanSupplier action) { + return publicationGate.readBoolean(() -> snapshot.isCurrent(ScopedMetaCacheRegistry.this) + && snapshot.leafNode().descendantPublication.get() == publicationState + && action.getAsBoolean()); + } + } + public ScopeMetrics metrics() { ScopeMetricsAccumulator accumulator = new ScopeMetricsAccumulator(); collectMetrics(root, accumulator); @@ -459,12 +479,16 @@ private void bumpPublicationStates(List nodes) { } private void cleanDetachedState(ScopeState state) { + MetaCacheCleanup.execute(() -> cleanDetachedStateNow(state)); + } + + private void cleanDetachedStateNow(ScopeState state) { state.entries.forEach((address, ownershipToken) -> { if (state.entries.remove(address, ownershipToken)) { address.removeExpected(ownershipToken); } }); - state.children.values().forEach(child -> cleanDetachedState(child.current.get())); + state.children.values().forEach(child -> cleanDetachedStateNow(child.current.get())); state.children.clear(); } diff --git a/fe/fe-connector/fe-connector-cache/src/test/java/org/apache/doris/connector/cache/CatalogMetaCacheTest.java b/fe/fe-connector/fe-connector-cache/src/test/java/org/apache/doris/connector/cache/CatalogMetaCacheTest.java index 3bb6ad66e80d68..9aebbc312da086 100644 --- a/fe/fe-connector/fe-connector-cache/src/test/java/org/apache/doris/connector/cache/CatalogMetaCacheTest.java +++ b/fe/fe-connector/fe-connector-cache/src/test/java/org/apache/doris/connector/cache/CatalogMetaCacheTest.java @@ -25,13 +25,190 @@ import java.util.ArrayDeque; import java.util.Arrays; import java.util.Queue; +import java.util.concurrent.CountDownLatch; import java.util.concurrent.Executor; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicReference; class CatalogMetaCacheTest { private static final CacheSpec SPEC = CacheSpec.of(true, CacheSpec.CACHE_NO_TTL, 100); + @Test + void deferredCleanupDetachesImmediatelyAndPreservesRecreatedValues() throws Exception { + Object guard = new Object(); + AtomicInteger removals = new AtomicInteger(); + AtomicInteger lockedRemovals = new AtomicInteger(); + ExecutorService worker = Executors.newSingleThreadExecutor(); + try (CatalogMetaCache owner = CatalogMetaCache.unmanaged()) { + MetaCache cache = owner.create(MetaCacheDefinition.builder( + "deferred", SPEC, ignored -> ScopePath.table("db", "t")) + .removalListener((key, value, cause) -> { + if (Thread.holdsLock(guard)) { + lockedRemovals.incrementAndGet(); + } + removals.incrementAndGet(); + }).build()); + Object old = new Object(); + Object recreated = new Object(); + cache.put("t", old); + MetaCacheCleanup.runDeferred(() -> { + synchronized (guard) { + MetaCacheCleanup.runDeferred(() -> owner.invalidateTable("db", "t")); + Assertions.assertNull(cache.getIfPresent("t")); + Assertions.assertEquals(0, removals.get()); + try { + // New publication can finish while physical cleanup of the old scope is pending. + worker.submit(() -> cache.put("t", recreated)).get(10, TimeUnit.SECONDS); + } catch (Exception e) { + throw new AssertionError(e); + } + Assertions.assertSame(recreated, cache.getIfPresent("t")); + } + }); + Assertions.assertEquals(1, removals.get()); + Assertions.assertSame(recreated, cache.getIfPresent("t")); + MetaCacheCleanup.runDeferred(() -> { + synchronized (guard) { + Assertions.assertTrue(cache.observe("t", recreated).get().invalidateIfCurrent(() -> true)); + Assertions.assertNull(cache.getIfPresent("t")); + Assertions.assertEquals(1, removals.get()); + } + }); + Assertions.assertEquals(2, removals.get()); + Assertions.assertEquals(0, lockedRemovals.get()); + } finally { + worker.shutdownNow(); + } + } + + @Test + void deferredCleanupDrainsAfterFailureAndKeepsOriginalException() { + AtomicInteger cleaned = new AtomicInteger(); + IllegalStateException original = new IllegalStateException("action"); + IllegalArgumentException cleanupFailure = new IllegalArgumentException("cleanup"); + Assertions.assertSame(original, Assertions.assertThrows(IllegalStateException.class, () -> + MetaCacheCleanup.runDeferred(() -> { + MetaCacheCleanup.execute(() -> { + throw cleanupFailure; + }); + MetaCacheCleanup.execute(cleaned::incrementAndGet); + throw original; + }))); + Assertions.assertEquals(1, cleaned.get()); + Assertions.assertArrayEquals(new Throwable[] {cleanupFailure}, original.getSuppressed()); + // The thread-local context is cleared even after failure. + MetaCacheCleanup.execute(cleaned::incrementAndGet); + Assertions.assertEquals(2, cleaned.get()); + } + + @Test + void observationRejectsReplacementScopeChangeAndSameInstanceReinsertion() { + CatalogMetaCache owner = CatalogMetaCache.unmanaged(); + try { + MetaCache cache = owner.create(MetaCacheDefinition.builder( + "observed", SPEC, ignored -> ScopePath.table("db", "t")).build()); + Object old = new Object(); + cache.put("t", old); + MetaCache.Observation first = cache.observe("t", old).get(); + cache.put("t", new Object()); + Assertions.assertFalse(first.invalidateIfCurrent(() -> true)); + cache.put("t", old); + Assertions.assertFalse(first.invalidateIfCurrent(() -> true)); + MetaCache.Observation beforeRefresh = cache.observe("t", old).get(); + owner.invalidateTable("db", "t"); + cache.put("t", old); + Assertions.assertFalse(beforeRefresh.invalidateIfCurrent(() -> true)); + MetaCache.Observation current = cache.observe("t", old).get(); + Assertions.assertFalse(current.invalidateIfCurrent(() -> false)); + Assertions.assertSame(old, cache.getIfPresent("t")); + Assertions.assertTrue(current.invalidateIfCurrent(() -> true)); + Assertions.assertNull(cache.getIfPresent("t")); + cache.put("t", old); + MetaCache.Observation beforeClose = cache.observe("t", old).get(); + owner.close(); + Assertions.assertFalse(beforeClose.invalidateIfCurrent(() -> true)); + } finally { + owner.close(); + } + } + + @Test + void nestedObservationLeavesNamesIntactWhenObjectWasReplaced() { + try (CatalogMetaCache namesOwner = CatalogMetaCache.unmanaged(); + CatalogMetaCache objectsOwner = CatalogMetaCache.unmanaged()) { + MetaCache names = namesOwner.create(MetaCacheDefinition.builder( + "names", SPEC, ignored -> ScopePath.catalog()).build()); + MetaCache objects = objectsOwner.create(MetaCacheDefinition.builder( + "objects", SPEC, ignored -> ScopePath.catalog()).build()); + Object snapshot = new Object(); + Object old = new Object(); + names.put("db", snapshot); + objects.put("t", old); + MetaCache.Observation nameObservation = names.observe("db", snapshot).get(); + MetaCache.Observation objectObservation = objects.observe("t", old).get(); + Object replacement = new Object(); + objects.put("t", replacement); + AtomicInteger actions = new AtomicInteger(); + Assertions.assertFalse(nameObservation.invalidateIfCurrent(() -> + objectObservation.invalidateIfCurrent(() -> { + actions.incrementAndGet(); + return true; + }))); + Assertions.assertSame(snapshot, names.getIfPresent("db")); + Assertions.assertSame(replacement, objects.getIfPresent("t")); + Assertions.assertEquals(0, actions.get()); + MetaCache.Observation currentObject = objects.observe("t", replacement).get(); + Assertions.assertThrows(IllegalStateException.class, () -> nameObservation.invalidateIfCurrent(() -> + currentObject.invalidateIfCurrent(() -> { + throw new IllegalStateException("dependency cleanup failed"); + }))); + Assertions.assertNull(names.getIfPresent("db")); + Assertions.assertNull(objects.getIfPresent("t")); + } + } + + @Test + void observedRemovalFencesPreexistingBulkAndSingleLoads() throws Exception { + ExecutorService worker = Executors.newSingleThreadExecutor(); + CountDownLatch started = new CountDownLatch(1); + CountDownLatch release = new CountDownLatch(1); + try (CatalogMetaCache owner = CatalogMetaCache.unmanaged()) { + MetaCache cache = owner.create(MetaCacheDefinition.builder( + "objects", SPEC, ignored -> ScopePath.table("db", "t")).build()); + Object old = new Object(); + cache.put("t", old); + try (MetaCache.BulkLoad bulk = cache.beginBulkLoad(ScopePath.database("db"))) { + MetaCache.Observation observation = cache.observe("t", old).get(); + Assertions.assertTrue(observation.invalidateIfCurrent(() -> true)); + Assertions.assertFalse(bulk.publish("t", old)); + Assertions.assertNull(cache.getIfPresent("t")); + } + Future load = worker.submit(() -> cache.get("t", key -> { + started.countDown(); + try { + Assertions.assertTrue(release.await(10, TimeUnit.SECONDS)); + } catch (InterruptedException e) { + throw new AssertionError(e); + } + return old; + })); + Assertions.assertTrue(started.await(10, TimeUnit.SECONDS)); + Object current = new Object(); + cache.put("t", current); + Assertions.assertTrue(cache.observe("t", current).get().invalidateIfCurrent(() -> true)); + release.countDown(); + Assertions.assertSame(old, load.get(10, TimeUnit.SECONDS)); + Assertions.assertNull(cache.getIfPresent("t")); + } finally { + release.countDown(); + worker.shutdownNow(); + } + } + @Test void unmanagedOwnershipMustBeSelectedExplicitly() throws Exception { Assertions.assertTrue(Modifier.isPrivate(CatalogMetaCache.class.getDeclaredConstructor().getModifiers())); diff --git a/fe/fe-connector/fe-connector-es/src/main/java/org/apache/doris/connector/es/EsConnectorRestClient.java b/fe/fe-connector/fe-connector-es/src/main/java/org/apache/doris/connector/es/EsConnectorRestClient.java index c509f45004fce2..d90c821f48b6c0 100644 --- a/fe/fe-connector/fe-connector-es/src/main/java/org/apache/doris/connector/es/EsConnectorRestClient.java +++ b/fe/fe-connector/fe-connector-es/src/main/java/org/apache/doris/connector/es/EsConnectorRestClient.java @@ -106,10 +106,16 @@ public boolean existIndex(String indexName) { OkHttpClient client = httpSslEnable ? getOrCreateSslClient() : PLAIN_CLIENT; String node = currentNode(); try (Response response = executeResponse(client, node, path, null)) { - return response.isSuccessful(); + if (response.isSuccessful()) { + return true; + } + if (response.code() == 404) { + return false; + } + throw new DorisConnectorException("Failed to check index " + indexName + + ": HTTP " + response.code()); } catch (IOException e) { - LOG.warn("existIndex error", e); - return false; + throw new DorisConnectorException("Failed to check index " + indexName, e); } } diff --git a/fe/fe-connector/fe-connector-es/src/test/java/org/apache/doris/connector/es/EsConnectorRestClientTest.java b/fe/fe-connector/fe-connector-es/src/test/java/org/apache/doris/connector/es/EsConnectorRestClientTest.java index 16439e765bd05b..d1354c93744c2f 100644 --- a/fe/fe-connector/fe-connector-es/src/test/java/org/apache/doris/connector/es/EsConnectorRestClientTest.java +++ b/fe/fe-connector/fe-connector-es/src/test/java/org/apache/doris/connector/es/EsConnectorRestClientTest.java @@ -17,10 +17,14 @@ package org.apache.doris.connector.es; +import org.apache.doris.connector.spi.DorisConnectorException; + +import com.sun.net.httpserver.HttpServer; import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.Test; import java.lang.reflect.Field; +import java.net.InetSocketAddress; import java.util.Collections; import java.util.Set; import java.util.concurrent.ConcurrentHashMap; @@ -34,6 +38,31 @@ */ public class EsConnectorRestClientTest { + @Test + public void indexAbsenceRequiresNotFoundResponse() throws Exception { + AtomicInteger status = new AtomicInteger(200); + HttpServer server = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0); + server.createContext("/t/_mapping", exchange -> { + exchange.sendResponseHeaders(status.get(), -1); + exchange.close(); + }); + server.start(); + EsConnectorRestClient client = new EsConnectorRestClient( + new String[]{"http://127.0.0.1:" + server.getAddress().getPort()}, null, null, false, null); + try { + Assertions.assertTrue(client.existIndex("t")); + status.set(404); + Assertions.assertFalse(client.existIndex("t")); + for (int failure : new int[]{401, 403, 500}) { + status.set(failure); + Assertions.assertThrows(DorisConnectorException.class, () -> client.existIndex("t")); + } + } finally { + server.stop(0); + } + Assertions.assertThrows(DorisConnectorException.class, () -> client.existIndex("t")); + } + @Test public void testConstructorWithAuth() throws Exception { EsConnectorRestClient client = new EsConnectorRestClient( diff --git a/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonConnectorMetadata.java b/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonConnectorMetadata.java index 245da688dd84f6..b7b365e81ebc4b 100644 --- a/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonConnectorMetadata.java +++ b/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonConnectorMetadata.java @@ -285,8 +285,7 @@ public Optional getTableHandle( return Optional.of(handle); }); } catch (Exception e) { - LOG.warn("Failed to get Paimon table handle: {}.{}", dbName, tableName, e); - return Optional.empty(); + throw new DorisConnectorException("Failed to get Paimon table handle: " + dbName + "." + tableName, e); } } diff --git a/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonConnectorMetadataTest.java b/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonConnectorMetadataTest.java index 4a7c5316f9f903..564b4809509e05 100644 --- a/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonConnectorMetadataTest.java +++ b/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonConnectorMetadataTest.java @@ -181,6 +181,15 @@ public void getTableHandleEmptyWhenTableMissing() { Assertions.assertFalse(handleOpt.isPresent()); } + @Test + public void getTableHandlePropagatesLookupFailure() { + RecordingPaimonCatalogOps ops = new RecordingPaimonCatalogOps(); + ops.tableLookupFailure = new IllegalStateException("metastore timeout"); + DorisConnectorException failure = Assertions.assertThrows(DorisConnectorException.class, + () -> metadataWith(ops).getTableHandle(null, "db1", "t1")); + Assertions.assertSame(ops.tableLookupFailure, failure.getCause()); + } + @Test public void getColumnHandlesReloadFallbackReloadsWhenTransientTableNull() { RecordingPaimonCatalogOps ops = new RecordingPaimonCatalogOps(); diff --git a/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/RecordingPaimonCatalogOps.java b/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/RecordingPaimonCatalogOps.java index 73147f40620a0d..a6465faf6c62ca 100644 --- a/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/RecordingPaimonCatalogOps.java +++ b/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/RecordingPaimonCatalogOps.java @@ -69,6 +69,7 @@ final class RecordingPaimonCatalogOps implements PaimonCatalogOps { boolean throwDatabaseNotExist; boolean throwTableNotExist; + RuntimeException tableLookupFailure; // ---- B3 DDL capture fields (inputs the metadata layer passed to the seam) ---- Schema lastCreatedSchema; @@ -163,6 +164,9 @@ public List listTables(String databaseName) throws Catalog.DatabaseNotEx @Override public Table getTable(Identifier identifier) throws Catalog.TableNotExistException { + if (tableLookupFailure != null) { + throw tableLookupFailure; + } log.add("getTable:" + identifier.getFullName()); lastGetTableId = identifier; if (throwTableNotExist) { diff --git a/fe/fe-connector/fe-connector-spi/src/main/java/org/apache/doris/connector/spi/ConnectorTableMetadataOps.java b/fe/fe-connector/fe-connector-spi/src/main/java/org/apache/doris/connector/spi/ConnectorTableMetadataOps.java index 4d29727e1b84d9..64263b639f59a4 100644 --- a/fe/fe-connector/fe-connector-spi/src/main/java/org/apache/doris/connector/spi/ConnectorTableMetadataOps.java +++ b/fe/fe-connector/fe-connector-spi/src/main/java/org/apache/doris/connector/spi/ConnectorTableMetadataOps.java @@ -62,7 +62,10 @@ */ public interface ConnectorTableMetadataOps { - /** Retrieves a table handle for the given database and table name. */ + /** + * Retrieves a table handle for the given database and table name. Empty means the table does not exist; + * authentication, transport and other lookup failures must propagate rather than report absence. + */ @ConnectorMustImplement default Optional getTableHandle( ConnectorSession session, String dbName, 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 6a92bf51ab3d8a..4e81dcabdaff4a 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 @@ -209,7 +209,7 @@ public void replayRefreshTable(ExternalObjectLog log) { try { invalidateAllConnectorCachesIfPresent(catalog); } finally { - if (log.isInvalidateTableLookup() && catalog.getLowerCaseTableNames() == 2) { + if (log.isInvalidateTableLookup() && catalog.getLowerCaseTableNames() != 0) { Env.getCurrentEnv().getExtMetaCacheMgr().invalidateDb(log.getCatalogId(), dbId, localDbName); } else { Env.getCurrentEnv().getExtMetaCacheMgr().invalidateTable( @@ -234,7 +234,7 @@ public void replayRefreshTable(ExternalObjectLog log) { db.get().invalidateTableLookup(localTableName); } } finally { - if (log.isInvalidateTableLookup() && catalog.getLowerCaseTableNames() == 2) { + if (log.isInvalidateTableLookup() && catalog.getLowerCaseTableNames() != 0) { Env.getCurrentEnv().getExtMetaCacheMgr().invalidateDb( log.getCatalogId(), dbId, db.get().getFullName()); } else { @@ -321,7 +321,7 @@ public void refreshTableInternal(ExternalDatabase db, ExternalTable table, long private void refreshTableInternal(ExternalDatabase db, ExternalTable table, long updateTime, boolean invalidateTableLookup) { - boolean invalidateDatabase = invalidateTableLookup && table.getCatalog().getLowerCaseTableNames() == 2; + boolean invalidateDatabase = invalidateTableLookup && table.getCatalog().getLowerCaseTableNames() != 0; table.unsetObjectCreated(); if (updateTime > 0) { table.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 3934483b97bf48..d828add101d375 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 @@ -1190,6 +1190,14 @@ public void gsonPostProcess() throws IOException { this.dbIdNameIndex = new IdNameIndex("external database"); } + /** Cache-only identity fence for a read through a retained database instance. */ + public Optional observeDatabase(ExternalDatabase database) { + if (!initialized || databases == null) { + return Optional.empty(); + } + return databases.observe(database.getFullName(), database); + } + public void addDatabaseForTest(ExternalDatabase db) { buildMetaCache(); // Test helpers only seed object/id state and keep names cache cold unless the test fills it explicitly. 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 afdcb9e0d2df4c..c874f46f07b7e3 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 @@ -32,6 +32,7 @@ import org.apache.doris.common.util.DebugPointUtil; import org.apache.doris.common.util.Util; import org.apache.doris.connector.cache.CacheSpec; +import org.apache.doris.connector.cache.MetaCacheCleanup; import org.apache.doris.datasource.infoschema.ExternalInfoSchemaDatabase; import org.apache.doris.datasource.infoschema.ExternalMysqlDatabase; import org.apache.doris.datasource.metacache.FeMetaCacheEntry; @@ -318,8 +319,10 @@ public T buildTableForInit(String remoteTableName, String localTableName, long t } } - // Step 4: Build and return the table instance using the resolved names and other parameters - return buildTableInternal(remoteTableName, localTableName, tblId, catalog, db); + // A successful object reload is a metadata refresh even if the caller never requests its schema. + T table = buildTableInternal(remoteTableName, localTableName, tblId, catalog, db); + table.setUpdateTime(System.currentTimeMillis()); + return table; } protected abstract T buildTableInternal(String remoteTableName, String localTableName, long tblId, @@ -347,6 +350,42 @@ public Optional getTableForReplay(long tableId) { return Optional.ofNullable(tables.getIfPresent(tableName)); } + /** + * Captures only a shared base table with a matching retained names snapshot. Guard order is catalog DB, + * table names, then table object, matching incremental name/object publication. No remote load is performed. + */ + public Optional observeTableLookup(ExternalTable table, Runnable invalidateMetadata) { + if (!initialized || extCatalog.shouldBypassSchemaCache(SessionContext.current())) { + return Optional.empty(); + } + NameCacheValue names = getTableNamesValue(false); + if (names == null || !table.getRemoteName().equals(names.remoteNameOfLocalName(table.getName()))) { + return Optional.empty(); + } + return extCatalog.observeDatabase(this).flatMap(database -> + tableNames.observe("", names).flatMap(membership -> + tables.observe(table.getName(), table).map(object -> () -> MetaCacheCleanup.runDeferred(() -> { + // Catalog reset/connector replacement holds this monitor too. Only local invalidation + // runs here; remote reads happened before entering this guarded commit. + synchronized (extCatalog) { + if (!initialized || !extCatalog.isInitialized() + || Env.getCurrentEnv().getCatalogMgr().getCatalog(extCatalog.getId()) + != extCatalog) { + return; + } + database.runIfCurrent(() -> membership.invalidateIfCurrent(() -> + object.invalidateIfCurrent(() -> { + try { + invalidateMetadata.run(); + } finally { + tableIdNameIndex.removeName(table.getName()); + } + return true; + }))); + } + })))); + } + /** * Resolve the retained local table name for replay without loading table metadata. * @@ -544,6 +583,10 @@ public T getTableNullable(String tableName) { } long expectedTableId = Util.genIdByName(extCatalog.getName(), name, finalName); tableIdNameIndex.checkCanPut(expectedTableId, finalName); + return getTableAndPublishId(finalName); + } + + private T getTableAndPublishId(String finalName) { return tables.getAndRunIfCurrent( finalName, (localTableName, table) -> !localTableName.equals(tableIdNameIndex.getName(table.getId())), @@ -748,9 +791,9 @@ public void invalidateTableLookup(String tableName) { if (!isInitialized()) { return; } - // In case-insensitive lookup mode an older spelling may still be loading while a refreshed - // names snapshot already contains the new spelling. Fence every object/ID load in this database. - if (isTableNamesCaseInsensitive()) { + // Folding or comparing local names without case can hide another retained remote spelling. + // Fence every object/ID load in this database when refreshing that lookup identity. + if (isStoredTableNamesLowerCase() || isTableNamesCaseInsensitive()) { invalidateAllTableCache(); return; } @@ -780,7 +823,7 @@ public T getTableNullable(long tableId) { if (tableName == null) { return null; } - return tables.get(tableName); + return getTableAndPublishId(tableName); } public long getLastUpdateTime() { diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/metacache/FeMetaCacheEntry.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/metacache/FeMetaCacheEntry.java index 4c140171110be4..13a4a428de732b 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/datasource/metacache/FeMetaCacheEntry.java +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/metacache/FeMetaCacheEntry.java @@ -34,12 +34,14 @@ import java.util.List; import java.util.Map; import java.util.Objects; +import java.util.Optional; import java.util.concurrent.ExecutorService; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicReferenceArray; import java.util.function.BiConsumer; import java.util.function.BiFunction; import java.util.function.BiPredicate; +import java.util.function.BooleanSupplier; import java.util.function.Function; import java.util.function.Predicate; import javax.annotation.Nullable; @@ -255,6 +257,49 @@ public V getAndRunIfCurrent(K key, BiPredicate actionRequired, } } + public Optional observe(K key, Object expectedValue) { + StripeState stripe = stripeState(key); + synchronized (stripe) { + return data.observe(key, expectedValue) + .map(value -> new Observation(value, stripe, () -> bumpAction(stripe, key))); + } + } + + /** Orders conditional removal with both cache publication and auxiliary identity actions. */ + public static final class Observation { + private final MetaCache.Observation value; + private final Object stripe; + private final Runnable invalidateActions; + + private Observation(MetaCache.Observation value, Object stripe, Runnable invalidateActions) { + this.value = value; + this.stripe = stripe; + this.invalidateActions = invalidateActions; + } + + public boolean runIfCurrent(BooleanSupplier action) { + synchronized (stripe) { + return value.runIfCurrent(action); + } + } + + public boolean invalidateIfCurrent(BooleanSupplier action) { + synchronized (stripe) { + return value.invalidateIfCurrent(() -> { + boolean invalidate = true; + try { + invalidate = action.getAsBoolean(); + return invalidate; + } finally { + if (invalidate) { + invalidateActions.run(); + } + } + }); + } + } + } + public V getIfPresent(K key) { return data.getIfPresent(key); } diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/plugin/PluginDrivenExternalTable.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/plugin/PluginDrivenExternalTable.java index 6cbea40af7505b..8ccdda1e72cec4 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/datasource/plugin/PluginDrivenExternalTable.java +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/plugin/PluginDrivenExternalTable.java @@ -119,8 +119,22 @@ public PluginDrivenExternalTable(long id, String name, String remoteName, */ public Optional resolveConnectorTableHandle( ConnectorSession session, ConnectorMetadata metadata) { + PluginDrivenExternalCatalog owner = (PluginDrivenExternalCatalog) catalog; + Connector connector = owner.getConnector(); + ExternalDatabase database = getDb(); + Optional observation = database.observeTableLookup(this, () -> { + try { + connector.invalidateTable(db.getRemoteName(), getRemoteName()); + } finally { + Env.getCurrentEnv().getExtMetaCacheMgr().invalidateTable(this); + } + }); String dbName = db != null ? db.getRemoteName() : ""; - return metadata.getTableHandle(session, dbName, getRemoteName()); + Optional handle = metadata.getTableHandle(session, dbName, getRemoteName()); + if (!handle.isPresent()) { + observation.ifPresent(Runnable::run); + } + return handle; } /** 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 index cbed01bb2aa075..ee0fd392315653 100644 --- 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 @@ -47,6 +47,8 @@ import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; import org.mockito.ArgumentCaptor; import org.mockito.InOrder; import org.mockito.MockedConstruction; @@ -218,9 +220,10 @@ public void testExplicitRefreshReplayClearsNamesWhenTableObjectIsCold() { Assertions.assertTrue(database.getTableNamesWithLock().isEmpty()); } - @Test - public void testExplicitRefreshReplayInvalidatesDatabaseInCaseInsensitiveMode() { - useMutableCatalog(2); + @ParameterizedTest + @ValueSource(ints = {1, 2}) + public void testExplicitRefreshReplayInvalidatesDatabaseInCaseInsensitiveMode(int mode) { + useMutableCatalog(mode); Assertions.assertNotNull(database.getTableNullable(TABLE_NAME)); ExternalObjectLog log = ExternalObjectLog.createForExplicitRefreshTable( CATALOG_ID, DATABASE_NAME, TABLE_NAME, 123L); @@ -232,9 +235,10 @@ public void testExplicitRefreshReplayInvalidatesDatabaseInCaseInsensitiveMode() Mockito.verify(metaCacheMgr).invalidateDb(CATALOG_ID, DATABASE_ID, DATABASE_NAME); } - @Test - public void testExplicitRefreshReplayInvalidatesColdDatabaseInCaseInsensitiveMode() { - useMutableCatalog(2); + @ParameterizedTest + @ValueSource(ints = {1, 2}) + public void testExplicitRefreshReplayInvalidatesColdDatabaseInCaseInsensitiveMode(int mode) { + useMutableCatalog(mode); catalog.setInitializedForTest(false); new RefreshManager().replayRefreshTable(ExternalObjectLog.createForExplicitRefreshTable( @@ -243,9 +247,10 @@ public void testExplicitRefreshReplayInvalidatesColdDatabaseInCaseInsensitiveMod Mockito.verify(metaCacheMgr).invalidateDb(CATALOG_ID, DATABASE_ID, DATABASE_NAME); } - @Test - public void testExplicitRefreshReplayInvalidatesConnectorDatabaseInCaseInsensitiveMode() { - PluginCatalogFixture fixture = usePluginCatalog(2); + @ParameterizedTest + @ValueSource(ints = {1, 2}) + public void testExplicitRefreshReplayInvalidatesConnectorDatabaseInCaseInsensitiveMode(int mode) { + PluginCatalogFixture fixture = usePluginCatalog(mode); ExternalTable oldTable = new ExternalTable(TABLE_ID, TABLE_NAME, TABLE_NAME, fixture.catalog, fixture.database, TableIf.TableType.PLUGIN_EXTERNAL_TABLE); String otherSpelling = TABLE_NAME.toUpperCase(java.util.Locale.ROOT); diff --git a/fe/fe-core/src/test/java/org/apache/doris/catalog/RefreshTableTest.java b/fe/fe-core/src/test/java/org/apache/doris/catalog/RefreshTableTest.java index b9935acbd342c6..40252053a5d5a8 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/catalog/RefreshTableTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/catalog/RefreshTableTest.java @@ -106,6 +106,8 @@ public void testRefreshTable() throws Exception { Assertions.assertTrue(l3 >= l2); TestExternalTable refreshed = (TestExternalTable) test1.getDbNullable("db1").getTable("tbl11").get(); Assertions.assertNotSame(table, refreshed); + // information_schema reads UPDATE_TIME without forcing schema loading. + Assertions.assertTrue(refreshed.getUpdateTime() >= l3); Assertions.assertSame(refreshed, test1.getDbNullable("db1").getTableNullable(refreshed.getId())); refreshed.getFullSchema(); // Schema loading updates the replacement table, not the invalidated table object. diff --git a/fe/fe-core/src/test/java/org/apache/doris/datasource/plugin/PluginDrivenMissingTableTest.java b/fe/fe-core/src/test/java/org/apache/doris/datasource/plugin/PluginDrivenMissingTableTest.java new file mode 100644 index 00000000000000..9e0e9f5ae0bb39 --- /dev/null +++ b/fe/fe-core/src/test/java/org/apache/doris/datasource/plugin/PluginDrivenMissingTableTest.java @@ -0,0 +1,550 @@ +// 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.plugin; + +import org.apache.doris.catalog.Column; +import org.apache.doris.catalog.Env; +import org.apache.doris.catalog.RefreshManager; +import org.apache.doris.common.FeConstants; +import org.apache.doris.common.jmockit.Deencapsulation; +import org.apache.doris.common.util.Util; +import org.apache.doris.connector.spi.Connector; +import org.apache.doris.connector.spi.ConnectorColumn; +import org.apache.doris.connector.spi.ConnectorMetadata; +import org.apache.doris.connector.spi.ConnectorSession; +import org.apache.doris.connector.spi.ConnectorTableSchema; +import org.apache.doris.connector.spi.ConnectorType; +import org.apache.doris.connector.spi.handle.ConnectorTableHandle; +import org.apache.doris.datasource.CacheException; +import org.apache.doris.datasource.CatalogIf; +import org.apache.doris.datasource.CatalogMgr; +import org.apache.doris.datasource.ExternalCatalog; +import org.apache.doris.datasource.ExternalDatabase; +import org.apache.doris.datasource.SchemaCacheKey; +import org.apache.doris.datasource.SchemaCacheValue; +import org.apache.doris.datasource.SessionContext; +import org.apache.doris.datasource.metacache.ExternalMetaCache; +import org.apache.doris.datasource.metacache.FeMetaCacheEntry; +import org.apache.doris.datasource.metacache.IdNameIndex; +import org.apache.doris.datasource.metacache.NameCacheValue; +import org.apache.doris.utframe.TestWithFeService; + +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.mockito.Mockito; + +import java.util.ArrayList; +import java.util.Collections; +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; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicLong; + +/** Tests missing-table repair through real FE lookup/schema caches and a controllable connector. */ +public class PluginDrivenMissingTableTest extends TestWithFeService { + private static final AtomicLong IDS = new AtomicLong(910000); + private final Map remoteTables = new ConcurrentHashMap<>(); + private Connector connector; + private ConnectorMetadata metadata; + private TestCatalog catalog; + private TestDatabase database; + private final Map remoteCache = new ConcurrentHashMap<>(); + private ExecutorService worker; + private CountDownLatch missingStarted; + private CountDownLatch releaseMissing; + private AtomicBoolean blockMissing; + private boolean failLookup; + private boolean failInvalidation; + + @Override + protected void runBeforeAll() throws Exception { + FeConstants.runningUnitTest = true; + createDefaultCtx().setThreadLocalInfo(); + } + + @BeforeEach + void setup() { + failLookup = false; + failInvalidation = false; + remoteTables.put("a", "old_column"); + remoteTables.put("b", "sibling_column"); + missingStarted = new CountDownLatch(1); + releaseMissing = new CountDownLatch(1); + blockMissing = new AtomicBoolean(); + worker = Executors.newFixedThreadPool(2); + connector = Mockito.mock(Connector.class); + metadata = Mockito.mock(ConnectorMetadata.class); + Mockito.when(connector.getMetadata(Mockito.any())).thenReturn(metadata); + Mockito.doAnswer(call -> { + if (failInvalidation) { + throw new IllegalStateException("connector cleanup failed"); + } + remoteCache.remove(call.getArgument(1)); + return null; + }).when(connector).invalidateTable(Mockito.anyString(), Mockito.anyString()); + Mockito.doAnswer(call -> { + remoteCache.clear(); + return null; + }).when(connector).invalidateDb(Mockito.anyString()); + Mockito.when(metadata.getTableHandle(Mockito.any(), Mockito.anyString(), Mockito.anyString())) + .thenAnswer(call -> { + if (failLookup) { + throw new IllegalStateException("metastore timeout"); + } + String name = call.getArgument(2); + String column = remoteTables.get(name); + if (column == null) { + if (blockMissing.compareAndSet(true, false)) { + missingStarted.countDown(); + Assertions.assertTrue(releaseMissing.await(10, TimeUnit.SECONDS)); + } + return Optional.empty(); + } + return Optional.of(new Handle(name, column)); + }); + Mockito.when(metadata.getTableSchema(Mockito.any(), Mockito.any())).thenAnswer(call -> { + Handle handle = call.getArgument(1); + return new ConnectorTableSchema(handle.name, Collections.singletonList(new ConnectorColumn( + handle.column, ConnectorType.of("INT", -1, -1), "", true, null, true)), + null, Collections.emptyMap()); + }); + Mockito.when(metadata.fromRemoteColumnName(Mockito.any(), Mockito.anyString(), + Mockito.anyString(), Mockito.anyString())).thenAnswer(call -> call.getArgument(3)); + Mockito.when(metadata.fromRemoteTableName(Mockito.any(), Mockito.anyString(), Mockito.anyString())) + .thenAnswer(call -> call.getArgument(2)); + Mockito.when(metadata.fromRemoteDatabaseName(Mockito.any(), Mockito.anyString())) + .thenAnswer(call -> call.getArgument(1)); + catalog = new TestCatalog(IDS.incrementAndGet()); + CatalogMgr manager = Env.getCurrentEnv().getCatalogMgr(); + manager.getIdToCatalog().put(catalog.getId(), catalog); + Map> byName = Deencapsulation.getField(manager, "nameToCatalog"); + byName.put(catalog.getName(), catalog); + installDatabase(); + } + + @AfterEach + void cleanup() throws Exception { + releaseMissing.countDown(); + worker.shutdownNow(); + Assertions.assertTrue(worker.awaitTermination(10, TimeUnit.SECONDS)); + Env.getCurrentEnv().getExtMetaCacheMgr().removeCatalog(catalog.getId()); + CatalogMgr manager = Env.getCurrentEnv().getCatalogMgr(); + manager.getIdToCatalog().remove(catalog.getId()); + Map> byName = Deencapsulation.getField(manager, "nameToCatalog"); + byName.remove(catalog.getName()); + remoteTables.clear(); + remoteCache.clear(); + } + + private void installDatabase() { + catalog.setInitializedForTest(true); + database = new TestDatabase(); + database.setInitializedForTest(true); + catalog.addDatabaseForTest(database); + } + + private org.apache.doris.connector.cache.MetaCache schemas() { + ExternalMetaCache cache = Deencapsulation.invoke(Env.getCurrentEnv().getExtMetaCacheMgr(), "engine", "default"); + return cache.entry(catalog.getId(), "schema", SchemaCacheKey.class, SchemaCacheValue.class); + } + + private PluginDrivenExternalTable warmTarget() { + Assertions.assertEquals(Set.of("a", "b"), database.getTableNamesWithLock()); + PluginDrivenExternalTable table = database.getTableNullable("a"); + table.getFullSchema(); + database.getTableNullable("b").getFullSchema(); + remoteCache.put("a", "old"); + remoteCache.put("b", "sibling"); + // Keep membership/object entries hot while forcing the next schema read to consult the source. + schemas().invalidateKey(new SchemaCacheKey(table.getOrBuildNameMapping())); + return table; + } + + @Test + void missingTableRepairsNamesObjectsIdsAndConnectorMetadata() { + PluginDrivenExternalTable old = warmTarget(); + PluginDrivenExternalTable sibling = database.getTableNullable("b"); + List siblingSchema = sibling.getFullSchema(); + remoteTables.remove("a"); + Assertions.assertThrows(CacheException.class, old::getFullSchema); + Assertions.assertEquals(Set.of("b"), database.getTableNamesWithLock()); + Assertions.assertNull(database.getTableNullable(old.getId())); + Assertions.assertNull(remoteCache.get("a")); + Assertions.assertEquals("sibling", remoteCache.get("b")); + Assertions.assertSame(sibling, database.getTableNullable("b")); + Assertions.assertSame(siblingSchema, sibling.getFullSchema()); + } + + @Test + void missingHandleRepairsNamesBeforeSchemaLoading() { + PluginDrivenExternalTable table = warmTarget(); + remoteTables.remove("a"); + ConnectorSession session = catalog.buildCrossStatementSession(); + try { + Assertions.assertFalse(table.resolveConnectorTableHandle(session, metadata).isPresent()); + } finally { + session.getStatementScope().closeAll(); + } + Assertions.assertEquals(Set.of("b"), database.getTableNamesWithLock()); + Assertions.assertNull(database.getTableNullable(table.getId())); + Assertions.assertNull(remoteCache.get("a")); + } + + @Test + void delayedMissingDoesNotInvalidateRecreatedTable() throws Exception { + verifyDelayedMissing(false); + } + + @Test + void delayedMissingDoesNotInvalidateExactKeyReplacement() throws Exception { + verifyDelayedMissing(true); + } + + private void verifyDelayedMissing(boolean replaceWithoutScopeInvalidation) throws Exception { + PluginDrivenExternalTable old = warmTarget(); + remoteTables.remove("a"); + blockMissing.set(true); + Future result = worker.submit(() -> Assertions.assertThrows(CacheException.class, + old::getFullSchema)); + Assertions.assertTrue(missingStarted.await(10, TimeUnit.SECONDS)); + remoteTables.put("a", "new_column"); + PluginDrivenExternalTable replacement; + if (replaceWithoutScopeInvalidation) { + replacement = database.buildTableInternal("a", "a", old.getId(), catalog, database); + database.addTableForTest(replacement); + schemas().put(new SchemaCacheKey(replacement.getOrBuildNameMapping()), replacement.initSchema().get()); + } else { + new RefreshManager().handleRefreshTable(catalog.getName(), "db", "a", false); + replacement = database.getTableNullable("a"); + } + List newSchema = replacement.getFullSchema(); + remoteCache.put("a", "new"); + database.getTableNamesWithLock(); + int nameLoads = catalog.nameLoads.get(); + releaseMissing.countDown(); + result.get(10, TimeUnit.SECONDS); + Assertions.assertSame(replacement, database.getTableNullable("a")); + Assertions.assertSame(replacement, database.getTableNullable(replacement.getId())); + Assertions.assertSame(newSchema, replacement.getFullSchema()); + Assertions.assertTrue(replacement.isObjectCreated()); + Assertions.assertEquals("new_column", newSchema.get(0).getName()); + Assertions.assertEquals("new", remoteCache.get("a")); + Assertions.assertEquals(Set.of("a", "b"), database.getTableNamesWithLock()); + Assertions.assertEquals(nameLoads, catalog.nameLoads.get()); + } + + @Test + void delayedMissingDoesNotInvalidateReplacementDatabase() throws Exception { + PluginDrivenExternalTable old = warmTarget(); + remoteTables.remove("a"); + blockMissing.set(true); + Future result = worker.submit(() -> readMissingHandle(old)); + Assertions.assertTrue(missingStarted.await(10, TimeUnit.SECONDS)); + remoteTables.put("a", "replacement_column"); + database = new TestDatabase(); + database.setInitializedForTest(true); + catalog.addDatabaseForTest(database); + PluginDrivenExternalTable replacement = database.getTableNullable("a"); + List schema = replacement.getFullSchema(); + remoteCache.put("a", "replacement"); + database.getTableNamesWithLock(); + int loads = catalog.nameLoads.get(); + releaseMissing.countDown(); + result.get(10, TimeUnit.SECONDS); + Assertions.assertSame(replacement, database.getTableNullable("a")); + Assertions.assertSame(schema, replacement.getFullSchema()); + Assertions.assertEquals("replacement", remoteCache.get("a")); + Assertions.assertEquals(loads, catalog.nameLoads.get()); + } + + @Test + void delayedMissingDoesNotInvalidateReplacementCatalog() throws Exception { + PluginDrivenExternalTable old = warmTarget(); + remoteTables.remove("a"); + blockMissing.set(true); + Future result = worker.submit(() -> readMissingHandle(old)); + Assertions.assertTrue(missingStarted.await(10, TimeUnit.SECONDS)); + long catalogId = catalog.getId(); + catalog = new TestCatalog(catalogId); + CatalogMgr manager = Env.getCurrentEnv().getCatalogMgr(); + manager.getIdToCatalog().put(catalogId, catalog); + Map> byName = Deencapsulation.getField(manager, "nameToCatalog"); + byName.put(catalog.getName(), catalog); + remoteTables.put("a", "replacement_column"); + installDatabase(); + PluginDrivenExternalTable replacement = database.getTableNullable("a"); + List schema = replacement.getFullSchema(); + remoteCache.put("a", "replacement"); + database.getTableNamesWithLock(); + int loads = catalog.nameLoads.get(); + releaseMissing.countDown(); + result.get(10, TimeUnit.SECONDS); + Assertions.assertSame(replacement, database.getTableNullable("a")); + Assertions.assertSame(schema, replacement.getFullSchema()); + Assertions.assertEquals("replacement", remoteCache.get("a")); + Assertions.assertEquals(loads, catalog.nameLoads.get()); + } + + @Test + void failedConnectorCleanupStillRemovesObservedLookup() { + PluginDrivenExternalTable old = warmTarget(); + remoteTables.remove("a"); + failInvalidation = true; + Assertions.assertThrows(IllegalStateException.class, () -> readMissingHandle(old)); + Assertions.assertEquals(Set.of("b"), database.getTableNamesWithLock()); + Assertions.assertNull(database.getTableNullable(old.getId())); + } + + private void readMissingHandle(PluginDrivenExternalTable table) { + ConnectorSession session = catalog.buildCrossStatementSession(); + try { + Assertions.assertFalse(table.resolveConnectorTableHandle(session, metadata).isPresent()); + } finally { + session.getStatementScope().closeAll(); + } + } + + @Test + void changedCanonicalSpellingFencesBothExistingAndNewOldObjectReads() throws Exception { + catalog.lowerCaseMode = 2; + try { + remoteTables.remove("a"); + remoteTables.put("Foo", "old_column"); + PluginDrivenExternalTable old = database.getTableNullable("foo"); + old.getFullSchema(); + schemas().invalidateKey(new SchemaCacheKey(old.getOrBuildNameMapping())); + remoteTables.remove("Foo"); + blockMissing.set(true); + Future missing = worker.submit(() -> Assertions.assertThrows(CacheException.class, + old::getFullSchema)); + Assertions.assertTrue(missingStarted.await(10, TimeUnit.SECONDS)); + remoteTables.put("FOO", "new_column"); + FeMetaCacheEntry names = Deencapsulation.getField(database, "tableNames"); + names.invalidateKey(""); + PluginDrivenExternalTable replacement = database.getTableNullable("foo"); + Assertions.assertEquals("FOO", replacement.getName()); + List schema = replacement.getFullSchema(); + remoteCache.put("FOO", "new"); + int loads = catalog.nameLoads.get(); + releaseMissing.countDown(); + missing.get(10, TimeUnit.SECONDS); + Assertions.assertSame(replacement, database.getTableNullable("foo")); + Assertions.assertSame(schema, replacement.getFullSchema()); + Assertions.assertEquals("new", remoteCache.get("FOO")); + Assertions.assertEquals(loads, catalog.nameLoads.get()); + // A fresh request holding the old object must not combine it with the new names snapshot. + ConnectorSession session = catalog.buildCrossStatementSession(); + try { + Assertions.assertFalse(old.resolveConnectorTableHandle(session, metadata).isPresent()); + } finally { + session.getStatementScope().closeAll(); + } + Assertions.assertSame(replacement, database.getTableNullable("foo")); + Assertions.assertSame(schema, replacement.getFullSchema()); + Assertions.assertEquals("new", remoteCache.get("FOO")); + } finally { + catalog.lowerCaseMode = 0; + } + } + + @Test + void lookupFailureDoesNotRepairNamesOrEvictObjects() { + PluginDrivenExternalTable table = warmTarget(); + remoteTables.remove("a"); + failLookup = true; + Assertions.assertThrows(IllegalStateException.class, table::getFullSchema); + Assertions.assertEquals(Set.of("a", "b"), database.getTableNamesWithLock()); + Assertions.assertSame(table, database.getTableNullable(table.getId())); + Assertions.assertEquals("old", remoteCache.get("a")); + Assertions.assertEquals(1, catalog.nameLoads.get()); + } + + @Test + void numericLookupCapturedBeforeRefreshRepublishesItsIdentity() throws Exception { + PluginDrivenExternalTable old = warmTarget(); + IdNameIndex original = Deencapsulation.getField(database, "tableIdNameIndex"); + IdNameIndex index = Mockito.spy(original); + Deencapsulation.setField(database, "tableIdNameIndex", index); + CountDownLatch captured = new CountDownLatch(1); + CountDownLatch proceed = new CountDownLatch(1); + AtomicBoolean pause = new AtomicBoolean(true); + Mockito.doAnswer(call -> { + String name = original.getName(old.getId()); + if (pause.compareAndSet(true, false)) { + captured.countDown(); + Assertions.assertTrue(proceed.await(10, TimeUnit.SECONDS)); + } + return name; + }).when(index).getName(old.getId()); + Future lookup = worker.submit(() -> database.getTableNullable(old.getId())); + try { + Assertions.assertTrue(captured.await(10, TimeUnit.SECONDS)); + remoteTables.put("a", "new_column"); + new RefreshManager().handleRefreshTable(catalog.getName(), "db", "a", false); + Assertions.assertNull(original.getName(old.getId())); + proceed.countDown(); + PluginDrivenExternalTable replacement = lookup.get(10, TimeUnit.SECONDS); + Assertions.assertNotSame(old, replacement); + Assertions.assertEquals("a", index.getName(old.getId())); + Assertions.assertSame(replacement, database.getTableNullable(old.getId())); + } finally { + proceed.countDown(); + } + } + + @Test + void foldedNamesRefreshClearsHistoricalRemoteSpellings() throws Exception { + catalog.lowerCaseMode = 1; + remoteTables.remove("a"); + remoteTables.put("Foo", "first"); + Assertions.assertNotNull(database.getTableNullable("foo")); + remoteCache.put("Foo", "first"); + remoteTables.remove("Foo"); + remoteTables.put("FOO", "second"); + FeMetaCacheEntry names = Deencapsulation.getField(database, "tableNames"); + names.invalidateKey(""); + database.evictTableObjectForTest("foo"); + PluginDrivenExternalTable second = database.getTableNullable("foo"); + Assertions.assertEquals("FOO", second.getRemoteName()); + remoteCache.put("FOO", "second"); + remoteTables.remove("FOO"); + remoteTables.put("Foo", "third"); + new RefreshManager().handleRefreshTable(catalog.getName(), "db", "foo", false); + Assertions.assertTrue(remoteCache.isEmpty()); + PluginDrivenExternalTable replacement = database.getTableNullable("foo"); + Assertions.assertEquals("Foo", replacement.getRemoteName()); + Assertions.assertEquals("third", replacement.getFullSchema().get(0).getName()); + } + + @Test + void ordinaryWritesDoNotReloadNames() { + verifyOrdinaryWrite(0); + } + + @Test + void foldedNameOrdinaryWritesDoNotReloadNames() { + verifyOrdinaryWrite(1); + } + + @Test + void caseInsensitiveOrdinaryWritesDoNotReloadNames() { + verifyOrdinaryWrite(2); + } + + private void verifyOrdinaryWrite(int mode) { + catalog.lowerCaseMode = mode; + PluginDrivenExternalTable table = warmTarget(); + int loads = catalog.nameLoads.get(); + new RefreshManager().refreshTableInternal(database, table, 123L); + Assertions.assertEquals(Set.of("a", "b"), database.getTableNamesWithLock()); + Assertions.assertSame(table, database.getTableNullable("a")); + Assertions.assertEquals(loads, catalog.nameLoads.get()); + } + + @Test + void sessionBypassDoesNotInvalidateSharedNames() { + PluginDrivenExternalTable table = warmTarget(); + remoteTables.remove("a"); + catalog.bypassSchema = true; + readMissingHandle(table); + Assertions.assertEquals(Set.of("a", "b"), database.getTableNamesWithLock()); + Assertions.assertSame(table, database.getTableNullable(table.getId())); + Assertions.assertEquals("old", remoteCache.get("a")); + } + + private static class Handle implements ConnectorTableHandle { + private final String name; + private final String column; + + Handle(String name, String column) { + this.name = name; + this.column = column; + } + } + + private class TestCatalog extends PluginDrivenExternalCatalog { + private final AtomicInteger nameLoads = new AtomicInteger(); + private boolean bypassSchema; + private int lowerCaseMode; + + TestCatalog(long id) { + super(id, "missing_test_" + id, "", new HashMap<>(), "", connector); + } + + @Override + protected void initLocalObjectsImpl() { + // The fixture injects the connector directly instead of loading a production plugin. + } + + @Override + public int getLowerCaseTableNames() { + return lowerCaseMode; + } + + @Override + public Connector getConnector() { + return connector; + } + + @Override + protected List listDatabaseNames() { + return Collections.singletonList("db"); + } + + @Override + protected List listTableNamesFromRemote(SessionContext context, String dbName) { + nameLoads.incrementAndGet(); + return new ArrayList<>(remoteTables.keySet()); + } + + @Override + public boolean shouldBypassSchemaCache(SessionContext context) { + return bypassSchema; + } + } + + private class TestDatabase extends ExternalDatabase { + TestDatabase() { + super(catalog, Util.genIdByName(catalog.getName(), "db"), "db", "db"); + } + + @Override + protected PluginDrivenExternalTable buildTableInternal(String remoteName, String localName, long id, + ExternalCatalog owner, ExternalDatabase db) { + return new PluginDrivenExternalTable(id, localName, remoteName, owner, db) { + @Override + protected synchronized void makeSureInitialized() { + // The fixture registers this database explicitly; connector schema loading remains real. + objectCreated = true; + } + }; + } + } +} diff --git a/regression-test/data/external_table_p0/hive/test_hive_external_lookup_refresh.out b/regression-test/data/external_table_p0/hive/test_hive_external_lookup_refresh.out new file mode 100644 index 00000000000000..7730655d8510a0 --- /dev/null +++ b/regression-test/data/external_table_p0/hive/test_hive_external_lookup_refresh.out @@ -0,0 +1,35 @@ +-- This file is automatically generated. You should know what you did if you want to edit this +-- !names_before -- +owner_sibling +owner_target + +-- !schema_before -- +before_col int Yes true \N + +-- !sibling_before -- +sibling_col int Yes true \N + +-- !names_after_drop -- +owner_sibling + +-- !sibling_after -- +sibling_col int Yes true \N + +-- !names_after_database_refresh -- +owner_sibling +owner_target + +-- !schema_after_database_refresh -- +after_col text Yes true \N + +-- !schema_after_recreation -- +recreated_col bigint Yes true \N + +-- !observed_names_before -- +owner_observed + +-- !observed_schema_before -- +observed_col int Yes true \N + +-- !observed_names_after -- + diff --git a/regression-test/suites/external_table_p0/hive/test_hive_external_lookup_refresh.groovy b/regression-test/suites/external_table_p0/hive/test_hive_external_lookup_refresh.groovy new file mode 100644 index 00000000000000..3d3ce7135e3902 --- /dev/null +++ b/regression-test/suites/external_table_p0/hive/test_hive_external_lookup_refresh.groovy @@ -0,0 +1,93 @@ +// 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. + +suite("test_hive_external_lookup_refresh", "p0,external") { + String enabled = context.config.otherConfigs.get("enableHiveTest") + if (enabled == null || !enabled.equalsIgnoreCase("true")) { + logger.info("disable Hive test.") + return + } + + String externalEnvIp = context.config.otherConfigs.get("externalEnvIp") + String hmsPort = context.config.otherConfigs.get("hive3HmsPort") + String hdfsPort = context.config.otherConfigs.get("hive3HdfsPort") + for (String catalog : ["hive_lookup_source", "hive_lookup_target"]) { + sql "DROP CATALOG IF EXISTS ${catalog}" + sql """CREATE CATALOG ${catalog} PROPERTIES ( + 'type' = 'hms', + 'hadoop.username' = 'hadoop', + 'hive.metastore.uris' = 'thrift://${externalEnvIp}:${hmsPort}', + 'fs.defaultFS' = 'hdfs://${externalEnvIp}:${hdfsPort}' + )""" + } + + // Two independent catalogs model a change made through another engine's cache domain. + sql "DROP DATABASE IF EXISTS hive_lookup_source.fe_lookup_refresh_db FORCE" + sql "CREATE DATABASE hive_lookup_source.fe_lookup_refresh_db" + sql "CREATE TABLE hive_lookup_source.fe_lookup_refresh_db.owner_target (before_col INT)" + sql "CREATE TABLE hive_lookup_source.fe_lookup_refresh_db.owner_sibling (sibling_col INT)" + order_qt_names_before "SHOW TABLES FROM hive_lookup_target.fe_lookup_refresh_db" + qt_schema_before "DESC hive_lookup_target.fe_lookup_refresh_db.owner_target" + qt_sibling_before "DESC hive_lookup_target.fe_lookup_refresh_db.owner_sibling" + + sql "DROP TABLE hive_lookup_source.fe_lookup_refresh_db.owner_target" + sql "REFRESH TABLE hive_lookup_target.fe_lookup_refresh_db.owner_target" + // Listing first proves refresh itself invalidated membership, without a failing SELECT repairing it. + order_qt_names_after_drop "SHOW TABLES FROM hive_lookup_target.fe_lookup_refresh_db" + test { + sql "SELECT * FROM hive_lookup_target.fe_lookup_refresh_db.owner_target" + exception "does not exist" + } + test { + sql "EXPLAIN SELECT * FROM hive_lookup_target.fe_lookup_refresh_db.owner_target" + exception "does not exist" + } + qt_sibling_after "DESC hive_lookup_target.fe_lookup_refresh_db.owner_sibling" + + sql "CREATE TABLE hive_lookup_source.fe_lookup_refresh_db.owner_target (after_col STRING)" + sql "REFRESH DATABASE hive_lookup_target.fe_lookup_refresh_db" + order_qt_names_after_database_refresh "SHOW TABLES FROM hive_lookup_target.fe_lookup_refresh_db" + qt_schema_after_database_refresh "DESC hive_lookup_target.fe_lookup_refresh_db.owner_target" + + sql "DROP TABLE hive_lookup_source.fe_lookup_refresh_db.owner_target" + sql "CREATE TABLE hive_lookup_source.fe_lookup_refresh_db.owner_target (recreated_col BIGINT)" + sql "REFRESH TABLE hive_lookup_target.fe_lookup_refresh_db.owner_target" + qt_schema_after_recreation "DESC hive_lookup_target.fe_lookup_refresh_db.owner_target" + + sql "DROP CATALOG IF EXISTS hive_lookup_observed" + sql """CREATE CATALOG hive_lookup_observed PROPERTIES ( + 'type' = 'hms', + 'hadoop.username' = 'hadoop', + 'hive.metastore.uris' = 'thrift://${externalEnvIp}:${hmsPort}', + 'fs.defaultFS' = 'hdfs://${externalEnvIp}:${hdfsPort}', + 'schema.cache.ttl-second' = '0' + )""" + sql "CREATE TABLE hive_lookup_source.fe_lookup_refresh_db.owner_observed (observed_col INT)" + // Schema/HMS table reads are live, while the FE table-name and object caches stay enabled. + order_qt_observed_names_before """SHOW TABLES FROM hive_lookup_observed.fe_lookup_refresh_db + LIKE 'owner_observed'""" + qt_observed_schema_before "DESC hive_lookup_observed.fe_lookup_refresh_db.owner_observed" + sql "DROP TABLE hive_lookup_source.fe_lookup_refresh_db.owner_observed" + test { + sql "SELECT * FROM hive_lookup_observed.fe_lookup_refresh_db.owner_observed" + // MVCC observes absence before a later schema lookup reports the now-missing cached object. + exception "failed to load schema cache value" + } + order_qt_observed_names_after """SHOW TABLES FROM hive_lookup_observed.fe_lookup_refresh_db + LIKE 'owner_observed'""" + +} From 584871c5dd3df5babea5f1daa5f7e9064fc11607 Mon Sep 17 00:00:00 2001 From: 924060929 Date: Fri, 23 Oct 2026 17:13:08 +0800 Subject: [PATCH 6/8] [fix](fe) Preserve cached external views on missing table handles ### What problem does this PR solve? Issue Number: None Problem Summary: Iceberg views have no table handle, but the new missing-table observation treated that normal result as a remote drop and evicted healthy view name, object, ID, and connector metadata. Confirm live view existence before clearing a missing table lookup, while still cleaning a truly dropped view. Update the Paimon authentication test for its explicit error propagation, and cover numeric lookup publication, refresh timestamps, and historical remote spellings with focused tests. ### Release note Avoid unnecessary cache invalidation when querying existing external views. ### Check List (For Author) - Test: Unit Test: 77 targeted Paimon and FE core tests passed via run-fe-ut.sh; Checkstyle passed. - Behavior changed: Yes (healthy views retain external lookup caches when a table handle is absent). - Does this need documentation: No --- .../PaimonConnectorMetadataReadAuthTest.java | 4 +- .../plugin/PluginDrivenExternalTable.java | 4 +- .../doris/catalog/RefreshManagerTest.java | 42 ++++++++++++++ .../datasource/ExternalDatabaseTest.java | 58 +++++++++++++++++++ .../plugin/PluginDrivenMissingTableTest.java | 35 ++++++++++- 5 files changed, 139 insertions(+), 4 deletions(-) diff --git a/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonConnectorMetadataReadAuthTest.java b/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonConnectorMetadataReadAuthTest.java index b85499d9e99fc6..a6c9571a56f60f 100644 --- a/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonConnectorMetadataReadAuthTest.java +++ b/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonConnectorMetadataReadAuthTest.java @@ -156,8 +156,8 @@ public void getTableHandleRunsSeamInsideAuthenticator() { RecordingPaimonCatalogOps ops = new RecordingPaimonCatalogOps(); RecordingConnectorContext ctx = new RecordingConnectorContext(); ctx.failAuth = true; - Optional result = metadata(ops, ctx).getTableHandle(null, "db1", "t1"); - Assertions.assertFalse(result.isPresent()); + Assertions.assertThrows(DorisConnectorException.class, + () -> metadata(ops, ctx).getTableHandle(null, "db1", "t1")); Assertions.assertTrue(ops.log.isEmpty(), "auth failure must abort BEFORE the getTable seam runs"); Assertions.assertEquals(1, ctx.authCount); diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/plugin/PluginDrivenExternalTable.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/plugin/PluginDrivenExternalTable.java index 8ccdda1e72cec4..a461d17c240237 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/datasource/plugin/PluginDrivenExternalTable.java +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/plugin/PluginDrivenExternalTable.java @@ -131,7 +131,9 @@ public Optional resolveConnectorTableHandle( }); String dbName = db != null ? db.getRemoteName() : ""; Optional handle = metadata.getTableHandle(session, dbName, getRemoteName()); - if (!handle.isPresent()) { + // A view has no table handle. Check live view existence before treating an empty handle as a + // dropped table; the cached isView flag can outlive a remote view deletion. + if (!handle.isPresent() && !metadata.viewExists(session, dbName, getRemoteName())) { observation.ifPresent(Runnable::run); } return handle; 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 index ee0fd392315653..dedd1b42b82720 100644 --- 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 @@ -190,6 +190,21 @@ public void testExplicitRefreshReloadsTableNamesAndObject() throws Exception { Assertions.assertTrue(log.getValue().isInvalidateTableLookup()); } + @Test + public void testExplicitRefreshReplayRetainsReplacementUpdateTimeBeforeSchemaLoad() { + useMutableCatalog(); + TestExternalTable oldTable = database.getTableNullable(TABLE_NAME); + Assertions.assertNotNull(oldTable); + long refreshTime = System.currentTimeMillis(); + + new RefreshManager().replayRefreshTable(ExternalObjectLog.createForExplicitRefreshTable( + CATALOG_ID, DATABASE_NAME, TABLE_NAME, refreshTime)); + + TestExternalTable replacement = database.getTableNullable(TABLE_NAME); + Assertions.assertNotSame(oldTable, replacement); + Assertions.assertTrue(replacement.getUpdateTime() >= refreshTime); + } + @Test public void testContentRefreshRetainsTableNamesAndObject() { useMutableCatalog(); @@ -269,6 +284,33 @@ public void testExplicitRefreshReplayInvalidatesConnectorDatabaseInCaseInsensiti Assertions.assertFalse(fixture.database.getTableForReplay(otherSpelling).isPresent()); } + @Test + public void testExplicitRefreshInLowerCaseModeClearsHistoricalRemoteSpellings() { + PluginCatalogFixture fixture = usePluginCatalog(1); + ExternalTable table = new ExternalTable(TABLE_ID, TABLE_NAME, "FOO", + fixture.catalog, fixture.database, TableIf.TableType.PLUGIN_EXTERNAL_TABLE); + fixture.database.addTableForTest(table); + Map connectorEntries = new HashMap<>(); + connectorEntries.put("Foo", "old incarnation"); + connectorEntries.put("FOO", "current incarnation"); + Mockito.doAnswer(inv -> { + connectorEntries.clear(); + return null; + }).when(fixture.connector).invalidateDb(DATABASE_NAME); + Mockito.doAnswer(inv -> { + connectorEntries.remove(inv.getArgument(1)); + return null; + }).when(fixture.connector).invalidateTable(Mockito.eq(DATABASE_NAME), Mockito.anyString()); + + new RefreshManager().replayRefreshTable(ExternalObjectLog.createForExplicitRefreshTable( + CATALOG_ID, DATABASE_NAME, TABLE_NAME, 123L)); + + Assertions.assertTrue(connectorEntries.isEmpty()); + Mockito.verify(fixture.connector).invalidateDb(DATABASE_NAME); + Mockito.verify(fixture.connector, Mockito.never()).invalidateTable(Mockito.anyString(), Mockito.anyString()); + Mockito.verify(metaCacheMgr).invalidateDb(CATALOG_ID, DATABASE_ID, DATABASE_NAME); + } + @Test public void testRefreshTableRemovesRowCountLoadedDuringConnectorInvalidation() throws Exception { PluginCatalogFixture fixture = usePluginCatalog(); diff --git a/fe/fe-core/src/test/java/org/apache/doris/datasource/ExternalDatabaseTest.java b/fe/fe-core/src/test/java/org/apache/doris/datasource/ExternalDatabaseTest.java index 7d16e07e96142a..8c0d1b91a69cfa 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/datasource/ExternalDatabaseTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/datasource/ExternalDatabaseTest.java @@ -55,6 +55,8 @@ import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; +import java.util.function.BiConsumer; +import java.util.function.BiPredicate; public class ExternalDatabaseTest extends TestWithFeService { @Override @@ -309,6 +311,62 @@ protected void beforeCurrentValueActionForTest(String key, TestExternalTable val } } + @Test + public void testNumericLookupAfterRefreshPublishesReplacementId() throws Exception { + ExecutorService refreshExecutor = Executors.newSingleThreadExecutor(); + ExecutorService queryExecutor = Executors.newSingleThreadExecutor(); + CountDownLatch nameCaptured = new CountDownLatch(1); + CountDownLatch releaseLookup = new CountDownLatch(1); + try { + InspectableCatalog catalog = new InspectableCatalog(); + InspectableDatabase db = new InspectableDatabase(catalog, 245L, "db1", "db1"); + db.setInitializedForTest(true); + TestExternalTable oldTable = db.getTableNullable("tbl_base"); + Assertions.assertNotNull(oldTable); + + FeMetaCacheEntry objectEntry = + new FeMetaCacheEntry( + "numeric_lookup_refresh_race", + ignored -> new TestExternalTable(oldTable.getId(), oldTable.getName(), + oldTable.getRemoteName(), catalog, db), + CacheSpec.of(true, CacheSpec.CACHE_NO_TTL, 10L), + refreshExecutor, + false) { + @Override + public TestExternalTable get(String key) { + nameCaptured.countDown(); + awaitLatch(releaseLookup); + return super.get(key); + } + + @Override + public TestExternalTable getAndRunIfCurrent(String key, + BiPredicate actionRequired, + BiConsumer currentValueAction) { + nameCaptured.countDown(); + awaitLatch(releaseLookup); + return super.getAndRunIfCurrent(key, actionRequired, currentValueAction); + } + }; + setTablesEntryForTest(db, objectEntry); + + Future lookup = queryExecutor.submit(() -> db.getTableNullable(oldTable.getId())); + Assertions.assertTrue(nameCaptured.await(3L, TimeUnit.SECONDS)); + db.invalidateTableLookup(oldTable.getName()); + releaseLookup.countDown(); + + TestExternalTable replacement = lookup.get(3L, TimeUnit.SECONDS); + Assertions.assertNotNull(replacement); + Assertions.assertNotSame(oldTable, replacement); + Assertions.assertSame(replacement, objectEntry.getIfPresent(oldTable.getName())); + Assertions.assertEquals(oldTable.getName(), db.getCachedTableNameByIdForTest(oldTable.getId())); + } finally { + releaseLookup.countDown(); + queryExecutor.shutdownNow(); + refreshExecutor.shutdownNow(); + } + } + @Test public void testHotNamedLookupSkipsLockWhenIdAlreadyMapped() throws Exception { ExecutorService refreshExecutor = Executors.newSingleThreadExecutor(); diff --git a/fe/fe-core/src/test/java/org/apache/doris/datasource/plugin/PluginDrivenMissingTableTest.java b/fe/fe-core/src/test/java/org/apache/doris/datasource/plugin/PluginDrivenMissingTableTest.java index 9e0e9f5ae0bb39..970450798e4205 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/datasource/plugin/PluginDrivenMissingTableTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/datasource/plugin/PluginDrivenMissingTableTest.java @@ -71,6 +71,7 @@ public class PluginDrivenMissingTableTest extends TestWithFeService { private static final AtomicLong IDS = new AtomicLong(910000); private final Map remoteTables = new ConcurrentHashMap<>(); + private final Set remoteViews = ConcurrentHashMap.newKeySet(); private Connector connector; private ConnectorMetadata metadata; private TestCatalog catalog; @@ -119,6 +120,9 @@ void setup() { throw new IllegalStateException("metastore timeout"); } String name = call.getArgument(2); + if (remoteViews.contains(name)) { + return Optional.empty(); + } String column = remoteTables.get(name); if (column == null) { if (blockMissing.compareAndSet(true, false)) { @@ -129,6 +133,8 @@ void setup() { } return Optional.of(new Handle(name, column)); }); + Mockito.when(metadata.viewExists(Mockito.any(), Mockito.anyString(), Mockito.anyString())) + .thenAnswer(call -> remoteViews.contains(call.getArgument(2))); Mockito.when(metadata.getTableSchema(Mockito.any(), Mockito.any())).thenAnswer(call -> { Handle handle = call.getArgument(1); return new ConnectorTableSchema(handle.name, Collections.singletonList(new ConnectorColumn( @@ -160,6 +166,7 @@ void cleanup() throws Exception { Map> byName = Deencapsulation.getField(manager, "nameToCatalog"); byName.remove(catalog.getName()); remoteTables.clear(); + remoteViews.clear(); remoteCache.clear(); } @@ -217,6 +224,30 @@ void missingHandleRepairsNamesBeforeSchemaLoading() { Assertions.assertNull(remoteCache.get("a")); } + @Test + void existingViewWithoutTableHandleKeepsLookupCaches() { + remoteViews.add("v"); + PluginDrivenExternalTable view = database.getTableNullable("v"); + Assertions.assertNotNull(view); + remoteCache.put("v", "view metadata"); + ConnectorSession session = catalog.buildCrossStatementSession(); + try { + Assertions.assertFalse(view.resolveConnectorTableHandle(session, metadata).isPresent()); + Assertions.assertFalse(view.resolveConnectorTableHandle(session, metadata).isPresent()); + } finally { + session.getStatementScope().closeAll(); + } + Assertions.assertSame(view, database.getTableNullable("v")); + Assertions.assertSame(view, database.getTableNullable(view.getId())); + Assertions.assertTrue(database.getTableNamesWithLock().contains("v")); + Assertions.assertEquals("view metadata", remoteCache.get("v")); + + remoteViews.remove("v"); + readMissingHandle(view); + Assertions.assertNull(database.getTableNullable(view.getId())); + Assertions.assertFalse(database.getTableNamesWithLock().contains("v")); + } + @Test void delayedMissingDoesNotInvalidateRecreatedTable() throws Exception { verifyDelayedMissing(false); @@ -521,7 +552,9 @@ protected List listDatabaseNames() { @Override protected List listTableNamesFromRemote(SessionContext context, String dbName) { nameLoads.incrementAndGet(); - return new ArrayList<>(remoteTables.keySet()); + List names = new ArrayList<>(remoteTables.keySet()); + names.addAll(remoteViews); + return names; } @Override From 738c1a757932f0a4295078b106858ef0d2ed41b7 Mon Sep 17 00:00:00 2001 From: 924060929 Date: Fri, 23 Oct 2026 18:26:06 +0800 Subject: [PATCH 7/8] [fix](fe) Provide database fixture for external table tests ### What problem does this PR solve? Issue Number: None Problem Summary: Connector table tests used real table methods on mocks without an ExternalDatabase. The table lookup observer now requires that database, so 15 FE unit tests failed with a null pointer. Give the affected test fixtures a database mock that returns no cache observation, matching the supported lookup path while keeping the test focus on connector behavior. ### Release note None ### Check List (For Author) - Test: Unit Test (PluginDrivenExternalTableTest, 47 tests passed) - Behavior changed: No - Does this need documentation: No --- .../plugin/PluginDrivenExternalTableTest.java | 22 +++++++++++++------ 1 file changed, 15 insertions(+), 7 deletions(-) diff --git a/fe/fe-core/src/test/java/org/apache/doris/datasource/plugin/PluginDrivenExternalTableTest.java b/fe/fe-core/src/test/java/org/apache/doris/datasource/plugin/PluginDrivenExternalTableTest.java index f21ed9f8006890..64deaeaab129f3 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/datasource/plugin/PluginDrivenExternalTableTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/datasource/plugin/PluginDrivenExternalTableTest.java @@ -101,6 +101,16 @@ private static ConnectorSession noneScopedSession() { return s; } + private static PluginDrivenExternalTable tableWithCachelessDatabase() { + PluginDrivenExternalTable table = Mockito.mock(PluginDrivenExternalTable.class, Mockito.CALLS_REAL_METHODS); + ExternalDatabase database = Mockito.mock(ExternalDatabase.class); + Mockito.when(database.getRemoteName()).thenReturn(""); + Mockito.when(database.observeTableLookup(Mockito.eq(table), Mockito.any())) + .thenReturn(Optional.empty()); + Deencapsulation.setField(table, "db", database); + return table; + } + // ==================== ยง4.4 W3: per-handle write-admission capability probes ==================== /** @@ -134,7 +144,7 @@ private static PluginDrivenExternalTable capabilityTable(boolean handlePresent, ConnectorSession session = noneScopedSession(); Mockito.when(catalog.buildConnectorSession()).thenReturn(session); Mockito.when(catalog.buildCrossStatementSession()).thenReturn(session); - PluginDrivenExternalTable table = Mockito.mock(PluginDrivenExternalTable.class, Mockito.CALLS_REAL_METHODS); + PluginDrivenExternalTable table = tableWithCachelessDatabase(); Deencapsulation.setField(table, "catalog", catalog); return table; } @@ -207,7 +217,7 @@ private static PluginDrivenExternalTable writeTargetTable(ConnectorTableHandle r ConnectorSession session = noneScopedSession(); Mockito.when(catalog.buildConnectorSession()).thenReturn(session); Mockito.when(catalog.buildCrossStatementSession()).thenReturn(session); - PluginDrivenExternalTable table = Mockito.mock(PluginDrivenExternalTable.class, Mockito.CALLS_REAL_METHODS); + PluginDrivenExternalTable table = tableWithCachelessDatabase(); Deencapsulation.setField(table, "catalog", catalog); return table; } @@ -249,7 +259,7 @@ private static PluginDrivenExternalTable syntheticPredicateTable(ConnectorTableH ConnectorSession session = noneScopedSession(); Mockito.when(catalog.buildConnectorSession()).thenReturn(session); Mockito.when(catalog.buildCrossStatementSession()).thenReturn(session); - PluginDrivenExternalTable table = Mockito.mock(PluginDrivenExternalTable.class, Mockito.CALLS_REAL_METHODS); + PluginDrivenExternalTable table = tableWithCachelessDatabase(); Deencapsulation.setField(table, "catalog", catalog); return table; } @@ -317,8 +327,7 @@ private static PluginDrivenExternalTable pluginTable(List synth Mockito.when(catalog.buildConnectorSession()).thenReturn(session); Mockito.when(catalog.buildCrossStatementSession()).thenReturn(session); - PluginDrivenExternalTable table = - Mockito.mock(PluginDrivenExternalTable.class, Mockito.CALLS_REAL_METHODS); + PluginDrivenExternalTable table = tableWithCachelessDatabase(); Deencapsulation.setField(table, "catalog", catalog); SchemaCacheValue scv = Mockito.mock(SchemaCacheValue.class); Mockito.when(scv.getSchema()).thenReturn(BASE_SCHEMA); @@ -450,8 +459,7 @@ public void resolveWriteColumnsRunsAndRestoresPluginContextClassLoader() { PluginDrivenExternalCatalog catalog = Mockito.mock(PluginDrivenExternalCatalog.class); Mockito.when(catalog.getConnector()).thenReturn(connector); Mockito.when(catalog.buildConnectorSession()).thenReturn(session); - PluginDrivenExternalTable table = Mockito.mock( - PluginDrivenExternalTable.class, Mockito.CALLS_REAL_METHODS); + PluginDrivenExternalTable table = tableWithCachelessDatabase(); Deencapsulation.setField(table, "catalog", catalog); Mockito.doReturn(99L).when(table).getId(); ConnectContext ctx = new ConnectContext(); From bee05669e8b933c5efb4c3cb2464e6b0db01ec76 Mon Sep 17 00:00:00 2001 From: 924060929 Date: Fri, 23 Oct 2026 18:52:51 +0800 Subject: [PATCH 8/8] [fix](fe) Fence stale table misses across metadata refreshes ### What problem does this PR solve? Issue Number: None Problem Summary: A missing-table response started before an ordinary metadata refresh could still evict the refreshed schema, table object, ID, and name snapshot because the refresh intentionally retained the same object registration. Advance a transient per-table metadata generation before refresh invalidation and compare it during guarded missing-table cleanup under the same catalog monitor. A delayed miss now leaves the current refreshed metadata intact. Also remove two unused Paimon test imports that failed Checkstyle. ### Release note None ### Check List (For Author) - Test: Unit Test (97 tests across PluginDrivenMissingTableTest, PluginDrivenExternalTableTest, RefreshManagerTest, and PaimonConnectorMetadataReadAuthTest; Maven Checkstyle) - Behavior changed: Yes (stale missing-table responses no longer evict metadata refreshed after their lookup began) - Does this need documentation: No --- .../PaimonConnectorMetadataReadAuthTest.java | 2 -- .../apache/doris/catalog/RefreshManager.java | 1 + .../doris/datasource/ExternalDatabase.java | 2 ++ .../doris/datasource/ExternalTable.java | 13 ++++++++++ .../plugin/PluginDrivenMissingTableTest.java | 25 +++++++++++++++++++ 5 files changed, 41 insertions(+), 2 deletions(-) diff --git a/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonConnectorMetadataReadAuthTest.java b/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonConnectorMetadataReadAuthTest.java index a6c9571a56f60f..ec2174d7029246 100644 --- a/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonConnectorMetadataReadAuthTest.java +++ b/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonConnectorMetadataReadAuthTest.java @@ -18,7 +18,6 @@ package org.apache.doris.connector.paimon; import org.apache.doris.connector.spi.DorisConnectorException; -import org.apache.doris.connector.spi.handle.ConnectorTableHandle; import org.apache.paimon.partition.Partition; import org.apache.paimon.types.DataTypes; @@ -31,7 +30,6 @@ import java.util.LinkedHashMap; import java.util.List; import java.util.Map; -import java.util.Optional; /** * M-11 (rereview2 #6) read-path Kerberos doAs tests: every remote metadata READ RPC must run INSIDE 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 4e81dcabdaff4a..9c1eaaf899dacc 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 @@ -322,6 +322,7 @@ public void refreshTableInternal(ExternalDatabase db, ExternalTable table, long private void refreshTableInternal(ExternalDatabase db, ExternalTable table, long updateTime, boolean invalidateTableLookup) { boolean invalidateDatabase = invalidateTableLookup && table.getCatalog().getLowerCaseTableNames() != 0; + table.advanceMetadataRefreshGeneration(); table.unsetObjectCreated(); if (updateTime > 0) { table.setUpdateTime(updateTime); 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 c874f46f07b7e3..8cf5501a960f3e 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 @@ -362,6 +362,7 @@ public Optional observeTableLookup(ExternalTable table, Runnable inval if (names == null || !table.getRemoteName().equals(names.remoteNameOfLocalName(table.getName()))) { return Optional.empty(); } + long metadataRefreshGeneration = table.getMetadataRefreshGeneration(); return extCatalog.observeDatabase(this).flatMap(database -> tableNames.observe("", names).flatMap(membership -> tables.observe(table.getName(), table).map(object -> () -> MetaCacheCleanup.runDeferred(() -> { @@ -369,6 +370,7 @@ public Optional observeTableLookup(ExternalTable table, Runnable inval // runs here; remote reads happened before entering this guarded commit. synchronized (extCatalog) { if (!initialized || !extCatalog.isInitialized() + || table.getMetadataRefreshGeneration() != metadataRefreshGeneration || Env.getCurrentEnv().getCatalogMgr().getCatalog(extCatalog.getId()) != extCatalog) { return; diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/ExternalTable.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/ExternalTable.java index 4c874fe372179d..3e81cb70aec630 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/datasource/ExternalTable.java +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/ExternalTable.java @@ -92,6 +92,8 @@ public class ExternalTable implements TableIf, Writable, GsonPostProcessable { // record the table update time, like insert/alter/delete protected volatile long updateTime = 0; + // Fences missing-table responses that started before a metadata-only refresh of this object. + private transient volatile long metadataRefreshGeneration; protected long dbId; protected boolean objectCreated; @@ -338,6 +340,17 @@ public void setUpdateTime(long updateTime) { this.updateTime = updateTime; } + public long getMetadataRefreshGeneration() { + return metadataRefreshGeneration; + } + + /** Ordered with guarded missing-table cleanup, which holds the same catalog monitor. */ + public void advanceMetadataRefreshGeneration() { + synchronized (catalog) { + metadataRefreshGeneration++; + } + } + @Override public long getLastCheckTime() { return 0; diff --git a/fe/fe-core/src/test/java/org/apache/doris/datasource/plugin/PluginDrivenMissingTableTest.java b/fe/fe-core/src/test/java/org/apache/doris/datasource/plugin/PluginDrivenMissingTableTest.java index 970450798e4205..d62b8f21101058 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/datasource/plugin/PluginDrivenMissingTableTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/datasource/plugin/PluginDrivenMissingTableTest.java @@ -258,6 +258,31 @@ void delayedMissingDoesNotInvalidateExactKeyReplacement() throws Exception { verifyDelayedMissing(true); } + @Test + void delayedMissingDoesNotInvalidateMetadataOnlyRefresh() throws Exception { + PluginDrivenExternalTable table = warmTarget(); + remoteTables.remove("a"); + blockMissing.set(true); + Future missing = worker.submit(() -> readMissingHandle(table)); + Assertions.assertTrue(missingStarted.await(10, TimeUnit.SECONDS)); + + remoteTables.put("a", "new_column"); + new RefreshManager().refreshTableInternal(database, table, 123L); + List refreshedSchema = table.getFullSchema(); + remoteCache.put("a", "refreshed"); + int nameLoads = catalog.nameLoads.get(); + + releaseMissing.countDown(); + missing.get(10, TimeUnit.SECONDS); + Assertions.assertSame(table, database.getTableNullable("a")); + Assertions.assertSame(table, database.getTableNullable(table.getId())); + Assertions.assertSame(refreshedSchema, table.getFullSchema()); + Assertions.assertEquals("new_column", refreshedSchema.get(0).getName()); + Assertions.assertEquals("refreshed", remoteCache.get("a")); + Assertions.assertEquals(Set.of("a", "b"), database.getTableNamesWithLock()); + Assertions.assertEquals(nameLoads, catalog.nameLoads.get()); + } + private void verifyDelayedMissing(boolean replaceWithoutScopeInvalidation) throws Exception { PluginDrivenExternalTable old = warmTarget(); remoteTables.remove("a");