GV

GeezerVison

The World Wide Web, restored to 800 pixels of glory.

Home|About|Gaming cabinet|Archives|How it works|Starter sites

On this site

  • Sign in
  • ActionsCI/CD and automation
  • CodespacesCloud development environments
  • All Features
  • Documentation
  • Training
  • Certifications
  • Build with Copilot
  • Customize Copilot
  • Integrations
  • Trending repos
  • GitHub Sponsors
  • Developer articles
  • Enterprises

This view

  • Original page
  • New address

GeezerVison is showing github.com — [Feature] Support alter task command (#68675) · APIJSON/starrocks@061d18e

File tree

Expand file treeCollapse file tree
Open diff view settings
Filter options
  • fe
    • fe-core/src
      • main/java/com/starrocks
        • persist
          • AlterTaskInfo.java
        • qe
          • DDLStmtExecutor.java
          • RedirectStatus.java
        • scheduler
          • TaskManager.java
        • sql
          • analyzer
            • Analyzer.java
            • TaskAnalyzer.java
          • ast
            • AstTraverser.java
            • AstVisitorExtendInterface.java
          • parser
            • AstBuilder.java
      • test/java/com/starrocks
        • scheduler
          • TaskManagerEditLogTest.java
        • sql/analyzer
          • AnalyzeSingleTest.java
    • fe-grammar/src/main/antlr/com/starrocks/grammar
      • StarRocks.g4
    • fe-parser/src/main/java/com/starrocks/sql/ast
      • AlterTaskStmt.java
      • AstVisitor.java
  • test/sql/test_task
    • R
      • test_alter_task
    • T
      • test_alter_task
Expand file treeCollapse file tree
Open diff view settings
Collapse file

‎fe/fe-core/src/main/java/com/starrocks/persist/AlterTaskInfo.java‎

Copy file name to clipboardExpand all lines: fe/fe-core/src/main/java/com/starrocks/persist/AlterTaskInfo.java
+26Lines changed: 26 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,8 @@
1818
import com.starrocks.scheduler.Constants;
1919
import com.starrocks.scheduler.persist.TaskSchedule;
2020

21+
import java.util.Map;
22+
2123
public class AlterTaskInfo implements Writable {
2224
@SerializedName("name")
2325
private String name;
@@ -28,6 +30,12 @@ public class AlterTaskInfo implements Writable {
2830
@SerializedName("schedule")
2931
private TaskSchedule schedule;
3032

33+
@SerializedName("state")
34+
private Constants.TaskState state;
35+
36+
@SerializedName("properties")
37+
private Map<String, String> properties;
38+
3139
public AlterTaskInfo() {
3240
// for persist
3341
}
@@ -38,6 +46,16 @@ public AlterTaskInfo(String name, Constants.TaskType type, TaskSchedule schedule
3846
this.schedule = schedule;
3947
}
4048

49+
public AlterTaskInfo(String name, Constants.TaskState state) {
50+
this.name = name;
51+
this.state = state;
52+
}
53+
54+
public AlterTaskInfo(String name, Map<String, String> properties) {
55+
this.name = name;
56+
this.properties = properties;
57+
}
58+
4159
public String getName() {
4260
return name;
4361
}
@@ -49,4 +67,12 @@ public Constants.TaskType getType() {
4967
public TaskSchedule getSchedule() {
5068
return schedule;
5169
}
70+
71+
public Constants.TaskState getState() {
72+
return state;
73+
}
74+
75+
public Map<String, String> getProperties() {
76+
return properties;
77+
}
5278
}
Collapse file

‎fe/fe-core/src/main/java/com/starrocks/qe/DDLStmtExecutor.java‎

Copy file name to clipboardExpand all lines: fe/fe-core/src/main/java/com/starrocks/qe/DDLStmtExecutor.java
+28Lines changed: 28 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -68,6 +68,7 @@
6868
import com.starrocks.sql.ast.AlterStorageVolumeStmt;
6969
import com.starrocks.sql.ast.AlterSystemStmt;
7070
import com.starrocks.sql.ast.AlterTableStmt;
71+
import com.starrocks.sql.ast.AlterTaskStmt;
7172
import com.starrocks.sql.ast.AlterUserStmt;
7273
import com.starrocks.sql.ast.AlterViewStmt;
7374
import com.starrocks.sql.ast.AstVisitorExtendInterface;
@@ -1155,6 +1156,33 @@ public ShowResultSet visitDropTaskStmt(DropTaskStmt dropTaskStmt, ConnectContext
11551156
return null;
11561157
}
11571158

1159+
@Override
1160+
public ShowResultSet visitAlterTaskStatement(AlterTaskStmt alterTaskStmt, ConnectContext context) {
1161+
TaskManager taskManager = context.getGlobalStateMgr().getTaskManager();
1162+
String taskName = alterTaskStmt.getTaskName().getName();
1163+
if (!taskManager.containTask(taskName)) {
1164+
if (alterTaskStmt.isIfExists()) {
1165+
return null;
1166+
}
1167+
throw new SemanticException("Task " + taskName + " is not exist");
1168+
}
1169+
Task task = taskManager.getTask(taskName);
1170+
switch (alterTaskStmt.getAction()) {
1171+
case RESUME:
1172+
taskManager.resumeTask(task);
1173+
break;
1174+
case SUSPEND:
1175+
taskManager.suspendTask(task);
1176+
break;
1177+
case SET:
1178+
taskManager.updateTaskProperties(task, alterTaskStmt.getProperties());
1179+
break;
1180+
default:
1181+
throw new SemanticException("Unsupported alter task action: " + alterTaskStmt.getAction());
1182+
}
1183+
return null;
1184+
}
1185+
11581186
@Override
11591187
public ShowResultSet visitCreateStorageVolumeStatement(CreateStorageVolumeStmt stmt, ConnectContext context) {
11601188
ErrorReport.wrapWithRuntimeException(() -> {
Collapse file

‎fe/fe-core/src/main/java/com/starrocks/qe/RedirectStatus.java‎

Copy file name to clipboardExpand all lines: fe/fe-core/src/main/java/com/starrocks/qe/RedirectStatus.java
+6Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -47,6 +47,7 @@
4747
import com.starrocks.sql.ast.AlterStorageVolumeStmt;
4848
import com.starrocks.sql.ast.AlterSystemStmt;
4949
import com.starrocks.sql.ast.AlterTableStmt;
50+
import com.starrocks.sql.ast.AlterTaskStmt;
5051
import com.starrocks.sql.ast.AlterUserStmt;
5152
import com.starrocks.sql.ast.AlterViewStmt;
5253
import com.starrocks.sql.ast.AnalyzeProfileStmt;
@@ -539,6 +540,11 @@ public RedirectStatus visitDropTaskStmt(DropTaskStmt statement, Void context) {
539540
return visitDDLStatement(statement, context);
540541
}
541542

543+
@Override
544+
public RedirectStatus visitAlterTaskStatement(AlterTaskStmt statement, Void context) {
545+
return visitDDLStatement(statement, context);
546+
}
547+
542548
// ---------------------------------------- Partition Statement ----------------------------------------------------
543549

544550
@Override
Collapse file

‎fe/fe-core/src/main/java/com/starrocks/scheduler/TaskManager.java‎

Copy file name to clipboardExpand all lines: fe/fe-core/src/main/java/com/starrocks/scheduler/TaskManager.java
+93-4Lines changed: 93 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -464,9 +464,23 @@ public SubmitResult executeTaskAsync(Task task, ExecuteOption option) {
464464

465465
/**
466466
* Suspend a task, which will stop the task scheduler and kill the running task run.
467-
* NOTE: this method is thread safe, no need to task lock again.
467+
* NOTE: this method will write edit log.
468468
*/
469469
public void suspendTask(Task task) {
470+
if (task == null) {
471+
return;
472+
}
473+
GlobalStateMgr.getCurrentState().getEditLog().logAlterTask(
474+
new AlterTaskInfo(task.getName(), Constants.TaskState.PAUSE),
475+
wal -> task.setState(Constants.TaskState.PAUSE)
476+
);
477+
suspendTaskInternal(task);
478+
}
479+
480+
/**
481+
* Internal method to suspend a task without writing edit log.
482+
*/
483+
private void suspendTaskInternal(Task task) {
470484
if (task == null) {
471485
return;
472486
}
@@ -496,9 +510,23 @@ public void suspendTask(Task task) {
496510

497511
/**
498512
* Resume a task, which will restart the task scheduler if the task is periodical.
499-
* NOTE: This method is thread safe, no need to task lock again.
513+
* NOTE: this method will write edit log.
500514
*/
501515
public void resumeTask(Task task) {
516+
if (task == null) {
517+
return;
518+
}
519+
GlobalStateMgr.getCurrentState().getEditLog().logAlterTask(
520+
new AlterTaskInfo(task.getName(), Constants.TaskState.ACTIVE),
521+
wal -> task.setState(Constants.TaskState.ACTIVE)
522+
);
523+
resumeTaskInternal(task);
524+
}
525+
526+
/**
527+
* Internal method to resume a task without writing edit log.
528+
*/
529+
private void resumeTaskInternal(Task task) {
502530
if (task == null) {
503531
return;
504532
}
@@ -520,6 +548,50 @@ public void resumeTask(Task task) {
520548
}
521549
}
522550

551+
/**
552+
* Update task properties for subsequent executions.
553+
* NOTE: this method will write edit log.
554+
*/
555+
public void updateTaskProperties(Task task, Map<String, String> properties) {
556+
if (task == null || properties == null || properties.isEmpty()) {
557+
return;
558+
}
559+
GlobalStateMgr.getCurrentState().getEditLog().logAlterTask(
560+
new AlterTaskInfo(task.getName(), properties),
561+
wal -> {
562+
Map<String, String> current = task.getProperties();
563+
if (current == null) {
564+
current = Maps.newHashMap();
565+
task.setProperties(current);
566+
}
567+
current.putAll(properties);
568+
}
569+
);
570+
updateTaskPropertiesInternal(task, properties);
571+
}
572+
573+
/**
574+
* Internal method to update task properties without writing edit log.
575+
*/
576+
private void updateTaskPropertiesInternal(Task task, Map<String, String> properties) {
577+
if (task == null || properties == null || properties.isEmpty()) {
578+
return;
579+
}
580+
if (!tryTaskLock()) {
581+
throw new RuntimeException("Failed to get task lock when update Task properties[" + task.getName() + "]");
582+
}
583+
try {
584+
Map<String, String> current = task.getProperties();
585+
if (current == null) {
586+
current = Maps.newHashMap();
587+
task.setProperties(current);
588+
}
589+
current.putAll(properties);
590+
} finally {
591+
taskUnlock();
592+
}
593+
}
594+
523595
public void dropTasks(List<Long> taskIdList) {
524596
takeTaskLock();
525597
try {
@@ -647,7 +719,23 @@ private void changeTask(Task currentTask, Constants.TaskType changedType, TaskSc
647719

648720
public void replayAlterTask(AlterTaskInfo alterTaskInfo) {
649721
Task currentTask = getTask(alterTaskInfo.getName());
650-
changeTask(currentTask, alterTaskInfo.getType(), alterTaskInfo.getSchedule());
722+
// Handle task type change (existing logic)
723+
if (alterTaskInfo.getType() != null) {
724+
changeTask(currentTask, alterTaskInfo.getType(), alterTaskInfo.getSchedule());
725+
}
726+
// Handle task state change (suspend/resume)
727+
if (alterTaskInfo.getState() != null) {
728+
currentTask.setState(alterTaskInfo.getState());
729+
}
730+
// Handle properties update
731+
if (alterTaskInfo.getProperties() != null) {
732+
Map<String, String> current = currentTask.getProperties();
733+
if (current == null) {
734+
current = Maps.newHashMap();
735+
currentTask.setProperties(current);
736+
}
737+
current.putAll(alterTaskInfo.getProperties());
738+
}
651739
}
652740

653741
@VisibleForTesting
@@ -718,12 +806,13 @@ void registerScheduler(Task task) {
718806
task.getName(), initialDelay, periodSeconds, taskStartTime, currentDateTime);
719807
// set task's next schedule time
720808
task.setNextScheduleTime(currentDateTime.plusSeconds(initialDelay).toEpochSecond(ZoneOffset.UTC));
721-
ExecuteOption option = new ExecuteOption(Constants.TaskRunPriority.LOWEST.value(), true, task.getProperties());
722809
ScheduledFuture<?> future = periodScheduler.scheduleAtFixedRate(() -> {
723810
// ensure an execute task will not throw exception
724811
try {
725812
task.setLastScheduleTime(TimeUtils.getEpochSeconds());
726813
task.setNextScheduleTime(TimeUtils.getEpochSeconds() + periodSeconds);
814+
ExecuteOption option = new ExecuteOption(Constants.TaskRunPriority.LOWEST.value(), true,
815+
task.getProperties());
727816
executeTask(task.getName(), option);
728817
} catch (Throwable e) {
729818
LOG.warn("failed to execute periodical task: {}", task, e);
Collapse file

‎fe/fe-core/src/main/java/com/starrocks/sql/analyzer/Analyzer.java‎

Copy file name to clipboardExpand all lines: fe/fe-core/src/main/java/com/starrocks/sql/analyzer/Analyzer.java
+7Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -43,6 +43,7 @@
4343
import com.starrocks.sql.ast.AlterStorageVolumeStmt;
4444
import com.starrocks.sql.ast.AlterSystemStmt;
4545
import com.starrocks.sql.ast.AlterTableStmt;
46+
import com.starrocks.sql.ast.AlterTaskStmt;
4647
import com.starrocks.sql.ast.AlterViewStmt;
4748
import com.starrocks.sql.ast.AnalyzeStmt;
4849
import com.starrocks.sql.ast.AstVisitorExtendInterface;
@@ -362,6 +363,12 @@ public Void visitSubmitTaskStatement(SubmitTaskStmt statement, ConnectContext co
362363
return null;
363364
}
364365

366+
@Override
367+
public Void visitAlterTaskStatement(AlterTaskStmt statement, ConnectContext context) {
368+
TaskAnalyzer.analyzeAlterTaskStmt(statement);
369+
return null;
370+
}
371+
365372
public static void analyzeSubmitTask(SubmitTaskStmt statement, ConnectContext context) {
366373
StatementBase workhouse = analyzeSubmitTaskWorkhorse(statement, context);
367374
analyzeSubmitTaskOnly(workhouse, statement, context);
Collapse file

‎fe/fe-core/src/main/java/com/starrocks/sql/analyzer/TaskAnalyzer.java‎

Copy file name to clipboardExpand all lines: fe/fe-core/src/main/java/com/starrocks/sql/analyzer/TaskAnalyzer.java
+7Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,7 @@
2323
import com.starrocks.qe.ConnectContext;
2424
import com.starrocks.qe.SessionVariable;
2525
import com.starrocks.scheduler.persist.TaskSchedule;
26+
import com.starrocks.sql.ast.AlterTaskStmt;
2627
import com.starrocks.sql.ast.SubmitTaskStmt;
2728
import org.apache.commons.collections.MapUtils;
2829

@@ -49,6 +50,12 @@ public static void analyzeSubmitTaskStmt(SubmitTaskStmt submitTaskStmt, ConnectC
4950
analyzeTaskSchedule(submitTaskStmt.getSchedule());
5051
}
5152

53+
public static void analyzeAlterTaskStmt(AlterTaskStmt alterTaskStmt) {
54+
if (alterTaskStmt.getAction() == AlterTaskStmt.AlterAction.SET) {
55+
analyzeTaskProperties(alterTaskStmt.getProperties());
56+
}
57+
}
58+
5259
private static void analyzeTaskSchedule(TaskSchedule schedule) {
5360
if (schedule == null) {
5461
return;
Collapse file

‎fe/fe-core/src/main/java/com/starrocks/sql/ast/AstTraverser.java‎

Copy file name to clipboardExpand all lines: fe/fe-core/src/main/java/com/starrocks/sql/ast/AstTraverser.java
+5Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -67,6 +67,11 @@ public R visitSubmitTaskStatement(SubmitTaskStmt statement, C context) {
6767
return null;
6868
}
6969

70+
@Override
71+
public R visitAlterTaskStatement(AlterTaskStmt statement, C context) {
72+
return null;
73+
}
74+
7075
@Override
7176
public R visitCreatePipeStatement(CreatePipeStmt statement, C context) {
7277
if (statement.getInsertStmt() != null) {
Collapse file

‎fe/fe-core/src/main/java/com/starrocks/sql/ast/AstVisitorExtendInterface.java‎

Copy file name to clipboardExpand all lines: fe/fe-core/src/main/java/com/starrocks/sql/ast/AstVisitorExtendInterface.java
+4Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -161,6 +161,10 @@ default R visitSubmitTaskStatement(SubmitTaskStmt statement, C context) {
161161
return visitDDLStatement(statement, context);
162162
}
163163

164+
default R visitAlterTaskStatement(AlterTaskStmt statement, C context) {
165+
return visitDDLStatement(statement, context);
166+
}
167+
164168
// ---------------------------------------- Partition Statement ----------------------------------------------------
165169

166170
default R visitRecoverPartitionStatement(RecoverPartitionStmt statement, C context) {
Collapse file

‎fe/fe-core/src/main/java/com/starrocks/sql/parser/AstBuilder.java‎

Copy file name to clipboardExpand all lines: fe/fe-core/src/main/java/com/starrocks/sql/parser/AstBuilder.java
+26Lines changed: 26 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -100,6 +100,7 @@
100100
import com.starrocks.sql.ast.AlterTableModifyDefaultBucketsClause;
101101
import com.starrocks.sql.ast.AlterTableOperationClause;
102102
import com.starrocks.sql.ast.AlterTableStmt;
103+
import com.starrocks.sql.ast.AlterTaskStmt;
103104
import com.starrocks.sql.ast.AlterUserStmt;
104105
import com.starrocks.sql.ast.AlterViewClause;
105106
import com.starrocks.sql.ast.AlterViewStmt;
@@ -2118,6 +2119,31 @@ public ParseNode visitSubmitTaskStatement(com.starrocks.sql.parser.StarRocksPars
21182119
return res;
21192120
}
21202121

2122+
@Override
2123+
public ParseNode visitAlterTaskStatement(com.starrocks.sql.parser.StarRocksParser.AlterTaskStatementContext context) {
2124+
QualifiedName qualifiedName = getQualifiedName(context.qualifiedName());
2125+
TaskName taskName = qualifiedNameToTaskName(qualifiedName);
2126+
boolean ifExists = context.IF() != null;
2127+
2128+
AlterTaskStmt.AlterAction action;
2129+
Map<String, String> properties = null;
2130+
if (context.RESUME() != null) {
2131+
action = AlterTaskStmt.AlterAction.RESUME;
2132+
} else if (context.SUSPEND() != null) {
2133+
action = AlterTaskStmt.AlterAction.SUSPEND;
2134+
} else {
2135+
action = AlterTaskStmt.AlterAction.SET;
2136+
// get properties
2137+
List<com.starrocks.sql.parser.StarRocksParser.PropertyContext> propertyContextList = null;
2138+
com.starrocks.sql.parser.StarRocksParser.PropertyListContext propertyListContext = context.propertyList();
2139+
if (propertyListContext != null) {
2140+
propertyContextList = propertyListContext.property();
2141+
}
2142+
properties = getProperties(propertyContextList, true);
2143+
}
2144+
return new AlterTaskStmt(taskName, ifExists, action, properties, createPos(context));
2145+
}
2146+
21212147
@Override
21222148
public ParseNode visitDropTaskStatement(com.starrocks.sql.parser.StarRocksParser.DropTaskStatementContext context) {
21232149
QualifiedName qualifiedName = getQualifiedName(context.qualifiedName());

0 commit comments

Comments
0 (0)
You are visitor 01402520 — GeezerVison does not store the pages you look up.
Best viewed with Netscape Navigator 4.7 or Internet Explorer 6 at 800×600. No cookies required.