924060929 commented on code in PR #66913:
URL: https://github.com/apache/doris/pull/66913#discussion_r3838735506
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/metacache/MetaCacheEntry.java:
##########
@@ -231,6 +257,7 @@ private V getWithManualLoad(K key, Function<K, V>
loadFunction) {
long generation = invalidateGeneration.get();
Review Comment:
已修复。catalog group 移除时会 seal 其中所有 entry；旧 entry 的后续 load 无法发布，并会退休刚加载的
Hudi/Iceberg value 后要求调用方重新解析当前 group。竞态测试已覆盖。
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/source/IcebergScanNode.java:
##########
@@ -267,6 +269,7 @@ protected void doInitialize() throws UserException {
getRelationSnapshot();
icebergTable = source.getIcebergTable();
icebergTable = useFrozenTableGeneration(icebergTable);
+ planningExecutor = getPlanningExecutor();
Review Comment:
已修复。冻结的 Iceberg snapshot 现在携带并保留精确 table generation，其 executor、authenticator
和 storage properties 都从同一 lease 读取，child scan 不再混用当前 catalog generation。
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergExternalMetaCache.java:
##########
@@ -105,12 +112,46 @@ public IcebergExternalMetaCache(ExecutorService
refreshExecutor) {
public Table getIcebergTable(ExternalTable dorisTable) {
NameMapping nameMapping = dorisTable.getOrBuildNameMapping();
- return
tableEntry.get(nameMapping.getCtlId()).get(nameMapping).getIcebergTable();
+ IcebergTableCacheValue.Lease lease = statementLease(nameMapping);
+ if (lease != null) {
+ return lease.getIcebergTable();
+ }
+ // Background/bootstrap callers without a StatementContext have no
deterministic release boundary.
+ // Load directly instead of borrowing a cache generation that could be
evicted while they use it.
+ return loadTable(nameMapping);
+ }
+
+ /** Returns the executor owned by the exact table generation retained by
this statement. */
+ ThreadPoolExecutor getIcebergTableExecutor(ExternalTable dorisTable) {
+ NameMapping nameMapping = dorisTable.getOrBuildNameMapping();
+ IcebergTableCacheValue.Lease lease = statementLease(nameMapping);
+ if (lease == null || lease.getPlanningExecutor() == null) {
+ return dorisTable.getCatalog().getThreadPoolWithPreAuth();
+ }
+ return lease.getPlanningExecutor();
+ }
+
+ /** Runs a bounded metadata operation while retaining the exact table
generation it uses. */
+ <T> T withIcebergTable(ExternalTable dorisTable, Function<Table, T>
action) {
+ NameMapping nameMapping = dorisTable.getOrBuildNameMapping();
+ IcebergTableCacheValue.Lease statementLease =
statementLease(nameMapping);
+ if (statementLease != null) {
+ return action.apply(statementLease.getIcebergTable());
+ }
+ try (IcebergTableCacheValue.Lease operationLease =
borrow(nameMapping)) {
+ return action.apply(operationLease.getIcebergTable());
+ }
}
public IcebergSnapshotCacheValue getSnapshotCache(ExternalTable
dorisTable) {
NameMapping nameMapping = dorisTable.getOrBuildNameMapping();
- return
tableEntry.get(nameMapping.getCtlId()).get(nameMapping).getLatestSnapshotCacheValue();
+ IcebergTableCacheValue.Lease lease = statementLease(nameMapping);
Review Comment:
这条评论要求处理 MTMVCache 内部 StatementContext 的通用生命周期与缓存 plan 所有权，属于既有 MV planning
生命周期问题，不属于本 PR 仅处理 Hudi/Iceberg 资源关闭与泄露的范围。当前 head 已撤回 MTMVCache 改动，本 PR 忽略该问题。
--
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]
