github-actions[bot] commented on code in PR #70282:
URL: https://github.com/apache/doris/pull/70282#discussion_r4480623182
##########
fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/InsertTask.java:
##########
@@ -180,17 +184,30 @@ public void before() throws JobException {
ctx.setStatementContext(statementContext);
if (StringUtils.isNotEmpty(sql)) {
NereidsParser parser = new NereidsParser();
- this.command = (InsertIntoTableCommand) parser.parseSingle(sql);
- this.command.setLabelName(Optional.of(getJobId() + LABEL_SPLITTER
+ getTaskId()));
- this.command.setJobId(getTaskId());
+ LogicalPlan parsed = parser.parseSingle(sql);
+ if (parsed instanceof InsertIntoTableCommand) {
+ InsertIntoTableCommand tableCommand = (InsertIntoTableCommand)
parsed;
+ tableCommand.setLabelName(Optional.of(getJobId() +
LABEL_SPLITTER + getTaskId()));
+ tableCommand.setJobId(getTaskId());
+ command = tableCommand;
+ } else if (parsed instanceof InsertIntoTVFCommand) {
+ try {
+ ((InsertIntoTVFCommand) parsed).validateJobTarget();
+ } catch (Exception e) {
+ throw new JobException(e.getMessage());
+ }
+ command = (InsertIntoTVFCommand) parsed;
Review Comment:
[P1] Bind the task executor before planning INSERT INTO TVF jobs. The parsed
StmtExecutor constructor does not call ctx.setExecutor(this), so
FileQueryScanNode skips SplitAssignment cancellation and
ConnectorSessionBuilder skips QueryTaskGroups::close. If planning is canceled
before activeCoordinator is published, InsertIntoTVFCommand has no coordinator
to cancel and its registered-query cleanup is not reached; planner resource
release does not close these scan resources. Set the executor on the context
and close the statement and scan scope on planning cancellation.
##########
fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/InsertTask.java:
##########
@@ -204,50 +221,107 @@ protected void closeOrReleaseResources() {
@Override
public void run() throws JobException {
- try {
+ // Cancellation releases the task's fields. The execution thread owns
these
+ // local handles until run() has exited, including after cancellation.
+ final Command runningCommand;
+ final ConnectContext runningContext;
+ final StmtExecutor runningExecutor;
+ synchronized (this) {
if (isCanceled.get()) {
- log.info("task has been canceled, task id is {}", getTaskId());
return;
}
- command.runWithUpdateInfo(ctx, stmtExecutor, loadStatistic);
- if (ctx.getState().getStateType() != QueryState.MysqlStateType.OK)
{
- throw new JobException(ctx.getState().getErrorMessage());
+ runningCommand = command;
+ runningContext = ctx;
+ runningExecutor = stmtExecutor;
+ }
+ try {
+ if (runningCommand instanceof InsertIntoTableCommand) {
+ ((InsertIntoTableCommand) runningCommand).runWithUpdateInfo(
+ runningContext, runningExecutor, loadStatistic);
+ } else if (runningCommand instanceof InsertIntoTVFCommand) {
Review Comment:
[P1] Preserve the worker ConnectContext for TaskProcessor cleanup. Insert
tasks run through TaskProcessor.runTask(), whose finally reads
ConnectContext.get(), unregisters the query, and closes StatementContext and
its connector scope. This new finally removes the thread-local first, so
closeTaskContext sees null and skips that cleanup on every success, failure, or
cancellation. Close the statement context here or leave context removal to the
outer TaskProcessor hook.
--
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]
