[Feature] ADD: show create routine load

docs/.vuepress/sidebar/en.js

@@ -484,6 +484,7 @@ module.exports = [
               "SHOW ALTER",
               "SHOW BACKUP",
               "SHOW CREATE FUNCTION",
+              "SHOW CREATE ROUTINE LOAD",
               "SHOW DATA",
               "SHOW DATABASES",
               "SHOW DELETE",

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

@@ -487,6 +487,7 @@ module.exports = [
               "SHOW ALTER",
               "SHOW BACKUP",
               "SHOW CREATE FUNCTION",
+              "SHOW CREATE ROUTINE LOAD",
               "SHOW DATA",
               "SHOW DATABASES",
               "SHOW DELETE",

docs/en/sql-reference/sql-statements/Data Manipulation/SHOW CREATE ROUTINE LOAD.md

@@ -0,0 +1,42 @@
+---
+{
+"title": "SHOW CREATE 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.
+-->
+
+# SHOW CREATE ROUTINE LOAD
+## description
+    The statement is used to show the routine load job creation statement of user-defined
+    grammar：
+        SHOW [ALL] CREATE ROUTINE LOAD for load_name;
+
+    Description：
+       `ALL`: optional，Is for getting all jobs, including history jobs
+       `load_name`: routine load name
+
+## example
+    1. Show the creation statement of the specified routine load under the default db
+        SHOW CREATE ROUTINE LOAD for test_load
+
+## keyword
+    SHOW,CREATE,ROUTINE,LOAD

docs/zh-CN/sql-reference/sql-statements/Data Manipulation/SHOW CREATE ROUTINE LOAD.md

@@ -0,0 +1,42 @@
+---
+{
+"title": "SHOW CREATE 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.
+-->
+
+# SHOW CREATE ROUTINE LOAD
+## description
+    该语句用于展示例行导入作业的创建语句
+    语法：
+        SHOW [ALL] CREATE ROUTINE LOAD for load_name;
+
+    说明：
+       `ALL`: 可选参数，代表获取所有作业，包括历史作业
+       `load_name`: 例行导入作业名称
+
+## example
+    1. 展示默认db下指定例行导入作业的创建语句
+        SHOW CREATE ROUTINE LOAD for test_load
+
+## keyword
+    SHOW,CREATE,ROUTINE,LOAD

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

@@ -284,7 +284,7 @@ terminal String COMMENTED_PLAN_HINTS;
 nonterminal List<StatementBase> stmts;
 nonterminal StatementBase stmt, show_stmt, show_param, help_stmt, load_stmt,
     create_routine_load_stmt, pause_routine_load_stmt, resume_routine_load_stmt, stop_routine_load_stmt,
-    show_routine_load_stmt, show_routine_load_task_stmt,
+    show_routine_load_stmt, show_routine_load_task_stmt, show_create_routine_load_stmt,
     describe_stmt, alter_stmt,
     use_stmt, kill_stmt, drop_stmt, recover_stmt, grant_stmt, revoke_stmt, create_stmt, set_stmt, sync_stmt, cancel_stmt, cancel_param, delete_stmt,
     link_stmt, migrate_stmt, enter_stmt, unsupported_stmt, export_stmt, admin_stmt, truncate_stmt,
@@ -666,6 +666,8 @@ stmt ::=
     {: RESULT = stmt; :}
     | show_routine_load_task_stmt : stmt
     {: RESULT = stmt; :}
+    | show_create_routine_load_stmt : stmt
+    {: RESULT = stmt; :}
     | cancel_stmt : stmt
     {: RESULT = stmt; :}
     | delete_stmt : stmt
@@ -1681,6 +1683,17 @@ show_routine_load_task_stmt ::=
     :}
     ;
 
+show_create_routine_load_stmt ::=
+    KW_SHOW KW_CREATE KW_ROUTINE KW_LOAD KW_FOR job_label:jobLabel
+    {:
+        RESULT = new ShowCreateRoutineLoadStmt(jobLabel, false);
+    :}
+    | KW_SHOW KW_ALL KW_CREATE KW_ROUTINE KW_LOAD KW_FOR job_label:jobLabel
+    {:
+        RESULT = new ShowCreateRoutineLoadStmt(jobLabel, true);
+    :}
+    ;
+
 // Grant statement
 grant_stmt ::=
     KW_GRANT privilege_list:privs KW_ON tbl_pattern:tblPattern KW_TO user_identity:userId

fe/fe-core/src/main/java/org/apache/doris/analysis/ShowCreateRoutineLoadStmt.java

@@ -0,0 +1,65 @@
+// 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.
+
+package org.apache.doris.analysis;
+
+import org.apache.doris.catalog.Column;
+import org.apache.doris.catalog.ScalarType;
+import org.apache.doris.common.AnalysisException;
+import org.apache.doris.qe.ShowResultSetMetaData;
+
+// SHOW CREATE ROUTINE LOAD statement.
+public class ShowCreateRoutineLoadStmt extends ShowStmt {
+
+    private static final ShowResultSetMetaData META_DATA =
+            ShowResultSetMetaData.builder()
+                    .addColumn(new Column("Routine Load Id", ScalarType.createVarchar(20)))
+                    .addColumn(new Column("Routine Load Name", ScalarType.createVarchar(20)))
+                    .addColumn(new Column("Create Routine Load", ScalarType.createVarchar(30)))
+                    .build();
+
+    private final LabelName labelName;
+
+    private final boolean includeHistory;
+
+    public ShowCreateRoutineLoadStmt(LabelName labelName, boolean includeHistory) {
+        this.labelName = labelName;
+        this.includeHistory = includeHistory;
+    }
+
+    public String getDb() {
+        return labelName.getDbName();
+    }
+
+    public String getLabel() {
+        return labelName.getLabelName();
+    }
+
+    public boolean isIncludeHistory() {
+        return includeHistory;
+    }
+
+    @Override
+    public void analyze(Analyzer analyzer) throws AnalysisException {
+        labelName.analyze(analyzer);
+    }
+
+    @Override
+    public ShowResultSetMetaData getMetaData() {
+        return META_DATA;
+    }
+}

fe/fe-core/src/main/java/org/apache/doris/load/routineload/KafkaRoutineLoadJob.java

@@ -63,6 +63,7 @@
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
+import java.util.HashMap;
 import java.util.TimeZone;
 import java.util.UUID;
 
@@ -494,6 +495,21 @@ protected String customPropertiesJsonToString() {
         return gson.toJson(customProperties);
     }
 
+    @Override
+    protected Map<String, String> getDataSourceProperties() {
+        Map<String, String> dataSourceProperties = Maps.newHashMap();
+        dataSourceProperties.put("kafka_broker_list", brokerList);
+        dataSourceProperties.put("kafka_topic", topic);
+        return dataSourceProperties;
+    }
+
+    @Override
+    protected Map<String, String> getCustomProperties() {
+        Map<String, String> ret = new HashMap<>();
+        customProperties.forEach((k, v) -> ret.put("property." + k, v));
+        return ret;
+    }
+
     @Override
     public void write(DataOutput out) throws IOException {
         super.write(out);

fe/fe-core/src/main/java/org/apache/doris/load/routineload/RoutineLoadJob.java

@@ -1333,6 +1333,84 @@ public List<List<String>> getTasksShowInfo() {
         return rows;
     }
 
+    public String getShowCreateInfo() {
+        Database db = Catalog.getCurrentCatalog().getDb(dbId);
+        Table tbl = (db == null) ? null : db.getTable(tableId);
+        StringBuilder sb = new StringBuilder();
+        // 1.job_name
+        sb.append("CREATE ROUTINE LOAD ").append(name);
+        // 2.tbl_name
+        sb.append(" ON ").append(tbl == null ? String.valueOf(tableId) : tbl.getName()).append("\n");
+        // 3.merge_type
+        sb.append("WITH ").append(mergeType.name()).append("\n");
+        // 4.load_properties
+        // 4.1.column_separator
+        if (columnSeparator != null) {
+            sb.append("COLUMNS TERMINATED BY \"").append(columnSeparator.getSeparator()).append("\",\n");
+        }
+        // 4.2.columns_mapping
+        if (columnDescs != null) {
+            sb.append("COLUMNS(").append(Joiner.on(",").join(columnDescs.descs)).append("),\n");
+        }
+        // 4.3.where_predicates
+        if (whereExpr != null) {
+            sb.append("WHERE ").append(whereExpr.toSql()).append(",\n");
+        }
+        // 4.4.partitions
+        if (partitions != null) {
+            sb.append("PARTITION(").append(Joiner.on(",").join(partitions.getPartitionNames())).append("),\n");
+        }
+        // 4.5.delete_on_predicates
+        if (deleteCondition != null) {
+            sb.append("DELETE ON ").append(deleteCondition.toSql()).append(",\n");
+        }
+        // 4.6.source_sequence
+        if (sequenceCol != null) {
+            sb.append("ORDER BY ").append(sequenceCol).append(",\n");
+        }
+        // 4.7.preceding_predicates
+        if (precedingFilter != null) {
+            sb.append("PRECEDING FILTER ").append(precedingFilter.toSql()).append(",\n");
+        }
+        // remove the last ,
+        if (",".equals(sb.charAt(sb.length() - 2))) {
+            sb.replace(sb.length() - 2, sb.length() - 1, "");
+        }
+        // 5.job_properties
+        sb.append("PROPERTIES\n(\n");
+        appendProperties(sb, CreateRoutineLoadStmt.DESIRED_CONCURRENT_NUMBER_PROPERTY, desireTaskConcurrentNum, false);
+        appendProperties(sb, CreateRoutineLoadStmt.MAX_BATCH_INTERVAL_SEC_PROPERTY, maxBatchIntervalS, false);
+        appendProperties(sb, CreateRoutineLoadStmt.MAX_BATCH_ROWS_PROPERTY, maxBatchRows, false);
+        appendProperties(sb, CreateRoutineLoadStmt.MAX_BATCH_SIZE_PROPERTY, maxBatchSizeBytes, false);
+        appendProperties(sb, CreateRoutineLoadStmt.MAX_ERROR_NUMBER_PROPERTY, maxErrorNum, false);
+        appendProperties(sb, LoadStmt.STRICT_MODE, isStrictMode(), false);
+        appendProperties(sb, LoadStmt.TIMEZONE, getTimezone(), false);
+        appendProperties(sb, PROPS_FORMAT, getFormat(), false);
+        appendProperties(sb, PROPS_JSONPATHS, getJsonPaths(), false);
+        appendProperties(sb, PROPS_STRIP_OUTER_ARRAY, isStripOuterArray(), false);
+        appendProperties(sb, PROPS_JSONROOT, getJsonRoot(), true);
+        sb.append(")\n");
+        // 6. data_source
+        sb.append("FROM ").append(dataSourceType).append("\n");
+        // 7. data_source_properties
+        sb.append("(\n");
+        getDataSourceProperties().forEach((k, v) -> appendProperties(sb, k, v, false));
+        getCustomProperties().forEach((k, v) -> appendProperties(sb, k, v, false));
+        // remove the last ,
+        sb.replace(sb.length() - 2, sb.length() - 1, "");
+        sb.append(");");
+        return sb.toString();
+    }
+
+    private static void appendProperties(StringBuilder sb, String key, Object value, boolean end) {
+        sb.append("\"").append(key).append("\"").append(" = ").append("\"").append(value).append("\"");
+        if (!end) {
+            sb.append(",\n");
+        } else {
+            sb.append("\n");
+        }
+    }
+
     public List<String> getShowStatistic() {
         Database db = Catalog.getCurrentCatalog().getDb(dbId);
 
@@ -1385,6 +1463,10 @@ private String jobPropertiesToJsonString() {
 
     abstract String customPropertiesJsonToString();
 
+    abstract Map<String, String> getDataSourceProperties();
+
+    abstract Map<String, String> getCustomProperties();
+
     public boolean needRemove() {
         if (!isFinal()) {
             return false;