新增 Mysql,Oracle,PostGreSql,ClickHouse,Doris,Java 方言及图标

README.md

@@ -21,7 +21,7 @@ Dlink 是一个交互式的 FlinkSQL Studio，可以在线开发、补全、校
 注意：以下功能均为对应版本已实现的功能，实测可用。
 
 |   应用    |     方向     | 功能                                        |  进展   |
-| :------: |:----------:|---------------------------------------| :---: |
+|:-------:|:----------:|-------------------------------------------|:-----:|
 |  开发中心   |  FlinkSQL  | 支持 sql-client 所有语法                        | 0.4.0 |
 |         |            | 支持 Flink 所有 Configuration                 | 0.4.0 |
 |         |            | 支持 Flink 所有 Connector                     | 0.4.0 |
@@ -29,6 +29,7 @@ Dlink 是一个交互式的 FlinkSQL Studio，可以在线开发、补全、校
 |         |            | 支持 INSERT 语句集                             | 0.4.0 |
 |         |            | 新增 SQL 片段语法                               | 0.4.0 |
 |         |            | 新增 AGGTABLE 表值聚合语法及 UDATF 支持              | 0.4.0 |
+|         |            | 新增 FlinkSQLEnv 执行环境复用                     | 0.5.0 |
 |         |            | 新增 Flink Catalog 交互查询                     | 0.4.0 |
 |         |            | 新增 执行环境的共享与私有会话机制                         | 0.4.0 |
 |         |            | 新增 多种方言的作业目录管理（FlinkSQL、SQL、Java）         | 0.5.0 |
@@ -46,13 +47,14 @@ Dlink 是一个交互式的 FlinkSQL Studio，可以在线开发、补全、校
 |         |            | 支持 yarn session 模式下 FlinkSQL 提交           | 0.4.0 |
 |         |            | 支持 yarn per-job 模式下 FlinkSQL 提交           | 0.4.0 |
 |         |            | 支持 yarn application 模式下 FlinkSQL 提交       | 0.4.0 |
-|          |            | 支持 kubernetes session 模式下 FlinkSQL 提交 | 0.4.0 |
-|          |            | 支持 kubernetes application 模式下 FlinkSQL 提交 | 0.4.0 |
+|         |            | 支持 kubernetes session 模式下 FlinkSQL 提交     | 0.5.0 |
+|         |            | 支持 kubernetes application 模式下 FlinkSQL 提交 | 0.5.0 |
 |         |  Flink 作业  | 支持 yarn application 模式下 Jar 提交            | 0.4.0 |
-|          |            | 支持 k8s application 模式下 Jar 提交         | 0.4.0 |
+|         |            | 支持 k8s application 模式下 Jar 提交             | 0.5.0 |
 |         |            | 支持 作业 Cancel                              | 0.4.0 |
 |         |            | 支持 作业 SavePoint 的 Cancel、Stop、Trigger     | 0.4.0 |
 |         |            | 新增 作业自动从 SavePoint 恢复机制（包含最近、最早、指定一次）     | 0.4.0 |
+|         |            | 新增 UDF java方言代码的开发                        | 0.5.0 |
 |         |  Flink 集群  | 支持 查看已注册集群的作业列表与运维                        | 0.4.0 |
 |         |            | 新增 自动注册 Yarn 创建的集群                        | 0.4.0 |
 |         |    SQL     | 新增 外部数据源的 SQL 校验                          | 0.5.0 |
@@ -69,6 +71,15 @@ Dlink 是一个交互式的 FlinkSQL Studio，可以在线开发、补全、校
 |         |            | 新增 Oracle 数据源注册及测试                        | 0.4.0 |
 |         |            | 新增 postgreSql 数据源注册及测试                    | 0.4.0 |
 |         |            | 新增 ClickHouse 数据源注册及测试                    | 0.4.0 |
+| OpenApi |     调度     | 新增 submitTask 调度接口                        | 0.5.0 |
+|         |  FlinkSQL  | 新增 executeSql 提交接口                        | 0.5.0 |
+|         |            | 新增 explainSql 验证接口                        | 0.5.0 |
+|         |            | 新增 getJobPlan 计划接口                        | 0.5.0 |
+|         |            | 新增 getStreamGraph 计划接口                    | 0.5.0 |
+|         |            | 新增 getJobData 数据接口                        | 0.5.0 |
+|         |   Flink    | 新增 executeJar 提交接口                        | 0.5.0 |
+|         |            | 新增 cancel 停止接口                            | 0.5.0 |
+|         |            | 新增 savepoint 触发接口                         | 0.5.0 |
 |   关于    |  关于 Dlink  | 版本更新记录                                    | 0.4.0 |
 
 ## 部署

dlink-admin/src/main/java/com/dlink/dto/SqlDTO.java

+package com.dlink.dto;
+
+import lombok.Getter;
+import lombok.Setter;
+
+/**
+ * SqlDTO
+ *
+ * @author wenmo
+ * @since 2021/12/29 19:42
+ */
+@Getter
+@Setter
+public class SqlDTO {
+    private String statement;
+    private Integer databaseId;
+    private Integer maxRowNum;
+
+    public SqlDTO(String statement, Integer databaseId, Integer maxRowNum) {
+        this.statement = statement;
+        this.databaseId = databaseId;
+        this.maxRowNum = maxRowNum;
+    }
+
+    public static SqlDTO build(String statement, Integer databaseId, Integer maxRowNum){
+        return new SqlDTO(statement,databaseId,maxRowNum);
+    }
+}

dlink-admin/src/main/java/com/dlink/service/StudioService.java

 package com.dlink.service;
 
 import com.dlink.dto.SessionDTO;
+import com.dlink.dto.SqlDTO;
 import com.dlink.dto.StudioDDLDTO;
 import com.dlink.dto.StudioExecuteDTO;
 import com.dlink.explainer.ca.ColumnCANode;
@@ -25,6 +26,8 @@ public interface StudioService {
 
     JobResult executeSql(StudioExecuteDTO studioExecuteDTO);
 
+    JobResult executeCommonSql(SqlDTO sqlDTO);
+
     IResult executeDDL(StudioDDLDTO studioDDLDTO);
 
     List<SqlExplainResult> explainSql(StudioExecuteDTO studioExecuteDTO);

dlink-admin/src/main/java/com/dlink/service/impl/StudioServiceImpl.java

@@ -3,10 +3,7 @@ package com.dlink.service.impl;
 import com.dlink.api.FlinkAPI;
 import com.dlink.assertion.Asserts;
 import com.dlink.config.Dialect;
-import com.dlink.dto.AbstractStatementDTO;
-import com.dlink.dto.SessionDTO;
-import com.dlink.dto.StudioDDLDTO;
-import com.dlink.dto.StudioExecuteDTO;
+import com.dlink.dto.*;
 import com.dlink.explainer.ca.CABuilder;
 import com.dlink.explainer.ca.ColumnCANode;
 import com.dlink.explainer.ca.TableCANode;
@@ -77,8 +74,9 @@ public class StudioServiceImpl implements StudioService {
 
     @Override
     public JobResult executeSql(StudioExecuteDTO studioExecuteDTO) {
-        if(Dialect.SQL.equalsVal(studioExecuteDTO.getDialect())){
-            return executeCommonSql(studioExecuteDTO);
+        if(Dialect.isSql(studioExecuteDTO.getDialect())){
+            return executeCommonSql(SqlDTO.build(studioExecuteDTO.getStatement(),
+                    studioExecuteDTO.getDatabaseId(),studioExecuteDTO.getMaxRowNum()));
         }else{
             return executeFlinkSql(studioExecuteDTO);
         }
@@ -97,17 +95,17 @@ public class StudioServiceImpl implements StudioService {
         return jobResult;
     }
 
-    private JobResult executeCommonSql(StudioExecuteDTO studioExecuteDTO) {
+    public JobResult executeCommonSql(SqlDTO sqlDTO) {
         JobResult result = new JobResult();
-        result.setStatement(studioExecuteDTO.getStatement());
+        result.setStatement(sqlDTO.getStatement());
         result.setStartTime(LocalDateTime.now());
-        if(Asserts.isNull(studioExecuteDTO.getDatabaseId())){
+        if(Asserts.isNull(sqlDTO.getDatabaseId())){
             result.setSuccess(false);
             result.setError("请指定数据源");
             result.setEndTime(LocalDateTime.now());
             return result;
         }else{
-            DataBase dataBase = dataBaseService.getById(studioExecuteDTO.getDatabaseId());
+            DataBase dataBase = dataBaseService.getById(sqlDTO.getDatabaseId());
             if(Asserts.isNull(dataBase)){
                 result.setSuccess(false);
                 result.setError("数据源不存在");
@@ -115,7 +113,7 @@ public class StudioServiceImpl implements StudioService {
                 return result;
             }
             Driver driver = Driver.build(dataBase.getDriverConfig()).connect();
-            JdbcSelectResult selectResult = driver.query(studioExecuteDTO.getStatement(),studioExecuteDTO.getMaxRowNum());
+            JdbcSelectResult selectResult = driver.query(sqlDTO.getStatement(),sqlDTO.getMaxRowNum());
             driver.close();
             result.setResult(selectResult);
             if(selectResult.isSuccess()){
@@ -141,7 +139,7 @@ public class StudioServiceImpl implements StudioService {
 
     @Override
     public List<SqlExplainResult> explainSql(StudioExecuteDTO studioExecuteDTO) {
-        if( Dialect.SQL.equalsVal(studioExecuteDTO.getDialect())){
+        if( Dialect.isSql(studioExecuteDTO.getDialect())){
             return explainCommonSql(studioExecuteDTO);
         }else{
             return explainFlinkSql(studioExecuteDTO);

dlink-admin/src/main/java/com/dlink/service/impl/TaskServiceImpl.java

@@ -6,6 +6,7 @@ import com.dlink.assertion.Asserts;
 import com.dlink.assertion.Tips;
 import com.dlink.config.Dialect;
 import com.dlink.db.service.impl.SuperServiceImpl;
+import com.dlink.dto.SqlDTO;
 import com.dlink.gateway.GatewayType;
 import com.dlink.job.JobConfig;
 import com.dlink.job.JobManager;
@@ -39,6 +40,8 @@ public class TaskServiceImpl extends SuperServiceImpl<TaskMapper, Task> implemen
     private SavepointsService savepointsService;
     @Autowired
     private JarService jarService;
+    @Autowired
+    private StudioService studioService;
 
     @Value("${spring.datasource.driver-class-name}")
     private String driver;
@@ -57,6 +60,10 @@ public class TaskServiceImpl extends SuperServiceImpl<TaskMapper, Task> implemen
     public JobResult submitByTaskId(Integer id) {
         Task task = this.getTaskInfoById(id);
         Asserts.checkNull(task, Tips.TASK_NOT_EXIST);
+        if(Dialect.isSql(task.getDialect())){
+            return studioService.executeCommonSql(SqlDTO.build(task.getStatement(),
+                    task.getDatabaseId(),null));
+        }
         boolean isJarTask = isJarTask(task);
         if(!isJarTask&&Asserts.isNotNull(task.getEnvId())){
             Task envTask = getTaskInfoById(task.getEnvId());

dlink-core/src/main/java/com/dlink/config/Dialect.java

@@ -10,7 +10,9 @@ import com.dlink.assertion.Asserts;
  **/
 public enum  Dialect {
 
-    FLINKSQL("FlinkSql"),FLINKSQLENV("FlinkSqlEnv"),SQL("Sql"),JAVA("Java");
+    FLINKSQL("FlinkSql"),FLINKSQLENV("FlinkSqlEnv"),SQL("Sql"),JAVA("Java"),
+    MYSQL("Mysql"),ORACLE("Oracle"),POSTGRESQL("PostGreSql"),CLICKHOUSE("ClickHouse"),
+    DORIS("Doris");
 
     private String value;
 
@@ -36,5 +38,15 @@ public enum  Dialect {
         }
         return Dialect.FLINKSQL;
     }
+
+    public static boolean isSql(String value){
+        Dialect dialect = Dialect.get(value);
+        switch (dialect){
+            case SQL:case MYSQL:case ORACLE:case POSTGRESQL:case CLICKHOUSE:case DORIS:
+                return true;
+            default:
+                return false;
+        }
+    }
 }
 

dlink-metadata/dlink-metadata-base/src/main/java/com/dlink/metadata/driver/AbstractJdbcDriver.java

@@ -248,6 +248,9 @@ public abstract class AbstractJdbcDriver extends AbstractDriver {
 
     @Override
     public JdbcSelectResult query(String sql, Integer limit) {
+        if(Asserts.isNull(limit)){
+            limit = 100;
+        }
         JdbcSelectResult result = new JdbcSelectResult();
         List<HashMap<String, Object>> datas = new ArrayList<>();
         List<Column> columns = new ArrayList<>();

dlink-web/src/components/Studio/StudioMenu/index.tsx

@@ -18,7 +18,7 @@ import StudioGraph from "./StudioGraph";
 import {showCluster, showTables, saveTask} from "@/components/Studio/StudioEvent/DDL";
 import {useEffect, useState} from "react";
 import StudioExplain from "../StudioConsole/StudioExplain";
-import {DIALECT} from "@/components/Studio/conf";
+import {DIALECT, isSql} from "@/components/Studio/conf";
 
 const menu = (
   <Menu>
@@ -358,7 +358,7 @@ const StudioMenu = (props: any) => {
                 onClick={onCheckSql}
               />
             </Tooltip>
-            {current.task.dialect == DIALECT.FLINKSQL &&(
+            {current.task.dialect === DIALECT.FLINKSQL &&(
             <Tooltip title="获取当前的 FlinkSql 的执行图">
               <Button
                 type="text"
@@ -366,7 +366,7 @@ const StudioMenu = (props: any) => {
                 onClick={onGetStreamGraph}
               />
             </Tooltip>)}
-            {(current.task.dialect == DIALECT.FLINKSQL||current.task.dialect == DIALECT.SQL) &&(
+            {(current.task.dialect === DIALECT.FLINKSQL||isSql( current.task.dialect )) &&(
             <Tooltip title="执行当前的 FlinkSql">
               <Button
                 type="text"
@@ -375,7 +375,7 @@ const StudioMenu = (props: any) => {
                 onClick={execute}
               />
             </Tooltip>)}
-            {current.task.dialect == DIALECT.FLINKSQL &&(<>
+            {(current.task.dialect === DIALECT.FLINKSQL||isSql( current.task.dialect )) &&(<>
               <Tooltip title="提交当前的作业到集群">
                 <Button
                   type="text"

dlink-web/src/components/Studio/StudioRightTool/StudioSqlConfig/index.tsx

@@ -38,11 +38,13 @@ const StudioSqlConfig = (props: any) => {
   const getDataBaseOptions = () => {
     const itemList = [];
     for (const item of database) {
+      if(item.type.toUpperCase() === current.task.dialect.toUpperCase()) {
         const tag = (<><Tag color={item.enabled ? "processing" : "error"}>{item.type}</Tag>{item.alias}</>);
         itemList.push(<Option key={item.id} value={item.id} label={tag}>
           {tag}
         </Option>)
       }
+    }
     return itemList;
   };
 

dlink-web/src/components/Studio/StudioRightTool/index.tsx

@@ -8,7 +8,7 @@ import StudioSetting from "./StudioSetting";
 import StudioSavePoint from "./StudioSavePoint";
 import StudioEnvSetting from "./StudioEnvSetting";
 import StudioSqlConfig from "./StudioSqlConfig";
-import {DIALECT} from "@/components/Studio/conf";
+import {DIALECT, isSql} from "@/components/Studio/conf";
 
 const { TabPane } = Tabs;
 
@@ -19,11 +19,16 @@ const StudioRightTool = (props:any) => {
   const {current,form,toolHeight} = props;
 
   const renderContent = () => {
-    switch (current.task.dialect){
-      case DIALECT.SQL: return renderSqlContent();
-      case DIALECT.FLINKSQLENV: return renderEnvContent();
-      default: return renderFlinkSqlContent();
+    if(isSql(current.task.dialect)){
+      return renderSqlContent();
     }
+    if(DIALECT.FLINKSQLENV === current.task.dialect){
+      return renderEnvContent();
+    }
+    if(DIALECT.JAVA === current.task.dialect){
+      return undefined;
+    }
+    return renderFlinkSqlContent();
   };
 
   const renderSqlContent = () => {

dlink-web/src/components/Studio/StudioTabs/index.tsx

@@ -5,6 +5,7 @@ import {StateType} from "@/pages/FlinkSqlStudio/model";
 import styles from './index.less';
 import StudioEdit from '../StudioEdit';
 import {saveTask} from "@/components/Studio/StudioEvent/DDL";
+import { DIALECT } from '../conf';
 
 const {TabPane} = Tabs;
 
@@ -79,7 +80,8 @@ const EditorTabs = (props: any) => {
     >
       {tabs.panes.map(pane => (
         <TabPane tab={pane.title} key={pane.key} closable={pane.closable}>
-          <StudioEdit tabsKey={pane.key} height={(toolHeight - 32)} width={width}/>
+          <StudioEdit tabsKey={pane.key} height={(toolHeight - 32)} width={width}
+                      language={current.task.dialect===DIALECT.JAVA?'java':'sql'}/>
         </TabPane>
       ))}
     </Tabs>

dlink-web/src/components/Studio/StudioTree/Function.ts

@@ -44,8 +44,7 @@ export function getTreeNodeByKey(node:any[], key:number) {
       if(result){
         return result;
       }
-    }else{
-      return null;
     }
   }
+  return null;
 }

dlink-web/src/components/Studio/StudioTree/components/SimpleTaskForm.tsx

@@ -53,8 +53,13 @@ const SimpleTaskForm: React.FC<UpdateFormProps> = (props) => {
           <Select defaultValue={DIALECT.FLINKSQL} value={DIALECT.FLINKSQL}>
             <Option value={DIALECT.FLINKSQL}>FlinkSql</Option>
             <Option value={DIALECT.FLINKSQLENV}>FlinkSql 环境</Option>
-            <Option value={DIALECT.SQL}>Sql</Option>
+            <Option value={DIALECT.MYSQL}>Mysql</Option>
+            <Option value={DIALECT.ORACLE}>Oracle</Option>
+            <Option value={DIALECT.POSTGRESQL}>PostGreSql</Option>
+            <Option value={DIALECT.CLICKHOUSE}>ClickHouse</Option>
+            <Option value={DIALECT.DORIS}>Doris</Option>
             <Option value={DIALECT.JAVA}>Java</Option>
+            <Option value={DIALECT.SQL}>Sql</Option>
           </Select>
         </Form.Item>):undefined}
         <Form.Item

dlink-web/src/components/Studio/StudioTree/index.tsx

@@ -13,6 +13,8 @@ import {
 import UpdateCatalogueForm from './components/UpdateCatalogueForm';
 import SimpleTaskForm from "@/components/Studio/StudioTree/components/SimpleTaskForm";
 import { Scrollbars } from "react-custom-scrollbars";
+import {getIcon} from "@/components/Studio/icon";
+
 const { DirectoryTree } = Tree;
 
 
@@ -32,25 +34,9 @@ type RightClickMenu = {
   categoryName: string
 };
 
-// const getParentKey = (key:any, tree:any) => {
-//   let parentKey;
-//   for (let i = 0; i < tree.length; i++) {
-//     const node = tree[i];
-//     if (node.children) {
-//       if (node.children.some((item:any) => item.key === key)) {
-//         parentKey = node.key;
-//       } else if (getParentKey(key, node.children)) {
-//         parentKey = getParentKey(key, node.children);
-//       }
-//     }
-//   }
-//   return parentKey;
-// };
-
 const StudioTree: React.FC<StudioTreeProps> = (props) => {
   const {rightClickMenu,dispatch,tabs,refs,toolHeight} = props;
   const [treeData, setTreeData] = useState<TreeDataNode[]>();
-  //const [dataList, setDataList] = useState<[]>();
   const [expandedKeys, setExpandedKeys] = useState<Key[]>();
   const [rightClickNodeTreeItem,setRightClickNodeTreeItem] = useState<RightClickMenu>();
   const [updateCatalogueModalVisible, handleUpdateCatalogueModalVisible] = useState<boolean>(false);
@@ -60,7 +46,7 @@ const StudioTree: React.FC<StudioTreeProps> = (props) => {
   const [taskFormValues, setTaskFormValues] = useState({});
   const [rightClickNode, setRightClickNode] = useState<TreeDataNode>();
   const [available, setAvailable] = useState<boolean>(true);
-  let sref:any = React.createRef<Scrollbars>();
+  const sref: any = React.createRef<Scrollbars>();
 
   const getTreeData = async () => {
     const result = await getCatalogueTreeData();
@@ -69,8 +55,10 @@ const StudioTree: React.FC<StudioTreeProps> = (props) => {
     for(let i=0;i<list.length;i++){
       list[i].title=list[i].name;
       list[i].key=list[i].id;
+      if(list[i].isLeaf){
+        list[i].icon = getIcon(list[i].type);
+      }
     }
-   // setDataList(list);
     data = convertToTreeData(data, 0);
     setTreeData(data);
   };
@@ -82,8 +70,10 @@ const StudioTree: React.FC<StudioTreeProps> = (props) => {
     for(let i=0;i<list.length;i++){
       list[i].title=list[i].name;
       list[i].key=list[i].id;
+      if(list[i].isLeaf){
+        list[i].icon = getIcon(list[i].type);
+      }
     }
-    //setDataList(list);
     data = convertToTreeData(data, 0);
     setTreeData(data);
     let node = getTreeNodeByKey(data,key);

dlink-web/src/components/Studio/conf.ts

@@ -12,5 +12,24 @@ export const DIALECT = {
   FLINKSQL:'FlinkSql',
   FLINKSQLENV:'FlinkSqlEnv',
   SQL:'Sql',
+  MYSQL:'Mysql',
+  ORACLE:'Oracle',
+  POSTGRESQL:'PostGreSql',
+  CLICKHOUSE:'ClickHouse',
+  DORIS:'Doris',
   JAVA:'Java',
 };
+
+export const isSql = (type: string)=>{
+  switch (type){
+    case DIALECT.SQL:
+    case DIALECT.MYSQL:
+    case DIALECT.ORACLE:
+    case DIALECT.POSTGRESQL:
+    case DIALECT.CLICKHOUSE:
+    case DIALECT.DORIS:
+      return true;
+    default:
+      return false;
+  }
+}

dlink-web/src/pages/Welcome.tsx

@@ -511,6 +511,9 @@ export default (): React.ReactNode => {
                 <li>
                   <Link>新增 FlinkSQL 执行环境方言及其应用功能</Link>
                 </li>
+                <li>
+                  <Link>新增 Mysql,Oracle,PostGreSql,ClickHouse,Doris,Java 方言及图标</Link>
+                </li>
               </ul>
             </Paragraph>
           </Timeline.Item>