github-actions[bot] commented on code in PR #66913:
URL: https://github.com/apache/doris/pull/66913#discussion_r3835811082
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergExternalMetaCache.java:
##########
@@ -105,12 +113,49 @@ 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);
Review Comment:
[P1] Bound no-context Iceberg table loads
When no `StatementContext` exists, this returns a raw `loadTable()` result
without a native/HMS load guard, lease, or `tableCleanup()`. With partition
analysis enabled, `StatisticsAutoCollector` reaches this path via
`isPartitionedTable()`, so repeated probes can leak table-owned FileIO for
Glue/S3 Tables/qualifying REST tables and race runtime retirement. Use a
bounded `withIcebergTable()` projection or another exact-generation operation
owner.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/source/IcebergScanNode.java:
##########
@@ -782,7 +785,7 @@ public TableScan createTableScan() throws UserException {
this.pushdownIcebergPredicates.add(predicate.toString());
}
- icebergTableScan =
scan.planWith(source.getCatalog().getThreadPoolWithPreAuth());
+ icebergTableScan = scan.planWith(planningExecutor);
Review Comment:
[P1] Retain the Iceberg generation for the async planner
`doStartSplit()` launches this planning work on an unrecorded future, but
the exact Table/FileIO/catalog/executor lease remains owned only by
`StatementContext`. On cancellation or timeout the statement can close while
the task remains blocked in `planFiles()`, after which refresh/reset may retire
G1 underneath it. Give the future an independent exact-generation owner
released from its actual-terminal callback and register a cancellation handle
with the split assignment.
##########
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] Keep credentials with the retained Iceberg generation
This selects the exact executor from the retained G1 table, but
`doInitialize()` immediately reads the authenticator and both storage-property
maps from the mutable live catalog. ALTER/reset can therefore assemble a G1
table/executor with G2 credentials/properties (or a temporarily null
authenticator), and retained-table sink paths repeat the same lookup. Carry an
immutable authenticator/property bundle in the same generation lease.
##########
fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/AbstractStreamingTask.java:
##########
@@ -94,35 +98,76 @@ public long getRunningBackendId() {
}
public void execute() throws JobException {
- while (retryCount <= MAX_RETRY) {
- try {
- before();
- run();
- onSuccess();
- return;
- } catch (Exception e) {
+ synchronized (executionCompletion) {
+ executionStarted = true;
+ executionOwner = Thread.currentThread();
+ }
+ try {
+ while (retryCount <= MAX_RETRY) {
+ Exception attemptFailure = null;
+ try {
+ before();
+ run();
+ } catch (Exception e) {
+ attemptFailure = e;
+ } 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.
+ try {
+ closeOrReleaseResources();
+ } catch (RuntimeException cleanupFailure) {
+ if (attemptFailure == null) {
+ attemptFailure = cleanupFailure;
+ } else {
+ attemptFailure.addSuppressed(cleanupFailure);
+ }
+ }
+ }
+ if (attemptFailure == null) {
+ onSuccess();
Review Comment:
[P1] Handle failures from the success callback
`onSuccess()` is still outside the attempt-failure block.
`StreamingInsertTask` publishes `SUCCESS` before job-level successor creation
can throw, while `scheduleOneTask()` only logs the escaped exception and
removes the old task; the RUNNING job can then have no successor or failure
transition. Make success/successor publication explicit and recover without
retrying an already-committed insert.
##########
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:
[P1] Publish Hudi batch errors immediately
This only stores the first batch exception; `SplitAssignment.setException()`
now runs from `finishBatchSplit()`, gated on every accepted sibling becoming
terminal. If one partition fails while another filesystem call remains blocked,
initial planning/fetchers see timeouts instead of the known error. Publish the
first error to `SplitAssignment` immediately while keeping independent
last-task accounting for exact lease release.
##########
fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJob.java:
##########
@@ -556,19 +556,33 @@ 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 {
- super.updateJobStatus(status);
- if (JobStatus.PAUSED.equals(getJobStatus())) {
- clearRunningStreamTask(status);
+ if ((JobStatus.PAUSED.equals(status) ||
JobStatus.STOPPED.equals(status))
+ && status != getJobStatus()) {
+ taskToCancel = runningStreamTask;
Review Comment:
[P1] Preserve the manual CDC reader handoff
The manual command calls `onManualStatusAltered()` only after this status
transition, but the transition has already removed the task. The hook therefore
neither sets `needRebuildReader` nor calls `releaseRemoteReader()`, while
`StreamingMultiTblTask.cancel()` intentionally leaves both actions to that
hook. Carry the captured task into the manual handoff or perform the
rebuild/release marking before discarding it.
##########
fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/AbstractStreamingTask.java:
##########
@@ -94,35 +98,76 @@ public long getRunningBackendId() {
}
public void execute() throws JobException {
- while (retryCount <= MAX_RETRY) {
- try {
- before();
- run();
- onSuccess();
- return;
- } catch (Exception e) {
+ synchronized (executionCompletion) {
+ executionStarted = true;
+ executionOwner = Thread.currentThread();
+ }
+ try {
+ while (retryCount <= MAX_RETRY) {
+ Exception attemptFailure = null;
+ try {
+ before();
+ run();
+ } catch (Exception e) {
+ attemptFailure = e;
+ } 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.
+ try {
+ closeOrReleaseResources();
+ } catch (RuntimeException cleanupFailure) {
Review Comment:
[P1] Do not retry a completed insert for cleanup failure
This turns a standalone cleanup exception into an ordinary attempt failure.
For `StreamingInsertTask`, `run()` has already completed the insert and
production leaves `noRetry` false, so a post-commit unregister or
statement-close failure re-enters `before()`/`run()` instead of calling
`onSuccess()`. Even if the reused label prevents duplicate rows, the committed
batch can be reported failed/paused and its transaction callback lock spans the
retry. Track execution and cleanup outcomes separately; cleanup-only failure
must not replay a completed insert.
##########
fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJob.java:
##########
@@ -556,19 +556,33 @@ 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 {
- super.updateJobStatus(status);
- if (JobStatus.PAUSED.equals(getJobStatus())) {
- clearRunningStreamTask(status);
+ if ((JobStatus.PAUSED.equals(status) ||
JobStatus.STOPPED.equals(status))
+ && status != getJobStatus()) {
+ taskToCancel = runningStreamTask;
+ runningStreamTask = null;
}
+ super.updateJobStatus(status);
Review Comment:
[P2] Preserve canceled-task accounting
At this point `runningStreamTask` has already been nulled, so
`super.updateJobStatus()` dynamically calls `cancelAllTasks()` and the override
returns before its active-state check and `canceledTaskCount` increment. The
later direct cancellation does not update the counter, making active PAUSE/STOP
transitions disappear from job statistics. Snapshot the captured task's active
state and increment exactly once.
##########
fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJob.java:
##########
@@ -556,19 +556,33 @@ 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 {
- super.updateJobStatus(status);
- if (JobStatus.PAUSED.equals(getJobStatus())) {
- clearRunningStreamTask(status);
+ if ((JobStatus.PAUSED.equals(status) ||
JobStatus.STOPPED.equals(status))
+ && status != getJobStatus()) {
+ taskToCancel = runningStreamTask;
+ runningStreamTask = null;
Review Comment:
[P1] Preserve the task through transaction callbacks
Clearing `runningStreamTask` here leaves a still-running insert with no task
identity for transaction callbacks. `beforeCommitted()` later takes the same
lock and unconditionally dereferences this field for cancellation, task ID,
offset, and backend data; it remains null even after the outside cancellation
returns. Keep a stable captured task through callback terminality or add a
job-status/null-safe transaction owner.
##########
fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertTask.java:
##########
@@ -173,24 +174,57 @@ 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] Make pre-executor cancellation bounded
`before()` can block in remote S3 listing or plan initialization before
`stmtExecutor` is assigned. In that interval `cancel(true)` has nothing to
signal and waits without timeout or owner interruption for `executionFinished`,
so PAUSE can hang indefinitely. Publish an interruptible owner earlier or make
this wait bounded/asynchronous, and cover the pre-publication barrier in a test.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/hive/source/HiveScanNode.java:
##########
@@ -142,6 +145,13 @@ protected void doInitialize() throws UserException {
Env.getCurrentHiveTransactionMgr().register(hiveTransaction);
skipCheckingAcidVersionFile =
sessionVariable.skipCheckingAcidVersionFile;
}
+ ensureHmsRuntimeGeneration();
Review Comment:
[P1] Clean up Hive transactions when the fence aborts planning
The Hive transaction is begun and registered before this new generation
check. If the check throws, Nereids planning returns an error before
`handleQueryWithRetry()` or QeProcessor registration, and ordinary MySQL
cleanup only closes `StatementContext`, leaving the local map entry and remote
read transaction open. Register the transaction with a planning/statement owner
immediately or deregister every post-registration initialization failure, using
the exact runtime needed for cleanup.
##########
fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJob.java:
##########
@@ -556,19 +556,33 @@ 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 {
- super.updateJobStatus(status);
- if (JobStatus.PAUSED.equals(getJobStatus())) {
- clearRunningStreamTask(status);
+ if ((JobStatus.PAUSED.equals(status) ||
JobStatus.STOPPED.equals(status))
+ && status != getJobStatus()) {
+ taskToCancel = runningStreamTask;
+ runningStreamTask = null;
}
+ super.updateJobStatus(status);
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) {
+ // 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);
Review Comment:
[P1] Make PAUSE and its reason scheduler-atomic
The method unlocks with `PAUSED` visible and can then wait in
`cancel(true)`, while manual alteration installs `MANUAL_PAUSE_ERR` only after
this method returns. A scheduler tick can observe the old resumable reason,
call `autoResumeHandler()`, and move the job to `PENDING`; the later hook
changes only the reason and never restores `PAUSED`. Gate auto-resume through
transition completion or publish reason/status atomically; `fetchMeta()` needs
the same ordering fix.
--
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]
