github-actions[bot] commented on code in PR #66913:
URL: https://github.com/apache/doris/pull/66913#discussion_r3838204870
##########
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:
[P1] Register this cancellation handle atomically with stop
`SplitAssignment.addCloseable()` mutates a plain `ArrayList`, while `stop()`
can concurrently traverse that same list with fail-fast `forEach`. If teardown
races this registration, the add can make `stop()` throw
`ConcurrentModificationException` before it notifies waiters and before
`FileQueryScanNode.stop()` removes the split sources. The following `isStop()`
check only repairs a clean missed-registration race; it cannot finish an
already-aborted stop. Please provide a synchronized register-or-close operation
shared with `stop()`, and cover registration versus stop with a barrier test.
##########
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:
[P1] Make admitted-value retirement non-droppable
This delegates retirement of admitted Hudi/Iceberg values to Caffeine on the
bounded shared refresh executor. That executor's rejection policy throws after
saturation/timeout or shutdown, so a rejected removal-listener dispatch never
reaches `retirementListener`; the evicted filesystem view or
table/FileIO/catalog generation then remains pinned. The inline fallback below
protects only suppressed values that never stayed in Caffeine. Please give
admitted eviction/invalidation/refresh replacement the same non-droppable
cleanup guarantee, and test an admitted resource removal with a rejecting
listener executor.
##########
fe/fe-core/src/main/java/org/apache/doris/qe/StmtExecutor.java:
##########
@@ -502,6 +506,12 @@ public boolean isCached() {
// query with a random sql
public void execute() throws Exception {
+ synchronized (executionAdmissionLock) {
+ if (pendingCancelReason != null) {
Review Comment:
[P1] Apply this admission guard to the proxy entry point
Ordinary forwarded execution never calls this overload:
`ConnectProcessor.proxyExecute()` invokes `executor.queryRetry(queryId)`
directly. A pre-registration cancel is therefore replayed into
`pendingCancelReason`, but the master still enters parsing/planning because
neither `queryRetry(TUniqueId)` nor `execute(TUniqueId)` checks it. FE-local
work can complete after cancellation, and external planning can block before
any coordinator exists to signal. Centralize this check in the shared query-ID
entry path and add a proxy-execution barrier test; the new test covers only
zero-argument `execute()`.
##########
fe/fe-core/src/main/java/org/apache/doris/service/arrowflight/sessions/FlightSqlConnectPoolMgr.java:
##########
@@ -56,6 +58,12 @@ public int registerConnection(ConnectContext ctx) {
@Override
public void unregisterConnection(ConnectContext ctx) {
+ // Reject new publications first, then signal the active query before
waiting for an admitted
+ // GetFlightInfo publisher. Waiting before cancellation can deadlock
KILL CONNECTION behind the
+ // publisher whose query must be canceled in order to leave
publication.
+ ctx.sealFlightSqlDeferredExecutors();
+ ctx.cancelQuery(new Status(TStatusCode.CANCELLED, "arrow flight
connection closed"));
Review Comment:
[P1] Keep teardown running when cancellation fails
For a forwarded query, this call can throw: `StmtExecutor.cancel()` wraps an
RPC or journal-wait failure from `MasterOpExecutor.cancel()` in
`RuntimeException`. That aborts this method before it drains deferred
executors, closes the channel/transaction, and removes the connection and token
mappings, so a transient leader/RPC failure can strand the entire Flight
session. Please make the teardown legs failure-atomic (for example, nested
`finally` cleanup with aggregated reporting), and add a test where
`cancelQuery()` throws but every later cleanup still runs.
##########
fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJob.java:
##########
@@ -556,19 +557,141 @@ public void alterJob(AlterJobCommand alterJobCommand)
throws AnalysisException,
@Override
public void updateJobStatus(JobStatus status) throws JobException {
+ AbstractStreamingTask taskToCancel = null;
+ boolean waitForTask = JobStatus.PAUSED.equals(status);
lock.writeLock().lock();
try {
+ if ((JobStatus.PAUSED.equals(status) ||
JobStatus.STOPPED.equals(status))
+ && status != getJobStatus()) {
+ taskToCancel = runningStreamTask;
+ }
+ JobStatus previousStatus = getJobStatus();
super.updateJobStatus(status);
- if (JobStatus.PAUSED.equals(getJobStatus())) {
- clearRunningStreamTask(status);
+ if (previousStatus != getJobStatus()) {
+ statusEpoch++;
}
if (isFinalStatus()) {
Env.getCurrentGlobalTransactionMgr().getCallbackFactory().removeCallback(getJobId());
}
log.info("Streaming insert job {} update status to {}",
getJobId(), getJobStatus());
+ } catch (RuntimeException | JobException e) {
+ if (taskToCancel != null) {
+ runningStreamTask = taskToCancel;
+ }
+ throw e;
+ } finally {
+ lock.writeLock().unlock();
+ }
+ if (taskToCancel != null && waitForTask) {
+ // The task owner can need this job's write lock while finishing
transaction callbacks.
+ // Cancel and wait only after publishing the status and releasing
the job lock.
+ taskToCancel.cancel(waitForTask);
+ }
+ }
+
+ /** Applies a validated manual transition while reason, status, and
cancellation share one job lock. */
+ public void updateManualJobStatus(JobStatus status, FailureReason reason)
throws JobException {
+ AbstractStreamingTask taskToWait = null;
+ AbstractStreamingTask taskToRelease = null;
+ JobStatus publishedStatus = JobStatus.RUNNING.equals(status) ?
JobStatus.PENDING : status;
+ lock.writeLock().lock();
+ try {
+ validateManualStatusTransition(status);
+ resetFailureInfo(reason);
+ if (JobStatus.PAUSED.equals(status) && runningStreamTask != null) {
+ needRebuildReader = true;
+ taskToRelease = runningStreamTask;
+ }
+ if (JobStatus.PAUSED.equals(status) ||
JobStatus.STOPPED.equals(status)) {
+ taskToWait = runningStreamTask;
+ if (taskToWait != null) {
+ // Linearize manual termination with callbacks before
publishing the new job status.
+ // Task-specific RPCs and waits remain outside the job
lock.
+ taskToWait.publishCancellation();
+ }
+ }
+ // RESUME must re-enter PENDING so the scheduler creates a
successor before publishing RUNNING.
+ super.updateJobStatus(publishedStatus);
+ statusEpoch++;
+ if (isFinalStatus()) {
+
Env.getCurrentGlobalTransactionMgr().getCallbackFactory().removeCallback(getJobId());
+ }
+ log.info("Streaming insert job {} manually updated status to {}",
getJobId(), getJobStatus());
} finally {
lock.writeLock().unlock();
}
+ boolean readerReleased = true;
+ if (taskToRelease != null) {
+ readerReleased = !(taskToRelease instanceof StreamingMultiTblTask)
+ || ((StreamingMultiTblTask)
taskToRelease).releaseRemoteReaderAndWait();
+ if (!(taskToRelease instanceof StreamingMultiTblTask)) {
+ taskToRelease.releaseRemoteReader();
+ }
+ }
+ if (taskToWait != null) {
+ taskToWait.cancel(true);
+ if (readerReleased && taskToWait.canHandoffAfterCancellation()) {
Review Comment:
[P1] Preserve the snapshot runtime owner for terminal cleanup
This clears the task after successful PAUSE handoff or an execution-terminal
STOP, but snapshot rounds may select an arbitrary runtime BE and DROP later
learns that exact `/api/close` destination only from `runningStreamTask`.
PAUSE's `/api/releaseReader` deliberately keeps the CDC slot, and STOP can
clear without either release or close; after this clear, `cleanMeta()` falls
back to the bound/random BE and can close the wrong host, leaving the live slot
behind. Retain a terminal cleanup owner until DROP consumes it or close through
the captured runtime BE, and test PAUSE/STOP-then-DROP with different runtime
and bound BEs.
##########
fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingMultiTblTask.java:
##########
@@ -390,6 +402,39 @@ public void releaseRemoteReader() {
}
}
+ /** Wait for the BE to acknowledge reader release before allowing a
successor to reuse the source. */
+ boolean releaseRemoteReaderAndWait() {
+ if (runningBackendId <= 0) {
Review Comment:
[P1] Linearize reader admission with PAUSE
This returns success when the worker has not published `runningBackendId`
yet, but that worker checks cancellation only before `resolveBackend()`; it can
publish the backend and send `/api/writeRecords` immediately after PAUSE
observes this branch. `cancel(true)` does not wait for this task, and the
manually paused job has no scheduler pass to retry predecessor release, so
PAUSE can complete while a new CDC reader starts and remains owned
indefinitely. Please serialize backend/reader admission with cancellation
(including a post-resolution cancel check) and add a barrier test for PAUSE
between the run check and backend publication.
##########
fe/fe-core/src/main/java/org/apache/doris/qe/ConnectProcessor.java:
##########
@@ -773,8 +777,17 @@ public TMasterOpResult proxyExecute(TMasterOpRequest
request) throws TException
LOG.warn("Process one query failed because unknown reason: ", e);
ctx.getState().setError(ErrorCode.ERR_UNKNOWN_ERROR, "Unexpected
exception: " + e.getMessage());
}
- // no matter the master execute success or fail, the master must
transfer the result to follower
- // and tell the follower the current journalID.
+ try {
+ return buildProxyResult(request, executor);
+ } finally {
+ if (ctx.getStatementContext() != null) {
+ ctx.getStatementContext().close();
Review Comment:
[P1] Do not let resource-close failure skip forwarded cleanup
`StatementContext.close()` can rethrow a resource's close failure from this
`finally`. That makes `proxyExecute()` exit before
`FrontendServiceImpl.forward()` reaches `ConnectContext.remove()` and the
proxy-query map callback, leaving the finished context registered for
cancellation and installed on the reusable Thrift worker. Please put
thread-local and identity-checked proxy-map removal in an outer `finally` while
preserving the close error, and add a forwarded execution test with a throwing
statement resource.
##########
fe/fe-core/src/main/java/org/apache/doris/qe/Coordinator.java:
##########
@@ -784,6 +785,9 @@ private boolean shouldQueue() {
// A call to Exec() must precede all other member function calls.
@Override
public void exec() throws Exception {
+ if (isQueryCancelled()) {
Review Comment:
[P1] Close the queue-token publication race with cancel
This check and the later `shouldQueue()` check can both pass, then
cancellation can run before `queueToken` is assigned. `cancel()` sees no token
and only marks the coordinator canceled; this thread subsequently publishes a
waiting token and blocks in `QueueToken.get()`, whose future does not observe
coordinator status, so the canceled query stays pinned until queue timeout.
Nereids has the same gap before `setQueueInfo()`. Linearize token publication
with cancellation or recheck-and-cancel immediately after publication, with
classic and Nereids barrier tests.
##########
fe/fe-core/src/main/java/org/apache/doris/service/FrontendServiceImpl.java:
##########
@@ -1214,6 +1227,10 @@ public TMasterOpResult forward(TMasterOpRequest params)
throws TException {
Runnable clearCallback = () -> {};
if (params.isSetQueryId()) {
proxyQueryIdToConnCtx.put(params.getQueryId(), context);
+ if (pendingProxyQueryCancels.remove(params.getQueryId()) != null) {
Review Comment:
[P1] Preserve cancellation across forwarded RPC retries
`FEOpExecutor.forward()` can retry the same request/query ID after a
transport failure, while the original attempt may also have reached this FE.
This one-shot `remove` lets the first arrival consume the pre-registration
cancel; a delayed/retried second arrival then registers and executes
uncanceled. Overlapping attempts also overwrite the single map value, and
either key-only cleanup can remove the other's publication. Please keep
cancellation durable for the whole retry-arrival window and make
registration/removal identity-safe, with a same-query-ID overlapping-attempt
test.
--
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]
