Skip to content

Commit c87876e

Browse files
924060929924060929
committed
[fix](paimon) Invalidate SDK cache on metadata refresh
### What problem does this PR solve? Issue Number: None Problem Summary: After a Paimon table is changed or dropped outside Doris, stale metadata before an explicit refresh is expected. However, table, database, and catalog refreshes only invalidated Doris-owned metadata entries while Paimon SDK CachingCatalog kept the old Table handle, so CREATE and query paths could still observe stale metadata after refresh. Make Paimon metadata invalidation evict every matching SDK table cache entry before clearing the corresponding Doris entries. ### Release note Fix explicit metadata refresh continuing to expose stale Paimon table metadata after an external table change. ### Check List (For Author) - Test: Unit Test - PaimonExternalMetaCacheTest - PaimonMetadataOpsTest - Behavior changed: Yes. Table, database, and catalog refreshes now invalidate both Doris and Paimon SDK table caches. - Does this need documentation: No
1 parent d08ff0f commit c87876e

4 files changed

Lines changed: 158 additions & 7 deletions

File tree

‎fe/fe-core/src/main/java/org/apache/doris/datasource/metacache/paimon/PaimonTableLoader.java‎

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -32,6 +32,16 @@
3232
*/
3333
public final class PaimonTableLoader {
3434

35+
public void invalidate(NameMapping nameMapping) {
36+
try {
37+
catalog(nameMapping).invalidatePaimonTable(nameMapping);
38+
} catch (Exception e) {
39+
throw new CacheException("failed to invalidate paimon table %s.%s.%s: %s",
40+
e, nameMapping.getCtlId(), nameMapping.getLocalDbName(), nameMapping.getLocalTblName(),
41+
e.getMessage());
42+
}
43+
}
44+
3545
public Table load(NameMapping nameMapping) {
3646
try {
3747
// Doris caches Paimon table handles to implement the external-catalog TTL and REFRESH

‎fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/PaimonExternalCatalog.java‎

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -130,6 +130,15 @@ public Table getPaimonTable(NameMapping nameMapping, String branch, String query
130130
return loadPaimonTable(nameMapping, queryType, identifier);
131131
}
132132

133+
public void invalidatePaimonTable(NameMapping nameMapping) throws Exception {
134+
makeSureInitialized();
135+
Identifier identifier = tableIdentifier(nameMapping, null, null);
136+
executionAuthenticator.execute(() -> {
137+
catalog.invalidateTable(identifier);
138+
return null;
139+
});
140+
}
141+
133142
/**
134143
* Invalidates Paimon's catalog-level table cache and reloads the table.
135144
*

‎fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/PaimonExternalMetaCache.java‎

Lines changed: 42 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -33,11 +33,14 @@
3333

3434
import org.apache.paimon.table.Table;
3535

36+
import java.util.ArrayList;
37+
import java.util.List;
3638
import java.util.Map;
3739
import java.util.concurrent.ConcurrentHashMap;
3840
import java.util.concurrent.ExecutorService;
3941
import java.util.concurrent.atomic.AtomicBoolean;
4042
import java.util.concurrent.atomic.AtomicLong;
43+
import java.util.function.Predicate;
4144
import javax.annotation.Nullable;
4245

4346
/**
@@ -118,6 +121,38 @@ public Table getPaimonTable(NameMapping nameMapping) {
118121
return tableEntry.get(nameMapping.getCtlId()).get(nameMapping).getPaimonTable();
119122
}
120123

124+
@Override
125+
public void invalidateTable(long catalogId, String dbName, String tableName) {
126+
try {
127+
invalidatePaimonTables(catalogId, nameMapping -> matchTable(nameMapping, dbName, tableName));
128+
} finally {
129+
super.invalidateTable(catalogId, dbName, tableName);
130+
}
131+
}
132+
133+
@Override
134+
public void invalidateDb(long catalogId, String dbName) {
135+
try {
136+
invalidatePaimonTables(catalogId, nameMapping -> matchDb(nameMapping, dbName));
137+
} finally {
138+
super.invalidateDb(catalogId, dbName);
139+
}
140+
}
141+
142+
private void invalidatePaimonTables(long catalogId, Predicate<NameMapping> predicate) {
143+
MetaCacheEntry<NameMapping, PaimonTableCacheValue> tables = tableEntry.getIfInitialized(catalogId);
144+
if (tables == null) {
145+
return;
146+
}
147+
List<NameMapping> nameMappings = new ArrayList<>();
148+
tables.forEach((nameMapping, ignored) -> {
149+
if (predicate.test(nameMapping)) {
150+
nameMappings.add(nameMapping);
151+
}
152+
});
153+
nameMappings.forEach(tableLoader::invalidate);
154+
}
155+
121156
public PaimonSnapshotCacheValue getSnapshotCache(ExternalTable dorisTable) {
122157
NameMapping nameMapping = dorisTable.getOrBuildNameMapping();
123158
MetaCacheEntry<NameMapping, PaimonTableCacheValue> tables = tableEntry.get(nameMapping.getCtlId());
@@ -555,9 +590,13 @@ public void onCatalogPermanentlyRemoved(long catalogId) {
555590

556591
@Override
557592
public void invalidateCatalogEntries(long catalogId) {
558-
latestObservedFences.keySet().removeIf(owner -> owner.nameMapping.getCtlId() == catalogId);
559-
fenceCaptureLocks.keySet().removeIf(owner -> owner.nameMapping.getCtlId() == catalogId);
560-
super.invalidateCatalogEntries(catalogId);
593+
try {
594+
invalidatePaimonTables(catalogId, ignored -> true);
595+
} finally {
596+
latestObservedFences.keySet().removeIf(owner -> owner.nameMapping.getCtlId() == catalogId);
597+
fenceCaptureLocks.keySet().removeIf(owner -> owner.nameMapping.getCtlId() == catalogId);
598+
super.invalidateCatalogEntries(catalogId);
599+
}
561600
}
562601

563602
@Override

‎fe/fe-core/src/test/java/org/apache/doris/datasource/paimon/PaimonExternalMetaCacheTest.java‎

Lines changed: 97 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -2151,6 +2151,77 @@ public void testDorisTableCacheMissReloadsPaimonCatalogOnce() throws Exception {
21512151
}
21522152
}
21532153

2154+
@Test
2155+
public void testRefreshTableInvalidatesPaimonCatalogCacheAfterExternalDrop() throws Exception {
2156+
java.io.File warehouse = temporaryFolder.newFolder("refresh_after_external_drop");
2157+
Map<String, String> properties = new HashMap<>();
2158+
properties.put("type", "paimon");
2159+
properties.put(PaimonExternalCatalog.PAIMON_CATALOG_TYPE,
2160+
PaimonExternalCatalog.PAIMON_FILESYSTEM);
2161+
properties.put("warehouse", warehouse.toURI().toString());
2162+
long catalogId = 94L;
2163+
PaimonExternalCatalog dorisCatalog = new PaimonExternalCatalog(
2164+
catalogId, "paimon_refresh_test", null, properties, "");
2165+
dorisCatalog.makeSureInitialized();
2166+
dorisCatalog.catalog.createDatabase("db", false);
2167+
Identifier identifier = Identifier.create("db", "table");
2168+
Schema schema = Schema.newBuilder()
2169+
.column("id", DataTypes.INT())
2170+
.build();
2171+
dorisCatalog.catalog.createTable(identifier, schema, false);
2172+
2173+
Map<String, String> externalProperties = new HashMap<>(properties);
2174+
externalProperties.put("paimon.cache-enabled", "false");
2175+
PaimonExternalCatalog externalCatalog = new PaimonExternalCatalog(
2176+
95L, "external_paimon", null, externalProperties, "");
2177+
externalCatalog.makeSureInitialized();
2178+
2179+
CatalogMgr catalogMgr = Mockito.mock(CatalogMgr.class);
2180+
Env env = Mockito.mock(Env.class);
2181+
Mockito.when(env.getCatalogMgr()).thenReturn(catalogMgr);
2182+
Mockito.doReturn(dorisCatalog).when(catalogMgr)
2183+
.getCatalogOrException(Mockito.eq(catalogId), Mockito.any());
2184+
ExecutorService executor = Executors.newSingleThreadExecutor();
2185+
try (MockedStatic<Env> mockedEnv = Mockito.mockStatic(Env.class)) {
2186+
mockedEnv.when(Env::getCurrentEnv).thenReturn(env);
2187+
PaimonExternalMetaCache cache = new PaimonExternalMetaCache(executor);
2188+
cache.initCatalog(catalogId, Collections.emptyMap());
2189+
NameMapping nameMapping = new NameMapping(catalogId, "db", "table", "db", "table");
2190+
Table staleTable = cache.getPaimonTable(nameMapping);
2191+
2192+
externalCatalog.catalog.dropTable(identifier, false);
2193+
Assert.assertSame(staleTable, cache.getPaimonTable(nameMapping));
2194+
2195+
cache.invalidateTable(catalogId, "db", "table");
2196+
assertPaimonTableMissing(cache, nameMapping);
2197+
2198+
externalCatalog.catalog.createTable(identifier, schema, false);
2199+
cache.getPaimonTable(nameMapping);
2200+
externalCatalog.catalog.dropTable(identifier, false);
2201+
cache.invalidateDb(catalogId, "db");
2202+
assertPaimonTableMissing(cache, nameMapping);
2203+
2204+
externalCatalog.catalog.createTable(identifier, schema, false);
2205+
cache.getPaimonTable(nameMapping);
2206+
externalCatalog.catalog.dropTable(identifier, false);
2207+
cache.invalidateCatalogEntries(catalogId);
2208+
assertPaimonTableMissing(cache, nameMapping);
2209+
} finally {
2210+
executor.shutdownNow();
2211+
dorisCatalog.catalog.close();
2212+
externalCatalog.catalog.close();
2213+
}
2214+
}
2215+
2216+
private void assertPaimonTableMissing(PaimonExternalMetaCache cache, NameMapping nameMapping) {
2217+
try {
2218+
cache.getPaimonTable(nameMapping);
2219+
Assert.fail("the externally dropped table must not remain visible after refresh");
2220+
} catch (CacheException e) {
2221+
Assert.assertTrue(e.getMessage().contains("failed to load paimon table"));
2222+
}
2223+
}
2224+
21542225
@Test
21552226
public void testPartitionProjectionRejectsUnsafeEffectiveTableBeforeEnumeration() throws Exception {
21562227
FileStoreTable unsafeTable = newPartitionedTable(
@@ -2557,7 +2628,7 @@ private Object readField(Object target, Class<?> owner, String fieldName) throws
25572628
}
25582629

25592630
@Test
2560-
public void testInvalidateTablePrecise() {
2631+
public void testInvalidateTablePrecise() throws Exception {
25612632
ExecutorService executor = Executors.newSingleThreadExecutor();
25622633
try {
25632634
PaimonExternalMetaCache cache = new PaimonExternalMetaCache(executor);
@@ -2582,8 +2653,19 @@ public void testInvalidateTablePrecise() {
25822653
snapshotEntry.put(snapshotKey, new PaimonSnapshotCacheValue(
25832654
PaimonPartitionInfo.EMPTY, new PaimonSnapshot(1L, 2L, null)));
25842655

2585-
cache.invalidateTable(catalogId, "db1", "tbl1");
2656+
PaimonExternalCatalog dorisCatalog = Mockito.mock(PaimonExternalCatalog.class);
2657+
CatalogMgr catalogMgr = Mockito.mock(CatalogMgr.class);
2658+
Env env = Mockito.mock(Env.class);
2659+
Mockito.when(env.getCatalogMgr()).thenReturn(catalogMgr);
2660+
Mockito.doReturn(dorisCatalog).when(catalogMgr)
2661+
.getCatalogOrException(Mockito.eq(catalogId), Mockito.any());
2662+
try (MockedStatic<Env> mockedEnv = Mockito.mockStatic(Env.class)) {
2663+
mockedEnv.when(Env::getCurrentEnv).thenReturn(env);
2664+
cache.invalidateTable(catalogId, "db1", "tbl1");
2665+
}
25862666

2667+
Mockito.verify(dorisCatalog).invalidatePaimonTable(t1);
2668+
Mockito.verify(dorisCatalog, Mockito.never()).invalidatePaimonTable(t2);
25872669
Assert.assertNull(tableEntry.getIfPresent(t1));
25882670
Assert.assertNotNull(tableEntry.getIfPresent(t2));
25892671
Assert.assertNull(snapshotEntry.getIfPresent(snapshotKey));
@@ -2593,7 +2675,7 @@ public void testInvalidateTablePrecise() {
25932675
}
25942676

25952677
@Test
2596-
public void testInvalidateDbAndStats() {
2678+
public void testInvalidateDbAndStats() throws Exception {
25972679
ExecutorService executor = Executors.newSingleThreadExecutor();
25982680
try {
25992681
PaimonExternalMetaCache cache = new PaimonExternalMetaCache(executor);
@@ -2618,8 +2700,19 @@ public void testInvalidateDbAndStats() {
26182700
schemaEntry.put(db1Schema, new SchemaCacheValue(Collections.emptyList()));
26192701
schemaEntry.put(db2Schema, new SchemaCacheValue(Collections.emptyList()));
26202702

2621-
cache.invalidateDb(catalogId, "db1");
2703+
PaimonExternalCatalog dorisCatalog = Mockito.mock(PaimonExternalCatalog.class);
2704+
CatalogMgr catalogMgr = Mockito.mock(CatalogMgr.class);
2705+
Env env = Mockito.mock(Env.class);
2706+
Mockito.when(env.getCatalogMgr()).thenReturn(catalogMgr);
2707+
Mockito.doReturn(dorisCatalog).when(catalogMgr)
2708+
.getCatalogOrException(Mockito.eq(catalogId), Mockito.any());
2709+
try (MockedStatic<Env> mockedEnv = Mockito.mockStatic(Env.class)) {
2710+
mockedEnv.when(Env::getCurrentEnv).thenReturn(env);
2711+
cache.invalidateDb(catalogId, "db1");
2712+
}
26222713

2714+
Mockito.verify(dorisCatalog).invalidatePaimonTable(db1Table);
2715+
Mockito.verify(dorisCatalog, Mockito.never()).invalidatePaimonTable(db2Table);
26232716
Assert.assertNull(tableEntry.getIfPresent(db1Table));
26242717
Assert.assertNotNull(tableEntry.getIfPresent(db2Table));
26252718
Assert.assertNull(schemaEntry.getIfPresent(db1Schema));

0 commit comments

Comments
 (0)