924060929 commented on code in PR #66913:
URL: https://github.com/apache/doris/pull/66913#discussion_r3838741649
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/hudi/source/HudiScanNode.java:
##########
@@ -558,53 +594,213 @@ private void initPrunedPartitions() throws UserException
{
public void startSplit(int numBackends) {
if (prunedPartitions.isEmpty()) {
splitAssignment.finishSchedule();
+ releaseFsViewOnce();
return;
}
- AtomicInteger numFinishedPartitions = new AtomicInteger(0);
+ acquireFsView();
ExecutorService scheduleExecutor =
Env.getCurrentEnv().getExtMetaCacheMgr().getScheduleExecutor();
+ Executor producerExecutor =
Env.getCurrentEnv().getExtMetaCacheMgr().getFileListingExecutor();
long startTime = System.currentTimeMillis();
- CompletableFuture.runAsync(() -> {
- for (HivePartition partition : prunedPartitions) {
- if (batchException.get() != null || splitAssignment.isStop()) {
- break;
- }
- try {
- splittersOnFlight.acquire();
- } catch (InterruptedException e) {
- batchException.set(new UserException(e.getMessage(), e));
- break;
+ BatchFsViewOwner createdOwner = new BatchFsViewOwner(splitAssignment,
fsViewLease);
+ BatchFsViewOwner batchOwner = createdOwner;
+ ConnectContext connectContext = ConnectContext.get();
+ StatementContext statementContext = connectContext == null ? null :
connectContext.getStatementContext();
+ if (statementContext != null) {
+ try {
+ batchOwner = statementContext.getOrRegisterStatementResource(
+ batchFsViewResourceKey, () -> createdOwner);
+ if (batchOwner != createdOwner) {
+ createdOwner.finish();
+ throw new IllegalStateException("Hudi batch split owner
was registered twice");
}
- CompletableFuture.runAsync(() -> {
+ } catch (RuntimeException e) {
+ createdOwner.finish();
+ throw e;
+ }
+ }
+
+ BatchFsViewOwner finalBatchOwner = batchOwner;
+ AtomicInteger pendingTasks = new AtomicInteger(1); // producer
reference
+ Runnable taskFinished = () -> {
+ if (pendingTasks.decrementAndGet() == 0) {
+ finishBatchSplit(finalBatchOwner, startTime);
+ }
+ };
+ TerminalTask producerTask = terminalTask(() -> {
+ try {
+ for (HivePartition partition : prunedPartitions) {
+ if (batchException.get() != null ||
splitAssignment.isStop()) {
+ break;
+ }
try {
- List<Split> allFiles = Lists.newArrayList();
- getPartitionSplits(partition, allFiles, false);
- if (allFiles.size() > numSplitsPerPartition.get()) {
- numSplitsPerPartition.set(allFiles.size());
- }
- if (splitAssignment.needMoreSplit()) {
- splitAssignment.addToQueue(allFiles);
- }
- } catch (Exception e) {
- batchException.set(new UserException(e.getMessage(),
e));
- } finally {
+ splittersOnFlight.acquire();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ recordBatchException(e);
+ break;
+ }
+ if (batchException.get() != null ||
splitAssignment.isStop()) {
splittersOnFlight.release();
- if (batchException.get() != null) {
- splitAssignment.setException(batchException.get());
- }
- if (numFinishedPartitions.incrementAndGet() ==
prunedPartitions.size()) {
- if (getSummaryProfile() != null) {
-
getSummaryProfile().addExternalTableGetFileScanTasksTime(
- System.currentTimeMillis() -
startTime);
+ break;
+ }
+ pendingTasks.incrementAndGet();
+ TerminalTask partitionTask = terminalTask(() -> {
+ try {
+ List<Split> allFiles = Lists.newArrayList();
+ getPartitionSplits(partition, allFiles, false);
+ if (allFiles.size() > numSplitsPerPartition.get())
{
+ numSplitsPerPartition.set(allFiles.size());
+ }
+ if (splitAssignment.needMoreSplit()) {
+ splitAssignment.addToQueue(allFiles);
}
- splitAssignment.finishSchedule();
+ } catch (Throwable t) {
+ recordBatchException(t);
}
+ }, () -> {
+ splittersOnFlight.release();
+ taskFinished.run();
+ });
+ finalBatchOwner.track(partitionTask);
+ try {
+ scheduleExecutor.execute(partitionTask);
+ } catch (RuntimeException e) {
+ recordBatchException(e);
+ partitionTask.cancelBeforeStart();
+ break;
}
- }, scheduleExecutor);
+ }
+ } catch (Throwable t) {
+ recordBatchException(t);
+ }
+ }, taskFinished);
+ finalBatchOwner.track(producerTask);
+ try {
+ producerExecutor.execute(producerTask);
+ } catch (RuntimeException e) {
+ recordBatchException(e);
+ producerTask.cancelBeforeStart();
+ }
+ }
+
+ private TerminalTask terminalTask(Runnable task, Runnable taskFinished) {
+ return new TerminalTask(task, taskFinished);
+ }
+
+ @VisibleForTesting
+ static class TerminalTask extends FutureTask<Void> {
+ private final AtomicBoolean started = new AtomicBoolean();
+ private final Runnable taskFinished;
+
+ TerminalTask(Runnable task, Runnable taskFinished) {
+ super(task, null);
+ this.taskFinished = taskFinished;
+ }
+
+ @Override
+ public void run() {
+ if (started.compareAndSet(false, true)) {
+ super.run();
}
+ }
+
+ boolean cancelBeforeStart() {
+ return started.compareAndSet(false, true) && cancel(false);
+ }
+
+ @Override
+ protected void done() {
+ taskFinished.run();
+ }
+ }
+
+ private void recordBatchException(Throwable t) {
+ batchException.compareAndSet(null, new UserException(t.getMessage(),
t));
+ }
+
+ private void finishBatchSplit(BatchFsViewOwner batchOwner, long startTime)
{
+ try {
if (batchException.get() != null) {
splitAssignment.setException(batchException.get());
}
- }, scheduleExecutor);
+ if (getSummaryProfile() != null) {
+
getSummaryProfile().addExternalTableGetFileScanTasksTime(System.currentTimeMillis()
- startTime);
+ }
+ splitAssignment.finishSchedule();
+ } finally {
+ batchOwner.finish();
+ }
+ }
+
+ @VisibleForTesting
+ static class BatchFsViewOwner implements Closeable {
+ private final SplitAssignment splitAssignment;
+ private final HudiFsViewCacheValue.Lease lease;
+ private final AtomicBoolean finished = new AtomicBoolean();
+ private final AtomicReference<RuntimeException> finishFailure = new
AtomicReference<>();
+ private final CountDownLatch terminal = new CountDownLatch(1);
+ private final ConcurrentLinkedQueue<TerminalTask> tasks = new
ConcurrentLinkedQueue<>();
+ private final AtomicBoolean stopping = new AtomicBoolean();
+
+ BatchFsViewOwner(SplitAssignment splitAssignment,
HudiFsViewCacheValue.Lease lease) {
+ this.splitAssignment = splitAssignment;
+ this.lease = lease;
+ }
+
+ void finish() {
+ if (finished.compareAndSet(false, true)) {
+ try {
+ lease.close();
+ } catch (RuntimeException e) {
+ finishFailure.set(e);
+ throw e;
+ } finally {
+ terminal.countDown();
+ }
+ }
+ }
+
+ void track(TerminalTask task) {
+ tasks.add(task);
+ if (stopping.get()) {
+ task.cancelBeforeStart();
+ }
+ }
+
+ @Override
+ public void close() {
+ RuntimeException stopFailure = null;
+ if (!finished.get()) {
+ stopping.set(true);
+ try {
+ splitAssignment.stop();
+ } catch (RuntimeException e) {
+ stopFailure = e;
+ }
+ tasks.forEach(TerminalTask::cancelBeforeStart);
+ }
+ boolean interrupted = false;
+ while (true) {
+ try {
+ terminal.await();
Review Comment:
已修复。Hudi batch owner close 只请求 running TerminalTask 停止并立即返回；精确 fs-view lease
仍由各任务 actual-terminal callback 计数，最后一个任务退出后才释放。
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergExternalMetaCache.java:
##########
@@ -159,23 +200,132 @@ 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,
ops.getThreadPoolWithPreAuth(),
+ () -> 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),
Review Comment:
已修复。HMS Iceberg table value 携带精确 runtime generation/executor lease；reset 先移除
Iceberg cache group，再延迟关闭旧 metadata ops/executor 到 borrower drain。
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/hive/HMSExternalCatalog.java:
##########
@@ -233,12 +249,59 @@ public void setDefaultPropsIfMissing(boolean isReplay) {
}
}
- public IcebergMetadataOps getIcebergMetadataOps() {
+ public synchronized IcebergMetadataOps getIcebergMetadataOps() {
makeSureInitialized();
if (icebergMetadataOps == null) {
HiveCatalog icebergHiveCatalog =
IcebergUtils.createIcebergHiveCatalog(this, getName());
icebergMetadataOps =
ExternalMetadataOperations.newIcebergMetadataOps(this, icebergHiveCatalog);
}
return icebergMetadataOps;
}
+
+ /** Retains the exact HMS Iceberg runtime while a table cache generation
is being loaded or borrowed. */
+ public synchronized IcebergTableLoadContext beginIcebergTableLoad() {
+ makeSureInitialized();
+ IcebergMetadataOps ops = getIcebergMetadataOps();
+ return new IcebergTableLoadContext(ops, threadPoolWithPreAuth,
icebergResourceTracker.beginLoad());
+ }
+
+ @Override
+ public synchronized void resetToUninitialized(boolean invalidCache) {
+
Env.getCurrentEnv().getExtMetaCacheMgr().removeCatalogByEngine(getId(),
IcebergExternalMetaCache.ENGINE);
Review Comment:
这条评论同时要求重构普通 Hive scan、Hudi 初始化和整个 HMS runtime generation 协议，超出本 PR 仅处理 Hudi
fsView 与 Iceberg FileIO/catalog 关闭泄露的范围。当前 head 已撤回 Hive scan 旁支改动，本 PR 忽略该问题。
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergExternalMetaCache.java:
##########
@@ -105,12 +113,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);
+ if (lease != null) {
+ return lease.getLatestSnapshotCacheValue();
+ }
+ Table table = loadTable(nameMapping);
Review Comment:
已修复。无 ConnectContext 的 snapshot/version 投影通过有界 exact-generation load owner
执行，导出非 runtime 数据后在 finally 释放 table FileIO/catalog runtime。
--
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]
