[Fix-4004] Fix unable to execute statements such as create database (#…

…4202)
DataLinkDC · Feb 8, 2025 · 4072b0e · 4072b0e
1 parent ade81bd
commit 4072b0e
Show file tree

Hide file tree

Showing 5 changed files with 17 additions and 7 deletions.
diff --git a/dinky-core/src/main/java/org/dinky/executor/ExecutorFactory.java b/dinky-core/src/main/java/org/dinky/executor/ExecutorFactory.java
@@ -38,7 +38,7 @@ public static Executor getDefaultExecutor() {
     }
 
     public static Executor buildExecutor(ExecutorConfig executorConfig, DinkyClassLoader classLoader) {
-        if (executorConfig.isRemote()) {
+        if (executorConfig.isRemote() && !executorConfig.isPlan()) {
             return buildRemoteExecutor(executorConfig, classLoader);
         } else {
             return buildLocalExecutor(executorConfig, classLoader);

diff --git a/dinky-core/src/main/java/org/dinky/job/Job.java b/dinky-core/src/main/java/org/dinky/job/Job.java
@@ -54,7 +54,7 @@ public class Job {
     private Executor executor;
     private boolean useGateway;
     private List<String> jids;
-    private boolean isPipeline = true;
+    private boolean isPipeline = false;
 
     @Getter
     public enum JobStatus {

diff --git a/dinky-core/src/main/java/org/dinky/job/runner/JobJarRunner.java b/dinky-core/src/main/java/org/dinky/job/runner/JobJarRunner.java
@@ -85,6 +85,7 @@ public JobJarRunner(JobManager jobManager) {
 
     @Override
     public void run(JobStatement jobStatement) throws Exception {
+        jobManager.getJob().setPipeline(true);
         if (!jobManager.isUseGateway()) {
             submitNormal(jobStatement);
         } else {
@@ -169,7 +170,12 @@ private GatewayResult submitNormalWithGateway(JobStatement jobStatement) {
     private Pipeline getPipeline(JobStatement jobStatement) {
         Pipeline pipeline = getJarStreamGraph(jobStatement.getStatement(), jobManager.getDinkyClassLoader());
         if (pipeline instanceof StreamGraph) {
-            if (Asserts.isNotNullString(jobManager.getConfig().getSavePointPath())) {
+            if (Asserts.isNotNullString(jobManager.getConfig().getSavePointPath())
+                    || (Asserts.isNotNull(jobManager.getConfig().getConfigJson())
+                            && Asserts.isNotNullString(jobManager
+                                    .getConfig()
+                                    .getConfigJson()
+                                    .get(SavepointConfigOptions.SAVEPOINT_PATH)))) {
                 ((StreamGraph) pipeline)
                         .setSavepointRestoreSettings(SavepointRestoreSettings.forPath(
                                 jobManager.getConfig().getSavePointPath(),

diff --git a/dinky-core/src/main/java/org/dinky/job/runner/JobPipelineRunner.java b/dinky-core/src/main/java/org/dinky/job/runner/JobPipelineRunner.java
@@ -91,6 +91,7 @@ public Optional<JobClient> execute(JobStatement jobStatement) throws Exception {
 
     @Override
     public void run(JobStatement jobStatement) throws Exception {
+        jobManager.getJob().setPipeline(true);
         if (ExecuteJarParseStrategy.INSTANCE.match(jobStatement.getStatement())) {
             JobJarRunner jobJarRunner = new JobJarRunner(jobManager);
             jobJarRunner.run(jobStatement);

diff --git a/dinky-core/src/main/java/org/dinky/job/runner/JobSqlRunner.java b/dinky-core/src/main/java/org/dinky/job/runner/JobSqlRunner.java
@@ -105,6 +105,7 @@ public Optional<JobClient> execute(JobStatement jobStatement) throws Exception {
     public void run(JobStatement jobStatement) throws Exception {
         statements.add(jobStatement);
         if (jobStatement.isFinalExecutableStatement()) {
+            jobManager.getJob().setPipeline(isPipeline());
             if (inferStatementSet()) {
                 handleStatementSet();
             } else {
@@ -225,6 +226,11 @@ public JobPlanInfo getJobPlanInfo(JobStatement jobStatement) {
         throw new DinkyException("None jobs in statement.");
     }
 
+    private boolean isPipeline() {
+        return statements.stream()
+                .anyMatch(jobStatement -> jobStatement.getSqlType().isPipeline());
+    }
+
     private boolean inferStatementSet() {
         boolean hasInsert = false;
         for (JobStatement item : statements) {
@@ -266,7 +272,6 @@ private void processWithoutGateway() {
 
     private void processSingleInsertWithGateway() {
         List<JobStatement> singleInsert = Collections.singletonList(statements.get(0));
-        jobManager.getJob().setPipeline(statements.get(0).getSqlType().isPipeline());
         GatewayResult gatewayResult = submitByGateway(singleInsert);
         setJobResultFromGatewayResult(gatewayResult);
     }
@@ -276,9 +281,7 @@ private void processFirstStatement() throws Exception {
             return;
         }
         // Only process the first statement when not using statement set
-        JobStatement item = statements.get(0);
-        jobManager.getJob().setPipeline(item.getSqlType().isPipeline());
-        processSingleStatement(item);
+        processSingleStatement(statements.get(0));
     }
 
     private void processSingleStatement(JobStatement item) {