924060929 commented on code in PR #66913:
URL: https://github.com/apache/doris/pull/66913#discussion_r3828227585
##########
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:
Fixed in f090dccb71b. The exact planning executor is now stored with
IcebergTableCacheValue, exposed through its statement lease, and captured by
IcebergScanNode together with the frozen table generation. Catalog close defers
both Catalog.close() and executor shutdown until the generation tracker drains.
Covered by IcebergTableCacheValueTest, IcebergCatalogResourceTrackerTest, and
IcebergScanNodeTest.
##########
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:
Fixed in f090dccb71b. Every Iceberg reset now invalidates this catalog
engine cache before rotating the runtime. Existing borrowers retain the retired
generation through the tracker, while new statements must load from the new
runtime configuration.
##########
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:
Fixed in f090dccb71b. proxyExecute now builds the forwarded result inside a
try/finally that closes the current StatementContext on ordinary,
prepared-forwarded, and error paths. COM_STMT_EXECUTE also closes the prepared
StatementContext after execution. Statement lifecycle and retry coverage passed
locally.
##########
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:
Fixed in f090dccb71b. AutoCloseConnectContext now closes the current
StatementContext before clear/remove/restore, with nested finally blocks
preserving cleanup and previous-context restoration even when release fails.
AutoCloseConnectContextTest verifies statement-resource release and restoration.
--
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]
