[Feature-4145][flink] Support CALL statement

dinky-admin/src/main/java/org/dinky/service/impl/TaskServiceImpl.java

@@ -569,7 +569,7 @@ public void initTenantByTaskId(Integer id) {
         Integer tenantId = baseMapper.getTenantByTaskId(id);
         Asserts.checkNull(tenantId, Status.TASK_NOT_EXIST.getMessage());
         TenantContextHolder.set(tenantId);
-        log.info("Init task tenan finished..");
+        log.info("Init task tenant finished..");
     }
 
     @Override
@@ -997,7 +997,7 @@ public Result<Tree<Integer>> queryAllCatalogue() {
     @Override
     public LineageResult getTaskLineage(Integer id) {
         TaskDTO task = getTaskInfoById(id);
-        if (!Dialect.isCommonSql(task.getDialect())) {
+        if (Dialect.isCommonSql(task.getDialect())) {
             if (Asserts.isNull(task.getDatabaseId())) {
                 return null;
             }

dinky-common/src/main/java/org/dinky/data/job/SqlType.java

@@ -88,18 +88,21 @@ public enum SqlType {
 
     WITH("WITH", "^WITH.*", SqlCategory.DQL),
 
+    CALL("CALL", "^CALL.*", SqlCategory.DQL),
+
     UNKNOWN("UNKNOWN", "^UNKNOWN.*", SqlCategory.UNKNOWN);
 
     private String type;
     private final Pattern pattern;
     private final SqlCategory category;
 
     private static final List<SqlType> TRANS_SQL_TYPES =
-            Lists.newArrayList(INSERT, SELECT, WITH, SHOW, DESCRIBE, DESC, CTAS, RTAS, UPDATE, DELETE);
+            Lists.newArrayList(INSERT, SELECT, WITH, SHOW, DESCRIBE, DESC, CTAS, RTAS, UPDATE, DELETE, CALL);
 
     private static final List<SqlType> CTAS_TYPES = Lists.newArrayList(CTAS, RTAS, PRINT);
 
-    private static final List<SqlType> PIPELINE_SQL_TYPES = Lists.newArrayList(INSERT, SELECT, WITH, CTAS, RTAS, PRINT);
+    private static final List<SqlType> PIPELINE_SQL_TYPES =
+            Lists.newArrayList(INSERT, SELECT, WITH, CTAS, RTAS, PRINT, CALL);
 
     private static final List<SqlType> SINK_MODIFY_SQL_TYPES = Lists.newArrayList(INSERT, CTAS, RTAS, PRINT);
 

dinky-core/src/main/java/org/dinky/job/JobManager.java

@@ -247,7 +247,6 @@ public JobResult executeJarSql(String statement) throws Exception {
                 .collect(Collectors.toList());
         statement = String.join(";\n", statements);
         jobStatementPlan = Explainer.build(this).parseStatements(SqlUtil.getStatements(statement));
-        jobStatementPlan.setSubmissionMode(config.isSubmissionMode());
         jobStatementPlan.buildFinalStatement();
         job = Job.build(runMode, config, executorConfig, executor, statement, useGateway);
         ready();
@@ -282,7 +281,6 @@ public JobResult executeSql(String statement) throws Exception {
         ready();
         try {
             jobStatementPlan = Explainer.build(this).parseStatements(SqlUtil.getStatements(statement));
-            jobStatementPlan.setSubmissionMode(config.isSubmissionMode());
             jobStatementPlan.buildFinalStatement();
             JobRunnerFactory jobRunnerFactory = JobRunnerFactory.create(this);
             for (JobStatement jobStatement : jobStatementPlan.getJobStatementList()) {

dinky-core/src/main/java/org/dinky/job/JobStatementPlan.java

@@ -35,22 +35,13 @@
 public class JobStatementPlan {
 
     private List<JobStatement> jobStatementList = new ArrayList<>();
-    private boolean isSubmissionMode;
 
     public JobStatementPlan() {}
 
     public List<JobStatement> getJobStatementList() {
         return jobStatementList;
     }
 
-    public boolean isSubmissionMode() {
-        return isSubmissionMode;
-    }
-
-    public void setSubmissionMode(boolean isSubmissionMode) {
-        this.isSubmissionMode = isSubmissionMode;
-    }
-
     public void addJobStatement(String statement, JobStatementType statementType, SqlType sqlType) {
         jobStatementList.add(new JobStatement(jobStatementList.size() + 1, statement, statementType, sqlType));
     }
@@ -76,7 +67,8 @@ public void buildFinalStatement() {
         if (executableIndex >= 0) {
             jobStatementList.get(executableIndex).asFinalExecutableStatement();
         } else {
-            // If there is no INSERT/CTAS/RTAS statement, use the first SELECT/WITH/SHOW/DESC SQL statement as the final
+            // If there is no INSERT/CTAS/RTAS/CALL statement, use the first SELECT/WITH/SHOW/DESC SQL statement as the
+            // final
             // statement.
             for (int i = 0; i < jobStatementList.size(); i++) {
                 if (jobStatementList.get(i).getStatementType().equals(JobStatementType.SQL)) {
@@ -94,43 +86,13 @@ public void buildFinalStatement() {
 
     public void checkStatement() {
         checkEmptyStatement();
-        checkSubmissionStatement();
         checkPipelineStatement();
     }
 
     private void checkEmptyStatement() {
         if (jobStatementList.isEmpty()) {
             throw new DinkyException("None of valid statement is executed. Please check your statements.");
         }
-        boolean hasSqlStatement = false;
-        for (JobStatement jobStatement : jobStatementList) {
-            if (jobStatement.getStatement().trim().isEmpty()) {
-                throw new DinkyException("The statement cannot be empty. Please check your statements.");
-            }
-            if (jobStatement.getStatementType().equals(JobStatementType.SQL)
-                    || jobStatement.getStatementType().equals(JobStatementType.PIPELINE)
-                    || jobStatement.getStatementType().equals(JobStatementType.EXECUTE_JAR)) {
-                hasSqlStatement = true;
-            }
-        }
-        if (!hasSqlStatement) {
-            throw new DinkyException(
-                    "The statements must contain at least one INSERT/CTAS/RTAS/SELECT/WITH/SHOW/DESC statement.");
-        }
-    }
-
-    private void checkSubmissionStatement() {
-        if (isSubmissionMode) {
-            for (JobStatement jobStatement : jobStatementList) {
-                if (jobStatement.getStatementType().equals(JobStatementType.SQL)) {
-                    if (!jobStatement.getSqlType().isSinkyModify()) {
-                        throw new DinkyException(
-                                "The submission mode cannot contain one statement which is not a sink operation."
-                                        + "\nThe valid statement is: " + jobStatement.getStatement());
-                    }
-                }
-            }
-        }
     }
 
     private void checkPipelineStatement() {

dinky-core/src/main/java/org/dinky/job/runner/JobSqlRunner.java

@@ -46,7 +46,11 @@
 import org.apache.flink.runtime.jobgraph.SavepointRestoreSettings;
 import org.apache.flink.runtime.rest.messages.JobPlanInfo;
 import org.apache.flink.streaming.api.graph.StreamGraph;
+import org.apache.flink.table.api.ResultKind;
 import org.apache.flink.table.api.TableResult;
+import org.apache.flink.table.api.internal.TableResultImpl;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.CloseableIterator;
 
 import java.time.LocalDateTime;
 import java.util.ArrayList;
@@ -297,6 +301,18 @@ private void updateJobWithTableResult(TableResult tableResult, SqlType sqlType)
             jobManager
                     .getJob()
                     .setJids(Collections.singletonList(jobManager.getJob().getJobId()));
+        } else if (ResultKind.SUCCESS_WITH_CONTENT.equals(tableResult.getResultKind())) {
+            TableResultImpl tableResultImpl = (TableResultImpl) tableResult;
+            CloseableIterator<Row> rowCloseableIterator = tableResultImpl.collect();
+            if (rowCloseableIterator.hasNext()) {
+                Row row = rowCloseableIterator.next();
+                String jobIDStringData = String.valueOf(row.getField(0));
+                if (Asserts.isNotNullString(jobIDStringData)) {
+                    String jobID = jobIDStringData.replace("JobID=", "");
+                    jobManager.getJob().setJobId(jobID);
+                    jobManager.getJob().setJids(Collections.singletonList(jobID));
+                }
+            }
         }
 
         if (jobManager.getConfig().isUseResult()) {

dinky-core/src/test/java/org/dinky/job/JobStatementPlanTest.java

@@ -37,28 +37,7 @@ void testEmptyStatements() {
     @Test
     void testEmptyStatement() {
         JobStatementPlan jobStatementPlan = new JobStatementPlan();
-        jobStatementPlan.addJobStatement("", JobStatementType.SQL, SqlType.UNKNOWN);
-        checkInvalidStatement(jobStatementPlan, "The statement cannot be empty. Please check your statements.");
-    }
-
-    @Test
-    void testNoSqlStatement() {
-        JobStatementPlan jobStatementPlan = new JobStatementPlan();
-        jobStatementPlan.addJobStatement("set 'parallelism.default' = '2';\n", JobStatementType.DDL, SqlType.SET);
-        checkInvalidStatement(
-                jobStatementPlan,
-                "The statements must contain at least one INSERT/CTAS/RTAS/SELECT/WITH/SHOW/DESC statement.");
-    }
-
-    @Test
-    void testSubmissionWithQueryStatement() {
-        JobStatementPlan jobStatementPlan = new JobStatementPlan();
-        jobStatementPlan.setSubmissionMode(true);
-        jobStatementPlan.addJobStatement("select 'A' as name;\n", JobStatementType.SQL, SqlType.SET);
-        checkInvalidStatement(
-                jobStatementPlan,
-                "The submission mode cannot contain one statement which is not a sink operation."
-                        + "\nThe valid statement is: select 'A' as name;\n");
+        checkInvalidStatement(jobStatementPlan, "None of valid statement is executed. Please check your statements.");
     }
 
     @Test