github-actions[bot] commented on code in PR #66913:
URL: https://github.com/apache/doris/pull/66913#discussion_r3822015436
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergExternalCatalog.java:
##########
@@ -173,17 +182,58 @@ protected List<String>
listTableNamesFromRemote(SessionContext ctx, String dbNam
}
@Override
- public void onClose() {
+ public synchronized void onClose() {
super.onClose();
- if (null != catalog) {
- try {
- if (catalog instanceof AutoCloseable) {
- ((AutoCloseable) catalog).close();
- }
- catalog = null;
- } catch (Exception e) {
- LOG.warn("Failed to close iceberg catalog: {}", getName(), e);
+ Catalog retiredCatalog = catalog;
+ catalog = null;
+ if (retiredCatalog != null) {
+ resourceTracker.retireCurrent(() -> closeCatalog(retiredCatalog));
Review Comment:
[P1] Invalidate table generations when rotating the catalog
A URI, warehouse, catalog-type, or credential update calls
`resetToUninitialized(false)`, so this retires the current catalog but leaves
the Iceberg table cache intact (the override only removes it for cache-control
keys). Subsequent statements can therefore keep borrowing old `Table`/FileIO
generations and continue using the previous runtime configuration until refresh
or expiry. Invalidate this catalog's Iceberg cache group whenever the runtime
generation rotates; the tracker can still defer cleanup for active old
borrowers.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergExternalMetaCache.java:
##########
@@ -159,23 +189,131 @@ public void invalidateCatalogEntries(long catalogId) {
}
private IcebergTableCacheValue loadTableCacheValue(NameMapping
nameMapping) {
+ CatalogIf catalog =
Env.getCurrentEnv().getCatalogMgr().getCatalog(nameMapping.getCtlId());
+ if (catalog instanceof IcebergExternalCatalog) {
+ IcebergExternalCatalog icebergCatalog = (IcebergExternalCatalog)
catalog;
+ try (IcebergExternalCatalog.TableLoadContext loadContext =
icebergCatalog.beginTableLoad()) {
+ IcebergMetadataOps ops = loadContext.getOps();
+ Table table;
+ try {
+ table =
loadContext.loadTable(nameMapping.getRemoteDbName(),
nameMapping.getRemoteTblName());
+ } catch (Exception e) {
+ throw new
RuntimeException(ExceptionUtils.getRootCauseMessage(e), e);
+ }
+ ExternalTable dorisTable = findExternalTable(nameMapping,
ENGINE);
+ Runnable tableCleanup =
tableCleanup(loadContext.getCatalogType(), ops, table);
+ IcebergCatalogResourceTracker.ResourceLease catalogLease =
loadContext.promote();
+ return new IcebergTableCacheValue(table, () ->
loadSnapshotProjection(dorisTable, table), () -> {
+ try {
+ tableCleanup.run();
+ } finally {
+ catalogLease.close();
+ }
+ });
+ }
+ }
+ Table table = loadTable(nameMapping);
+ IcebergMetadataOps ops = resolveMetadataOps(catalog);
+ ExternalTable dorisTable = findExternalTable(nameMapping, ENGINE);
+ return new IcebergTableCacheValue(table, () ->
loadSnapshotProjection(dorisTable, table),
+ tableCleanup(catalog, ops, table));
+ }
+
+ private Table loadTable(NameMapping nameMapping) {
CatalogIf catalog =
Env.getCurrentEnv().getCatalogMgr().getCatalog(nameMapping.getCtlId());
if (catalog == null) {
throw new RuntimeException(String.format("Cannot find catalog %d
when loading table %s/%s.",
nameMapping.getCtlId(), nameMapping.getLocalDbName(),
nameMapping.getLocalTblName()));
}
- IcebergMetadataOps ops = resolveMetadataOps(catalog);
+ return loadTable(nameMapping, catalog, resolveMetadataOps(catalog));
+ }
+
+ private Table loadTable(NameMapping nameMapping, CatalogIf catalog,
IcebergMetadataOps ops) {
try {
- Table table = ((ExternalCatalog)
catalog).getExecutionAuthenticator()
+ return ((ExternalCatalog) catalog).getExecutionAuthenticator()
.execute(() ->
ops.loadTable(nameMapping.getRemoteDbName(), nameMapping.getRemoteTblName()));
- ExternalTable dorisTable = findExternalTable(nameMapping, ENGINE);
- return new IcebergTableCacheValue(table, () ->
loadSnapshotProjection(dorisTable, table));
} catch (Exception e) {
throw new RuntimeException(ExceptionUtils.getRootCauseMessage(e),
e);
}
}
+ private IcebergTableCacheValue.Lease statementLease(NameMapping
nameMapping) {
+ ConnectContext connectContext = ConnectContext.get();
+ StatementContext statementContext = connectContext == null ? null :
connectContext.getStatementContext();
+ if (statementContext == null) {
+ return null;
+ }
+ String resourceKey = "iceberg-table:" + nameMapping.getCtlId() +
"\u0000"
+ + nameMapping.getRemoteDbName() + "\u0000" +
nameMapping.getRemoteTblName();
+ return statementContext.getOrRegisterStatementResource(resourceKey, ()
-> borrow(nameMapping));
Review Comment:
[P1] Close leases after forwarded master execution
This lease is registered in the `StatementContext` created by the proxy-side
`StmtExecutor`, but `ConnectProcessor.proxyExecute()` returns without ever
closing that context; `FrontendServiceImpl.forward()` only removes the
thread-local. A forwarded Iceberg query therefore leaves a permanent borrower,
so later eviction cannot close its table FileIO or retired catalog generation.
Close the proxy statement context in a `finally` covering ordinary,
prepared-forwarded, and error paths, and add a forwarded-query lifecycle test.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergExternalCatalog.java:
##########
@@ -173,17 +182,58 @@ protected List<String>
listTableNamesFromRemote(SessionContext ctx, String dbNam
}
@Override
- public void onClose() {
+ public synchronized void onClose() {
Review Comment:
[P1] Keep the planning executor with the leased generation
The tracker only defers `Catalog.close()`, but `super.onClose()` has already
shut down this generation's pre-auth executor. An active scan can acquire a
leased old table and call
`planWith(source.getCatalog().getThreadPoolWithPreAuth())` only after a
concurrent reset; it then gets either the shut-down pool or a pool from the new
runtime generation. Keep the executor and other operations resources used after
table acquisition inside the tracked generation, and test reset between table
acquisition and `planFiles()`/commit.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/StatementContext.java:
##########
@@ -918,7 +1025,11 @@ protected void finalize() throws Throwable {
@Override
public void close() {
clearExternalScanTasks();
- releasePlannerResources();
+ try {
+ releaseStatementResources();
Review Comment:
[P1] Close resources in auto-close contexts
`AutoCloseConnectContext.close()` only reaches `ConnectContext.clear()`,
which nulls the StatementContext without invoking this release. External
statistics queries in `BaseAnalysisTask` use that owner, so an Iceberg analysis
discards its lease's only close handle; refresh or invalidation can then never
drain the old table/FileIO/catalog generation, and repeated analyze runs
accumulate borrowers. Close the current StatementContext before
clearing/restoring the context on every success and failure path, and add an
external-analyze lifecycle test that does not close it manually.
--
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]
