[feature-15837] [api] Auto create workflow while import sql script

Author: eye-gu

dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/controller/ProcessDefinitionController.java

@@ -711,12 +711,18 @@ public Result queryAllProcessDefinitionByProjectCode(@Parameter(hidden = true) @
     public Result importProcessDefinition(@Parameter(hidden = true) @RequestAttribute(value = Constants.SESSION_USER) User loginUser,
                                           @Parameter(name = "projectCode", description = "PROJECT_CODE", required = true) @PathVariable long projectCode,
                                           @RequestParam("file") MultipartFile file) {
-        Map<String, Object> result;
-        if ("application/zip".equals(file.getContentType())) {
-            result = processDefinitionService.importSqlProcessDefinition(loginUser, projectCode, file);
-        } else {
-            result = processDefinitionService.importProcessDefinition(loginUser, projectCode, file);
-        }
+        Map<String, Object> result = processDefinitionService.importProcessDefinition(loginUser, projectCode, file);
+        return returnDataList(result);
+    }
+
+    @Operation(summary = "importSqlProcessDefinition", description = "IMPORT_SQL_PROCESS_DEFINITION_NOTES")
+    @Parameter(name = "file", description = "RESOURCE_FILE", required = true, schema = @Schema(implementation = MultipartFile.class))
+    @PostMapping(value = "/importSql")
+    @ApiException(IMPORT_PROCESS_DEFINE_ERROR)
+    public Result importSqlProcessDefinition(@Parameter(hidden = true) @RequestAttribute(value = Constants.SESSION_USER) User loginUser,
+                                             @Parameter(name = "projectCode", description = "PROJECT_CODE", required = true) @PathVariable long projectCode,
+                                             @RequestParam("file") MultipartFile file) {
+        Map<String, Object> result = processDefinitionService.importSqlProcessDefinition(loginUser, projectCode, file);
         return returnDataList(result);
     }
 

dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/ProcessDefinitionServiceImpl.java

@@ -0,0 +1,60 @@
+/*
+ * 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.dolphinscheduler.api.task;
+
+import org.apache.dolphinscheduler.dao.entity.DataSource;
+
+import java.util.Map;
+import java.util.Objects;
+
+import lombok.Data;
+
+@Data
+public class SqlTaskParseContext {
+
+    private Map<String, Object> processProperties;
+
+    private Map<String, Object> hints;
+
+    private String sql;
+
+    private String taskName;
+
+    private DataSource dataSource;
+
+    public <T> T hint(String key) {
+        Object o = hints.get(key);
+        if (Objects.nonNull(o)) {
+            return (T) o;
+        }
+        o = processProperties.get(key);
+        if (Objects.nonNull(o)) {
+            return (T) o;
+        }
+        return null;
+    }
+
+    public <T> T hintOrDefault(String key, T defaultValue) {
+        T hint = hint(key);
+        if (Objects.nonNull(hint)) {
+            return hint;
+        } else {
+            return defaultValue;
+        }
+    }
+}

dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/task/SqlTaskParseContext.java

@@ -0,0 +1,25 @@
+/*
+ * 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.dolphinscheduler.api.task;
+
+public interface SqlTaskParsePlugin {
+
+    String name();
+
+    SqlTaskParseResult parse(SqlTaskParseContext context);
+}

dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/task/SqlTaskParsePlugin.java

@@ -0,0 +1,23 @@
+/*
+ * 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.dolphinscheduler.api.task;
+
+public interface SqlTaskParsePluginFactory {
+
+    SqlTaskParsePlugin build();
+}

dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/task/SqlTaskParsePluginFactory.java

@@ -0,0 +1,62 @@
+/*
+ * 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.dolphinscheduler.api.task;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.ServiceLoader;
+import java.util.stream.Collectors;
+
+import com.google.common.collect.ImmutableMap;
+
+public class SqlTaskParsePluginLoader {
+
+    private SqlTaskParsePluginLoader() {
+    }
+
+    private static final Map<String, SqlTaskParsePlugin> PLUGIN_MAP;
+
+    static {
+        List<SqlTaskParsePlugin> plugins = load();
+        PLUGIN_MAP = ImmutableMap
+                .copyOf(plugins.stream().collect(Collectors.toMap(SqlTaskParsePlugin::name, plugin -> plugin)));
+    }
+
+    private static List<SqlTaskParsePlugin> load() {
+        List<SqlTaskParsePlugin> plugins = new ArrayList<>();
+        for (SqlTaskParsePluginFactory sqlTaskParsePluginFactory : ServiceLoader
+                .load(SqlTaskParsePluginFactory.class)) {
+            SqlTaskParsePlugin plugin = sqlTaskParsePluginFactory.build();
+            plugins.add(plugin);
+        }
+        return plugins;
+    }
+
+    public static List<SqlTaskParsePlugin> allSqlTaskParsePlugins() {
+        return new ArrayList<>(PLUGIN_MAP.values());
+    }
+
+    public static SqlTaskParsePlugin getSqlTaskParsePlugin(String pluginName) {
+        return PLUGIN_MAP.get(pluginName);
+    }
+
+    public static SqlTaskParsePlugin defaultSqlTaskParsePlugin() {
+        return PLUGIN_MAP.get("hint");
+    }
+}

dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/task/SqlTaskParsePluginLoader.java

@@ -0,0 +1,30 @@
+/*
+ * 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.dolphinscheduler.api.task;
+
+import java.util.Set;
+
+import lombok.Data;
+
+@Data
+public class SqlTaskParseResult {
+
+    private Set<String> upstreamSet;
+
+    private Set<String> downstreamSet;
+}

dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/task/SqlTaskParseResult.java

@@ -0,0 +1,101 @@
+/*
+ * 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.dolphinscheduler.api.task.druid;
+
+import org.apache.dolphinscheduler.api.task.SqlTaskParseContext;
+import org.apache.dolphinscheduler.api.task.SqlTaskParsePlugin;
+import org.apache.dolphinscheduler.api.task.SqlTaskParseResult;
+import org.apache.dolphinscheduler.dao.entity.DataSource;
+
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import com.alibaba.druid.DbType;
+import com.alibaba.druid.sql.SQLUtils;
+import com.alibaba.druid.sql.ast.SQLStatement;
+import com.alibaba.druid.sql.visitor.SchemaStatVisitor;
+import com.alibaba.druid.stat.TableStat;
+
+public class DruidSqlTaskParsePlugin implements SqlTaskParsePlugin {
+
+    @Override
+    public String name() {
+        return "druid parser";
+    }
+
+    @Override
+    public SqlTaskParseResult parse(SqlTaskParseContext context) {
+        DbType dbType = getDbType(context.getDataSource());
+        List<SQLStatement> sqlStatements = SQLUtils.parseStatements(context.getSql(), dbType);
+        Set<String> upstreamList = new HashSet<>();
+        Set<String> downstreamList = new HashSet<>();
+        for (SQLStatement sqlStatement : sqlStatements) {
+            SchemaStatVisitor schemaStatVisitor = SQLUtils.createSchemaStatVisitor(dbType);
+            sqlStatement.accept(schemaStatVisitor);
+
+            Map<TableStat.Name, TableStat> tables = schemaStatVisitor.getTables();
+            for (Map.Entry<TableStat.Name, TableStat> table : tables.entrySet()) {
+                if (table.getValue().getSelectCount() > 0) {
+                    upstreamList.add(table.getKey().getName());
+                } else {
+                    downstreamList.add(table.getKey().getName());
+                }
+            }
+        }
+
+        SqlTaskParseResult result = new SqlTaskParseResult();
+        result.setUpstreamSet(upstreamList);
+        result.setDownstreamSet(downstreamList);
+        return result;
+    }
+
+    private DbType getDbType(DataSource dataSource) {
+        switch (dataSource.getType()) {
+            case MYSQL:
+                return DbType.mysql;
+            case ORACLE:
+                return DbType.oracle;
+            case SQLSERVER:
+                return DbType.sqlserver;
+            case POSTGRESQL:
+                return DbType.postgresql;
+            case PRESTO:
+                return DbType.presto;
+            case CLICKHOUSE:
+                return DbType.clickhouse;
+            case TRINO:
+                return DbType.trino;
+            case DB2:
+                return DbType.db2;
+            case OCEANBASE:
+                return DbType.oceanbase;
+            case STARROCKS:
+                return DbType.starrocks;
+            case H2:
+                return DbType.h2;
+            case HIVE:
+            case SPARK:
+            case KYUUBI:
+                return DbType.hive;
+            default:
+                return DbType.other;
+        }
+    }
+}

dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/task/druid/DruidSqlTaskParsePlugin.java

@@ -0,0 +1,32 @@
+/*
+ * 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.dolphinscheduler.api.task.druid;
+
+import org.apache.dolphinscheduler.api.task.SqlTaskParsePlugin;
+import org.apache.dolphinscheduler.api.task.SqlTaskParsePluginFactory;
+
+import com.google.auto.service.AutoService;
+
+@AutoService(SqlTaskParsePluginFactory.class)
+public class DruidSqlTaskParsePluginFactory implements SqlTaskParsePluginFactory {
+
+    @Override
+    public SqlTaskParsePlugin build() {
+        return new DruidSqlTaskParsePlugin();
+    }
+}