github-actions[bot] commented on code in PR #69458:
URL: https://github.com/apache/doris/pull/69458#discussion_r4355558692
##########
fe/fe-connector/fe-connector-fluss/src/main/java/org/apache/doris/connector/fluss/FlussConnector.java:
##########
@@ -229,53 +232,50 @@ synchronized void rememberLakePath(TablePath sourcePath,
TablePath lakePath) {
}
@Override
- public synchronized void invalidateTable(String dbName, String tableName) {
- if (lakeSibling != null) {
- // Keep the mapping until the table is resolved again (which
replaces it) or the whole catalog
- // is invalidated. A single engine refresh may emit more than one
targeted invalidation, and a
- // second one must not fall back to the source name merely because
the first already ran.
- TablePath lakePath = lakeTablePaths.get(TablePath.of(dbName,
tableName));
- lakeSibling.invalidateTable(
- lakePath == null ? dbName : lakePath.getDatabaseName(),
- lakePath == null ? tableName : lakePath.getTableName());
+ public synchronized List<ScopePath> cacheDependencies(ScopePath scope) {
+ List<ScopePath> dependencies = new java.util.ArrayList<>();
+ if (scope.database() == null) {
+ return dependencies;
}
- }
-
- @Override
- public synchronized void invalidateDb(String dbName) {
- if (lakeSibling != null) {
- Set<String> lakeDatabases = new LinkedHashSet<>();
- // An unresolved table cannot have populated the sibling's table
cache, but the sibling may
- // still hold database-level metadata for the source namespace.
Invalidating it as well as every
- // known physical override is conservative and prevents one
overridden table from accidentally
- // suppressing invalidation of ordinary lake tables in the same
Fluss database.
- lakeDatabases.add(dbName);
+ if (scope.table() != null) {
+ TablePath lake = lakeTablePaths.get(TablePath.of(scope.database(),
scope.table()));
+ if (lake != null) {
+ dependencies.add(ScopePath.table(lake.getDatabaseName(),
lake.getTableName()));
+ }
+ } else {
lakeTablePaths.forEach((source, lake) -> {
- if (source.getDatabaseName().equals(dbName)) {
- lakeDatabases.add(lake.getDatabaseName());
+ if (source.getDatabaseName().equals(scope.database())) {
+ dependencies.add(scope.level() == ScopePath.Level.TABLE
+ ? ScopePath.tableCollection(lake.getDatabaseName())
+ : ScopePath.database(lake.getDatabaseName()));
}
});
- lakeDatabases.forEach(lakeSibling::invalidateDb);
}
+ return dependencies;
}
@Override
- public synchronized void invalidateAll() {
- lakeTablePaths.clear();
- if (lakeSibling != null) {
- lakeSibling.invalidateAll();
- }
+ public void invalidateTable(String database, String table) {
+ List<ScopePath> scopes = new
java.util.ArrayList<>(cacheDependencies(ScopePath.table(database, table)));
+ scopes.add(ScopePath.table(database, table));
+ metaCache.invalidateScopes(scopes);
}
@Override
- public synchronized void invalidatePartition(
- String dbName, String tableName, List<String> partitionNames) {
- if (lakeSibling != null) {
- TablePath lakePath = lakeTablePaths.get(TablePath.of(dbName,
tableName));
- lakeSibling.invalidatePartition(
- lakePath == null ? dbName : lakePath.getDatabaseName(),
- lakePath == null ? tableName : lakePath.getTableName(),
partitionNames);
- }
+ public void invalidateDb(String database) {
+ List<ScopePath> scopes = new
java.util.ArrayList<>(cacheDependencies(ScopePath.database(database)));
+ scopes.add(ScopePath.database(database));
+ metaCache.invalidateScopes(scopes);
+ }
+
+ @Override
+ public void invalidateAll() {
+ metaCache.invalidateCatalog();
Review Comment:
[P2] Retire Fluss lake-path history on catalog refresh
After querying lake-enabled Fluss tables, dropping them externally and
issuing `REFRESH CATALOG`, every source/lake `TablePath` pair now remains in
`lakeTablePaths`. Previously catalog refresh called
`FlussConnector.invalidateAll()`, which cleared this map; the new engine path
invalidates the shared root directly, and the map is only cleared by `close()`.
A long-lived catalog with table churn therefore retains all historical names
indefinitely, and each database dependency projection scans that history under
the connector monitor. Give this auxiliary mapping a lifetime coordinated with
root catalog invalidation, preserving dependencies until sibling scopes are
fenced. Restoring a clear only in this method is insufficient because engine
refresh bypasses it; cover repeated catalog refreshes in a regression test.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
