[RoutineLoad] Support modify routine load job

docs/.vuepress/sidebar/en.js

@@ -405,6 +405,7 @@ module.exports = [
               "LOAD",
               "MINI LOAD",
               "MULTI LOAD",
+              "alter-routine-load",
               "PAUSE ROUTINE LOAD",
               "RESTORE TABLET",
               "RESUME ROUTINE LOAD",

docs/.vuepress/sidebar/zh-CN.js

@@ -418,6 +418,7 @@ module.exports = [
               "LOAD",
               "MINI LOAD",
               "MULTI LOAD",
+              "alter-routine-load",
               "PAUSE ROUTINE LOAD",
               "RESUME ROUTINE LOAD",
               "ROUTINE LOAD",

docs/en/administrator-guide/load-data/routine-load-manual.md

@@ -237,6 +237,10 @@ Specific commands and examples for viewing the **Task** status can be viewed wit
 
 You can only view tasks that are currently running, and tasks that have ended and are not started cannot be viewed.
 
+### Alter job
+
+Users can modify jobs that have been created. Specific instructions can be viewed through the `HELP ALTER ROUTINE LOAD;` command. Or refer to [ALTER ROUTINE LOAD](../../sql-reference/sql-statements/Data Manipulation/alter-routine-load.md).
+
 ### Job Control
 
 The user can control the stop, pause and restart of the job by the three commands `STOP/PAUSE/RESUME`. You can view help and examples with the three commands `HELP STOP ROUTINE LOAD;`, `HELP PAUSE ROUTINE LOAD;` and `HELP RESUME ROUTINE LOAD;`.

docs/en/sql-reference/sql-statements/Data Manipulation/alter-routine-load.md

@@ -0,0 +1,111 @@
+---
+{
+    "title": "ALTER ROUTINE LOAD",
+    "language": "en"
+}
+---
+
+<!-- 
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements.  See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership.  The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License.  You may obtain a copy of the License at
+
+  http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied.  See the License for the
+specific language governing permissions and limitations
+under the License.
+-->
+
+# ALTER ROUTINE LOAD
+## description
+
+This syntax is used to modify a routine import job that has been created.
+
+Only jobs in the PAUSED state can be modified.
+
+Syntax:
+
+    ALTER ROUTINE LOAD FOR [db.]job_name
+    [job_properties]
+    FROM data_source
+    [data_source_properties]
+
+1. `[db.]job_name`
+
+    Specify the name of the job to be modified.
+
+2. `job_properties`
+
+    Specify the job parameters that need to be modified. Currently only supports the modification of the following parameters:
+
+    1. `desired_concurrent_number`
+    2. `max_error_number`
+    3. `max_batch_interval`
+    4. `max_batch_rows`
+    5. `max_batch_size`
+    6. `jsonpaths`
+    7. `json_root`
+    8. `strip_outer_array`
+    9. `strict_mode`
+    10. `timezone`
+
+
+3. `data_source`
+
+    The type of data source. Currently supported:
+
+    KAFKA
+
+4. `data_source_properties`
+
+    The relevant attributes of the data source. Currently only supports:
+
+    1. `kafka_partitions`
+    2. `kafka_offsets`
+    3. Custom property, such as `property.group.id`
+
+    Notice：
+
+    1. `kafka_partitions` and `kafka_offsets` are used to modify the offset of the kafka partition to be consumed, and can only modify the currently consumed partition. Cannot add partition.
+
+## example
+
+1. Modify `desired_concurrent_number` to 1
+
+    ```
+    ALTER ROUTINE LOAD FOR db1.label1
+    PROPERTIES
+    (
+        "desired_concurrent_number" = "1"
+    );
+    ```
+
+2.  Modify `desired_concurrent_number` to 10, modify partition offset, and modify group id。
+
+    ```
+    ALTER ROUTINE LOAD FOR db1.label1
+    PROPERTIES
+    (
+        "desired_concurrent_number" = "10"
+    )
+    FROM kafka
+    (
+        "kafka_partitions" = "0, 1, 2",
+        "kafka_offsets" = "100, 200, 100",
+        "property.group.id" = "new_group"
+    );
+    ```
+
+
+## keyword
+
+    ALTER,ROUTINE,LOAD
+

docs/zh-CN/administrator-guide/load-data/routine-load-manual.md

@@ -238,6 +238,10 @@ FE 中的 JobScheduler 根据汇报结果，继续生成后续新的 Task，或
 
 只能查看当前正在运行中的任务，已结束和未开始的任务无法查看。
 
+### 修改作业属性
+
+用户可以修改已经创建的作业。具体说明可以通过 `HELP ALTER ROUTINE LOAD;` 命令查看。或参阅 [ALTER ROUTINE LOAD](../../sql-reference/sql-statements/Data Manipulation/alter-routine-load.md)。
+
 ### 作业控制
 
 用户可以通过 `STOP/PAUSE/RESUME` 三个命令来控制作业的停止，暂停和重启。可以通过 `HELP STOP ROUTINE LOAD;`, `HELP PAUSE ROUTINE LOAD;` 以及 `HELP RESUME ROUTINE LOAD;` 三个命令查看帮助和示例。

docs/zh-CN/sql-reference/sql-statements/Data Manipulation/alter-routine-load.md

@@ -0,0 +1,115 @@
+---
+{
+    "title": "ALTER ROUTINE LOAD",
+    "language": "zh-CN"
+}
+---
+
+<!-- 
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements.  See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership.  The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License.  You may obtain a copy of the License at
+
+  http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied.  See the License for the
+specific language governing permissions and limitations
+under the License.
+-->
+
+# ALTER ROUTINE LOAD
+## description
+
+该语法用于修改已经创建的例行导入作业。
+
+只能修改处于 PAUSED 状态的作业。
+
+语法：
+
+    ALTER ROUTINE LOAD FOR [db.]job_name
+    [job_properties]
+    FROM data_source
+    [data_source_properties]
+
+1. `[db.]job_name`
+
+    指定要修改的作业名称。
+
+2. `tbl_name`
+
+    指定需要导入的表的名称。
+
+3. `job_properties`
+
+    指定需要修改的作业参数。目前仅支持如下参数的修改：
+
+    1. `desired_concurrent_number`
+    2. `max_error_number`
+    3. `max_batch_interval`
+    4. `max_batch_rows`
+    5. `max_batch_size`
+    6. `jsonpaths`
+    7. `json_root`
+    8. `strip_outer_array`
+    9. `strict_mode`
+    10. `timezone`
+
+
+4. `data_source`
+
+    数据源的类型。当前支持：
+
+    KAFKA
+
+5. `data_source_properties`
+
+    数据源的相关属性。目前仅支持：
+
+    1. `kafka_partitions`
+    2. `kafka_offsets`
+    3. 自定义 property，如 `property.group.id`
+
+    注：
+
+    1. `kafka_partitions` 和 `kafka_offsets` 用于修改待消费的 kafka partition 的offset，仅能修改当前已经消费的 partition。不能新增 partition。
+
+## example
+
+1. 将 `desired_concurrent_number` 修改为 1
+
+    ```
+    ALTER ROUTINE LOAD FOR db1.label1
+    PROPERTIES
+    (
+        "desired_concurrent_number" = "1"
+    );
+    ```
+
+2.  将 `desired_concurrent_number` 修改为 10，修改 partition 的offset，修改 group id。
+
+    ```
+    ALTER ROUTINE LOAD FOR db1.label1
+    PROPERTIES
+    (
+        "desired_concurrent_number" = "10"
+    )
+    FROM kafka
+    (
+        "kafka_partitions" = "0, 1, 2",
+        "kafka_offsets" = "100, 200, 100",
+        "property.group.id" = "new_group"
+    );
+    ```
+
+
+## keyword
+
+    ALTER,ROUTINE,LOAD
+

fe/fe-core/src/main/cup/sql_parser.cup

@@ -473,11 +473,13 @@ nonterminal Boolean opt_verbose;
 nonterminal Boolean opt_tmp;
 
 nonterminal OutFileClause opt_outfile;
+nonterminal RoutineLoadDataSourceProperties opt_datasource_properties;
 
 precedence nonassoc COMMA;
 precedence nonassoc STRING_LITERAL;
 precedence nonassoc KW_COLUMNS;
 precedence nonassoc KW_WITH;
+
 precedence left KW_FULL, KW_MERGE;
 precedence left DOT;
 precedence left SET_VAR;
@@ -728,11 +730,26 @@ alter_stmt ::=
     {:
         RESULT = new AlterDatabaseQuotaStmt(dbName, QuotaType.REPLICA, String.valueOf(number));
     :}
-
     | KW_ALTER KW_DATABASE ident:dbName KW_RENAME ident:newDbName
     {:
         RESULT = new AlterDatabaseRename(dbName, newDbName);
     :}
+    | KW_ALTER KW_ROUTINE KW_LOAD KW_FOR job_label:jobLabel opt_properties:jobProperties
+      opt_datasource_properties:datasourceProperties
+    {:
+        RESULT = new AlterRoutineLoadStmt(jobLabel, jobProperties, datasourceProperties);
+    :}
+    ;
+
+opt_datasource_properties ::=
+    // empty
+    {:
+        RESULT = new RoutineLoadDataSourceProperties();
+    :}
+    | KW_FROM ident:type LPAREN key_value_map:customProperties RPAREN
+    {:
+        RESULT = new RoutineLoadDataSourceProperties(type, customProperties);
+    :}
     ;
 
 quantity ::=