github-actions[bot] commented on code in PR #66913:
URL: https://github.com/apache/doris/pull/66913#discussion_r3834870612
##########
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:
[P1] Keep non-batch Hudi planning cancellable
This unconditional `join()` fixes early lease release, but it also makes the
synchronous planning thread impossible to cancel while any accepted filesystem
listing is stalled. These futures have no owner that signals running tasks, and
`getSplits()` cannot reach its lease-release `finally` until every remote call
exits, so query cancellation can remain wedged indefinitely. This is the
non-batch counterpart to the already-reported batch cleanup stall, not the
earlier premature-release issue. Give accepted work a cancellation-aware owner
that can return promptly while retaining the exact view until terminal
callbacks drain, and cover an already-started blocked listing.
##########
fe/fe-core/src/main/java/org/apache/doris/service/arrowflight/sessions/FlightSqlConnectPoolMgr.java:
##########
@@ -74,7 +74,7 @@ public void unregisterConnection(ConnectContext ctx) {
// Finalize any Arrow Flight query whose coordinator was kept alive
across the
// GetFlightInfo -> DoGet phases (see #62259), releasing its resources
(e.g. external-table
// batch SplitSources and the query queue slot).
- ctx.closeFlightSqlDeferredExecutors();
+ ctx.sealAndCloseFlightSqlDeferredExecutors();
Review Comment:
[P1] Seal publication before destroying the Flight channel
`unregisterConnection()` closes `FlightSqlChannel` before installing the
terminal publication seal. A concurrent GetFlightInfo publisher can therefore
finish its schema/ticket work after the channel close, call
`endFlightSqlResultPublication()` in this interval, observe
`flightSqlDeferredExecutorsSealed == false`, and return success; teardown then
seals/drains its coordinator and detached resources, leaving the client a dead
ticket (or a local ticket backed by the closed channel). This is a distinct
caller-side window from the earlier producer-finally race. Make sealing the
first teardown transition, before closing the channel, and add a barrier test
for this ordering.
##########
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:
[P1] Fence lazy cache-group publication across reset
Reset can miss an initialization that has snapshotted the old catalog
properties but not yet published its group. In that interleaving, this removal
sees no group and completes; the accessor then reaches `initCatalog()` and
`computeIfAbsent` installs a G1-configured group after the G2 reset. Subsequent
G2 accessors keep it, so obsolete enable/TTL/capacity settings can persist
indefinitely. This is different from retaining an already-detached entry: the
stale group is first published after reset. Fence publication with a
per-catalog epoch (the native Iceberg reset has the same shape), reject stale
installs, and retry with current properties.
##########
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:
[P1] Close native table IO when wrapper assembly fails
After `loadContext.loadTable()` returns, this local-table lookup can fail if
a Doris table is dropped/refreshed while the remote load is in flight. At that
point Glue/S3 Tables (and REST-vended) per-table FileIO already exists, but
`tableCleanup` has not been created and no `IcebergTableCacheValue` exists for
the retirement listener to reach; the try-with-resources close releases only
the catalog-generation load guard. The unreachable table IO is therefore
leaked. Establish the table cleanup handle immediately after the raw load, run
it on every later assembly failure, and transfer it only once the cache value
is fully constructed. Add a barrier test for drop-after-remote-load.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/StatementContext.java:
##########
@@ -918,7 +1025,11 @@ protected void finalize() throws Throwable {
@Override
public void close() {
clearExternalScanTasks();
- releasePlannerResources();
+ try {
+ releaseStatementResources();
Review Comment:
[P1] Transfer statement resources to lazy PLSQL cursors
`runProcedure` intentionally returns before `DorisRowResult` starts its lazy
`Coordinator.getNext()` calls, but the generic query/context finally closes
this `StatementContext` immediately. With this release, a batch Hudi cursor's
`BatchFsViewOwner` stops `SplitAssignment` before the cursor's first or later
fetch, which can truncate the result. This deferred consumer is separate from
Arrow DoGet and external-statistics contexts. Detach these statement resources
into the returned `QueryResult`/row result and close them on EOF, cursor close,
error, or abandonment; `DorisRowResult.close()` is currently empty.
##########
fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/AbstractStreamingTask.java:
##########
@@ -94,34 +97,58 @@ public long getRunningBackendId() {
}
public void execute() throws JobException {
- while (retryCount <= MAX_RETRY) {
- try {
- before();
- run();
- onSuccess();
- return;
- } catch (Exception e) {
- if (TaskStatus.CANCELED.equals(status)) {
- return;
- }
- this.errMsg = e.getMessage();
- retryCount++;
- if (noRetry || retryCount > MAX_RETRY) {
- log.error("Task execution failed, job id {}, task id {},
noRetry {}, retry {}.",
- jobId, taskId, noRetry, retryCount, e);
- onFail(e.getMessage());
+ synchronized (executionCompletion) {
+ executionStarted = true;
+ }
+ try {
+ while (retryCount <= MAX_RETRY) {
+ try {
+ before();
+ run();
+ onSuccess();
return;
- }
- log.warn("execute streaming task error, job id is {}, task id
is {}, retrying {}/{}: {}",
- jobId, taskId, retryCount, MAX_RETRY, e.getMessage());
- } finally {
- // The cancel logic will call the closeOrReleased Resources
method by itself.
- // If it is also called here,
- // it may result in the inability to obtain relevant
information when canceling the task
- if (!TaskStatus.CANCELED.equals(status)) {
+ } catch (Exception e) {
+ if (TaskStatus.CANCELED.equals(status)) {
+ return;
+ }
+ this.errMsg = e.getMessage();
+ retryCount++;
+ if (noRetry || retryCount > MAX_RETRY) {
+ log.error("Task execution failed, job id {}, task id
{}, noRetry {}, retry {}.",
+ jobId, taskId, noRetry, retryCount, e);
+ onFail(e.getMessage());
+ return;
+ }
+ log.warn("execute streaming task error, job id is {}, task
id is {}, retrying {}/{}: {}",
+ jobId, taskId, retryCount, MAX_RETRY,
e.getMessage());
+ } finally {
+ // Only the scheduler worker that created this attempt's
ConnectContext may tear it down.
+ // A cancelling thread waits for this handoff instead of
racing before() and clearing fields
+ // while planning is still publishing them.
closeOrReleaseResources();
Review Comment:
[P1] Keep cleanup failures inside the streaming state machine
`closeOrReleaseResources()` runs in this inner `finally`, after the catch
has decided to retry or return. If cleanup throws, Java propagates it past the
loop, so `onFail()` is never reached and the job/task can remain `RUNNING`.
This is reachable because `StatementContext.close()` rethrows resource-close
failures; moreover, `StreamingInsertTask` performs unregister then statement
close in one try and nulls `ctx` in an outer finally, so an earlier failure can
skip the only remaining cleanup handle. Run every cleanup leg with
nested/aggregated failure handling, then feed any failure through
retry/terminal bookkeeping rather than letting it escape this boundary.
##########
fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertTask.java:
##########
@@ -173,24 +174,41 @@ protected void onFail(String errMsg) throws JobException {
@Override
public void cancel(boolean needWaitCancelComplete) {
super.cancel(needWaitCancelComplete);
- if (null != stmtExecutor) {
+ StmtExecutor executor = stmtExecutor;
+ if (null != executor) {
log.info("cancelling streaming insert task, job id is {}, task id
is {}",
getJobId(), getTaskId());
- stmtExecutor.cancel(new Status(TStatusCode.CANCELLED, "streaming
insert task cancelled"),
+ executor.cancel(new Status(TStatusCode.CANCELLED, "streaming
insert task cancelled"),
needWaitCancelComplete);
}
+ if (needWaitCancelComplete) {
+ awaitExecutionCompletion();
Review Comment:
[P1] Avoid waiting for the streaming worker from itself
On the last failed attempt, `execute()` calls `onFail()` before its outer
`executionFinished` finally. That callback synchronously pauses the job, clears
the running task, and calls this task's `cancel(true)`. Even though
`super.cancel()` returns for the already-`FAILED` task, this override continues
here and waits; the current scheduler worker is the only thread that can set
`executionFinished`, so it deadlocks while the pause transition still holds the
job lock. Skip this wait when cancellation runs on the execution
owner/already-terminal task, or publish completion before the synchronous
pause/cancel callback. Add a final-attempt failure regression.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/metacache/MetaCacheEntry.java:
##########
@@ -230,7 +270,9 @@ private V getWithManualLoad(K key, Function<K, V>
loadFunction) {
long generation = invalidateGeneration.get();
V loaded = loadAndTrack(key, loadFunction);
- if (generation != invalidateGeneration.get()) {
+ if (sealed.get() || generation != invalidateGeneration.get()) {
+ notifySuppressedRemoval(key, loaded);
Review Comment:
[P1] Consume the Hudi loader reference for suppressed values
When a Hudi view finishes loading after this entry is sealed, the
suppressed-removal callback does run, but it only calls `evict()`. A new
`HudiFsViewCacheValue` starts with its transferable loader reference counted at
one, and only `tryAcquire()` consumes that reference; `ensureOpen()` now throws
before this value is returned to `getFsView()`, so no caller can ever
consume/release it. The detached, uncached view therefore stays evicted at
refcount one and never closes. This is the terminal-path leak exposed by the
new seal fix, not the earlier detached-publication race. Retire never-returned
values with an operation that consumes the loader reference, and retry
`getFsView()` from the current group after the stale-entry signal.
--
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]
