924060929 commented on code in PR #66913:
URL: https://github.com/apache/doris/pull/66913#discussion_r3838742132
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/hudi/source/HudiScanNode.java:
##########
@@ -494,22 +532,31 @@ private List<HudiSplit> planPartitionSplits(HivePartition
partition) throws IOEx
private void getPartitionsSplits(List<HivePartition> partitions,
List<Split> splits) {
Executor executor =
Env.getCurrentEnv().getExtMetaCacheMgr().getFileListingExecutor();
- CountDownLatch countDownLatch = new CountDownLatch(partitions.size());
+ List<CompletableFuture<Void>> acceptedTasks = new
ArrayList<>(partitions.size());
AtomicReference<Throwable> throwable = new AtomicReference<>();
+ RuntimeException submissionFailure = null;
long startTime = System.currentTimeMillis();
- partitions.forEach(partition -> executor.execute(() -> {
+ for (HivePartition partition : partitions) {
try {
- getPartitionSplits(partition, splits);
- } catch (Throwable t) {
- throwable.set(t);
- } finally {
- countDownLatch.countDown();
+ acceptedTasks.add(CompletableFuture.runAsync(() -> {
+ try {
+ ensureHmsRuntimeGeneration();
+ getPartitionSplits(partition, splits);
+ ensureHmsRuntimeGeneration();
+ } catch (Throwable t) {
+ throwable.compareAndSet(null, t);
+ }
+ }, executor));
+ } catch (RuntimeException e) {
+ submissionFailure = e;
+ break;
}
- }));
- try {
- countDownLatch.await();
- } catch (InterruptedException e) {
- throw new RuntimeException(e.getMessage(), e);
+ }
+ // CompletableFuture.allOf has no Phaser party limit and join is
uninterruptible: every accepted task is
+ // terminal before the caller releases the filesystem-view lease,
including submission rejection.
+ CompletableFuture.allOf(acceptedTasks.toArray(new
CompletableFuture[0])).join();
Review Comment:
已修复。non-batch listing owner 支持 cancellation-aware TerminalTask；取消立即返回，已开始任务的
lease 由 actual-terminal callback 保留到退出。
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/hive/HMSExternalCatalog.java:
##########
@@ -233,12 +263,62 @@ 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) {
+ runtimeGeneration.incrementAndGet();
+
Env.getCurrentEnv().getExtMetaCacheMgr().removeCatalogByEngine(getId(),
HiveExternalMetaCache.ENGINE);
Review Comment:
已修复。per-catalog lifecycle fence 覆盖 group init/publication 与 reset
removal；旧属性快照不能在 reset 完成后发布 stale group，detached entry 同时会被 seal。
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergExternalMetaCache.java:
##########
@@ -159,23 +204,155 @@ 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);
Review Comment:
已修复。raw Iceberg Table load 后立即建立 cleanup ownership；后续 Doris wrapper/table
lookup 组装失败会在 finally 清理 table-owned FileIO，只有完整 cache value 成功后才转移所有权。
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/hudi/source/HudiScanNode.java:
##########
@@ -552,59 +628,305 @@ private void initPrunedPartitions() throws UserException
{
throw new UserException(ExceptionUtils.getRootCauseMessage(e), e);
}
partitionInit = true;
+ ensureHmsRuntimeGeneration();
}
@Override
public void startSplit(int numBackends) {
+ ensureHmsRuntimeGeneration();
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 {
+ ensureHmsRuntimeGeneration();
+ 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 {
+ ensureHmsRuntimeGeneration();
+ List<Split> allFiles = Lists.newArrayList();
+ getPartitionSplits(partition, allFiles, false);
+ if (allFiles.size() > numSplitsPerPartition.get())
{
+ numSplitsPerPartition.set(allFiles.size());
}
- splitAssignment.finishSchedule();
+ if (splitAssignment.needMoreSplit()) {
+ ensureHmsRuntimeGeneration();
+ splitAssignment.addToQueue(allFiles);
+ }
+ } 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 AtomicBoolean interruptRequested = new AtomicBoolean();
+ private final Runnable taskFinished;
+ private volatile Thread runner;
+ private volatile Runnable ownerDone = () -> { };
+
+ TerminalTask(Runnable task, Runnable taskFinished) {
+ super(task, null);
+ this.taskFinished = taskFinished;
+ }
+
+ @Override
+ public void run() {
+ if (started.compareAndSet(false, true)) {
+ runner = Thread.currentThread();
+ if (interruptRequested.get()) {
+ runner.interrupt();
+ }
+ try {
+ super.run();
+ } finally {
+ runner = null;
+ }
}
+ }
+
+ boolean cancelBeforeStart() {
+ return started.compareAndSet(false, true) && cancel(false);
+ }
+
+ void requestStop() {
+ if (cancelBeforeStart()) {
+ return;
+ }
+ interruptRequested.set(true);
+ Thread runningThread = runner;
+ if (runningThread != null) {
+ runningThread.interrupt();
+ }
+ }
+
+ void setOwnerDone(Runnable ownerDone) {
+ this.ownerDone = ownerDone;
+ }
+
+ @Override
+ protected void done() {
+ try {
+ taskFinished.run();
+ } finally {
+ ownerDone.run();
+ }
+ }
+ }
+
+ private void recordBatchException(Throwable t) {
+ batchException.compareAndSet(null, new UserException(t.getMessage(),
t));
Review Comment:
已修复。首个 Hudi batch failure 会立即调用 SplitAssignment.setException；sibling
terminal accounting 独立继续，确保 fs-view lease 仍在所有任务结束后释放。
--
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]
