924060929 commented on code in PR #66913:
URL: https://github.com/apache/doris/pull/66913#discussion_r3838736426
##########
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 table lease 现在携带同一 generation 的 executor、authenticator 和
storage-property 快照，scan/sink 不再从可变 live catalog 拼装跨代运行时。
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergExternalMetaCache.java:
##########
@@ -105,12 +113,49 @@ 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);
Review Comment:
已修复。无 StatementContext 的探测改为有界 withIcebergTable 操作，在精确 generation guard
内执行投影并在 finally 清理表级 FileIO。
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/ExternalMetaCacheMgr.java:
##########
@@ -228,15 +244,30 @@ public void invalidateCatalogByEngine(long catalogId,
String engine) {
}
public void removeCatalog(long catalogId) {
- routeCatalogEngines(catalogId, cache -> safeInvalidate(
- cache, catalogId, "removeCatalog",
- () -> cache.invalidateCatalog(catalogId)));
+ synchronized (catalogLifecycleLock(catalogId)) {
+ routeCatalogEngines(catalogId, cache -> safeInvalidate(
+ cache, catalogId, "removeCatalog",
+ () -> cache.invalidateCatalog(catalogId)));
+ }
}
public void removeCatalogByEngine(long catalogId, String engine) {
- routeSpecifiedEngine(engine, cache -> safeInvalidate(
- cache, catalogId, "removeCatalogByEngine",
- () -> cache.invalidateCatalog(catalogId)));
+ synchronized (catalogLifecycleLock(catalogId)) {
+ routeSpecifiedEngine(engine, cache -> safeInvalidate(
+ cache, catalogId, "removeCatalogByEngine",
+ () -> cache.invalidateCatalog(catalogId)));
+ }
+ }
+
+ /**
+ * Fences a catalog runtime transition against lazy cache-group
initialization. The transition callback
+ * must cover both cache removal and the catalog property/runtime
mutation; otherwise an accessor can
+ * snapshot the retiring properties after removal and publish that group
into the new generation.
+ */
+ public void runCatalogLifecycle(long catalogId, Runnable transition) {
Review Comment:
这条评论要求改变 CatalogMgr 的 tentative ALTER validation/rollback 协议，属于既有 Catalog
ALTER 生命周期问题，不属于本 PR 仅处理 Hudi/Iceberg 资源关闭与泄露的范围。当前 head 已撤回 CatalogMgr 改动，本 PR
忽略该问题。
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/source/IcebergScanNode.java:
##########
@@ -692,52 +692,86 @@ public void startSplit(int numBackends) throws
UserException {
}
public void doStartSplit() throws UserException {
- TableScan scan = createTableScan();
- CompletableFuture.runAsync(() -> {
- AtomicReference<CloseableIterable<FileScanTask>> taskRef = new
AtomicReference<>();
- try {
- preExecutionAuthenticator.execute(
- () -> {
- long startTime = System.currentTimeMillis();
- try {
- CloseableIterable<FileScanTask> fileScanTasks
= planFileScanTask(scan);
- taskRef.set(fileScanTasks);
- CloseableIterator<FileScanTask> iterator =
fileScanTasks.iterator();
- while (splitAssignment.needMoreSplit() &&
iterator.hasNext()) {
- try {
- splitAssignment.addToQueue(
-
Lists.newArrayList(createIcebergSplit(iterator.next())));
- } catch (UserException e) {
- throw new RuntimeException(e);
+ IcebergTableCacheValue.Lease planningLease =
retainPlanningGeneration();
+ TableScan scan;
+ try {
+ scan = createTableScan();
+ } catch (UserException | RuntimeException | Error t) {
+ planningLease.close();
+ throw t;
+ }
+ Future<?> planningFuture;
+ try {
+ planningFuture =
Env.getCurrentEnv().getExtMetaCacheMgr().getScheduleExecutor().submit(() -> {
+ AtomicReference<CloseableIterable<FileScanTask>> taskRef = new
AtomicReference<>();
+ try {
+ preExecutionAuthenticator.execute(
+ () -> {
+ long startTime = System.currentTimeMillis();
+ try {
+ CloseableIterable<FileScanTask>
fileScanTasks = planFileScanTask(scan);
+ taskRef.set(fileScanTasks);
+ CloseableIterator<FileScanTask> iterator =
fileScanTasks.iterator();
+ while (splitAssignment.needMoreSplit() &&
iterator.hasNext()) {
+ try {
+ splitAssignment.addToQueue(
+
Lists.newArrayList(createIcebergSplit(iterator.next())));
+ } catch (UserException e) {
+ throw new RuntimeException(e);
+ }
+ }
+ } finally {
+ if (getSummaryProfile() != null) {
+
getSummaryProfile().addExternalTableGetFileScanTasksTime(
+ System.currentTimeMillis() -
startTime);
}
- }
- } finally {
- if (getSummaryProfile() != null) {
-
getSummaryProfile().addExternalTableGetFileScanTasksTime(
- System.currentTimeMillis() -
startTime);
}
}
+ );
+ splitAssignment.finishSchedule();
+ recordManifestCacheProfile();
+ } catch (Exception e) {
+ Optional<NotSupportedException> opt =
checkNotSupportedException(e);
+ if (opt.isPresent()) {
+ splitAssignment.setException(new
UserException(opt.get().getMessage(), opt.get()));
+ } else {
+ splitAssignment.setException(new
UserException(e.getMessage(), e));
+ }
+ } finally {
+ if (taskRef.get() != null) {
+ try {
+ taskRef.get().close();
+ } catch (IOException e) {
+ // ignore
}
- );
- splitAssignment.finishSchedule();
- recordManifestCacheProfile();
- } catch (Exception e) {
- Optional<NotSupportedException> opt =
checkNotSupportedException(e);
- if (opt.isPresent()) {
- splitAssignment.setException(new
UserException(opt.get().getMessage(), opt.get()));
- } else {
- splitAssignment.setException(new
UserException(e.getMessage(), e));
- }
- } finally {
- if (taskRef.get() != null) {
- try {
- taskRef.get().close();
- } catch (IOException e) {
- // ignore
}
+ planningLease.close();
}
- }
- }, Env.getCurrentEnv().getExtMetaCacheMgr().getScheduleExecutor());
+ });
+ } catch (RuntimeException | Error t) {
+ planningLease.close();
+ throw t;
+ }
+ // Cancellation interrupts the worker, while the lease remains owned
by its actual-terminal finally.
+ splitAssignment.addCloseable(() -> planningFuture.cancel(true));
Review Comment:
已修复。SplitAssignment.stop() 与 addCloseable() 现在共享同一同步临界区：stop 原子取得已注册资源；stop
后注册的 handle 会立即关闭，不再并发遍历 ArrayList。新增回归测试已通过。
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/metacache/MetaCacheEntry.java:
##########
@@ -107,7 +127,13 @@ public MetaCacheEntry(String name, @Nullable Function<K,
V> loader, CacheSpec ca
maxSize,
true,
null);
- this.loadingData =
cacheFactory.buildCache(this::loadFromDefaultLoader, refreshExecutor);
+ if (retirementListener != null) {
+ this.loadingData = cacheFactory.buildCacheWithAsyncRemovalListener(
Review Comment:
已修复。Hudi/Iceberg admitted-value retirement 改为同步 removal
callback，不再依赖可拒绝或已关闭的 refresh executor；新增 executor 关闭后 invalidation 仍完成
retirement 的测试。
--
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]
