github-actions[bot] commented on code in PR #66913:
URL: https://github.com/apache/doris/pull/66913#discussion_r3828624016
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/ExecuteCommand.java:
##########
@@ -73,6 +73,9 @@ public <R, C> R accept(PlanVisitor<R, C> visitor, C context) {
@Override
public void run(ConnectContext ctx, StmtExecutor executor) throws
Exception {
StatementContext statementContext = ctx.getStatementContext();
+ // PREPARE retains this StatementContext, but ConnectProcessor closes
the resources from each
+ // COM_STMT_EXECUTE. Reopen an empty generation before the next
execution starts planning.
+ statementContext.beginStatementResourceGeneration();
Review Comment:
[P1] End the forwarded PREPARE generation before EXECUTE
`prepareExecuteBuffer` runs analyzed PREPARE and immediately calls
`handleExecute()` without closing resources between the two phases. Binding an
Iceberg relation during PREPARE registers a table lease, so this call sees a
nonempty map and throws `Previous statement resources are still active` before
the first forwarded execution. Please close the PREPARE resource generation
before `handleExecute()` and let EXECUTE open a fresh one, with coverage for a
forwarded prepared Iceberg query.
##########
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:
[P1] Do not wait uninterruptibly for stalled listing tasks during cleanup
Once a `TerminalTask` has passed `started`, `cancelBeforeStart()` cannot
signal it. This close path stops only the split assignment and then ignores
interrupts while waiting for every running remote listing to finish. If one
filesystem call stalls, statement cancellation (including Arrow DoGet
finalization) never returns. Please request cooperative
cancellation/interruption of running tasks and move exact lease release to
their actual-terminal callback, so cleanup can return without releasing the
fs-view early. A test with an already-started blocked listing should prove
cleanup returns while the lease stays pinned until that task exits.
##########
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:
[P1] Give MV-cache planning leases an owner
This registration also runs while `MTMVCache.from()` plans an Iceberg-backed
MV definition with a fresh internal `StatementContext`. Its finally restores
flags and the caller's thread-local context but never closes or transfers that
statement's leases, and the fresh-context callers have no later owner. Repeated
cache generation therefore pins table/FileIO/catalog/executor generations;
simply closing at the end of planning may also be too early if the returned
cached plan retains the frozen table. Please either transfer the leases to the
`MTMVCache` lifetime or remove runtime-bearing snapshot state before returning
it, with success/failure and cache-replacement coverage.
##########
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:
[P1] Seal entries when their catalog group is removed
The generation checks here are local to this `MetaCacheEntry`, so they do
not cover removal of the containing catalog group. A caller can obtain old
entry E, pause, let `invalidateCatalog()` remove E's group and finish its one
`invalidateAll()`, then resume here: it snapshots the already-incremented
generation, loads, and publishes into E successfully. Future catalog
invalidations address a replacement group, so the Hudi/Iceberg retirement
callback for this detached value is no longer reachable through catalog
lifecycle and can pin its filesystem/catalog generation. Please seal removed
groups/entries (retiring any post-seal load and forcing the caller to
re-resolve the current group), with a race test that starts the old-entry load
only after removal completes.
##########
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:
[P1] Carry the frozen snapshot's executor into child scans
An installed snapshot can come from catalog generation G1, but this child
first borrows current G2, replaces only its table with the frozen G1 table, and
then resolves `planningExecutor` from the G2 lease. A reset between parent
snapshot capture and child initialization therefore runs G1
`TableOperations`/FileIO work under G2's pre-auth executor. Please transfer the
exact table lease/executor with `IcebergMvccSnapshot` (or otherwise choose both
atomically from that snapshot), and cover reset-between-capture-and-child-plan
for rewrite and MTMV contexts.
##########
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:
[P1] Fence HMS Iceberg table generations across runtime reset
This fallback is used by Iceberg-format `HMSExternalTable`s, but it misses
both sides of the runtime-generation fence. First, an active statement can
retain the G1 table while reset immediately closes G1's `icebergMetadataOps`
and pre-auth pool; the scan then plans that table on a shut-down or G2 executor
because this value stores neither the exact executor nor a runtime owner.
Second, `HMSExternalCatalog.resetToUninitialized(false)` and its
property-update hook never remove `IcebergExternalMetaCache.ENGINE`, so even a
statement starting after reset can keep borrowing the retired G1 value. Please
remove the HMS Iceberg group before rotation, capture the exact
executor/runtime with each value, and defer G1 retirement until pre-fence
borrowers drain. A regression should prove an active G1 statement survives
while a new post-reset statement reloads G2.
--
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]
