数据源语句查询与结果展示

dlink-admin/pom.xml

@@ -76,11 +76,11 @@
                 </exclusion>
             </exclusions>
         </dependency>
-        <dependency>
+        <!--<dependency>
             <groupId>org.slf4j</groupId>
             <artifactId>slf4j-api</artifactId>
             <version>1.7.30</version>
-        </dependency>
+        </dependency>-->
         <dependency>
             <groupId>mysql</groupId>
             <artifactId>mysql-connector-java</artifactId>
@@ -97,10 +97,6 @@
             <groupId>org.hibernate</groupId>
             <artifactId>hibernate-validator</artifactId>
         </dependency>
-        <!--<dependency>
-            <groupId>org.flywaydb</groupId>
-            <artifactId>flyway-core</artifactId>
-        </dependency>-->
         <dependency>
             <groupId>junit</groupId>
             <artifactId>junit</artifactId>

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

@@ -69,6 +69,14 @@ public class StudioServiceImpl implements StudioService {
 
     @Override
     public JobResult executeSql(StudioExecuteDTO studioExecuteDTO) {
+        if(Dialect.SQL.equalsVal(studioExecuteDTO.getDialect())){
+            return executeCommonSql(studioExecuteDTO);
+        }else{
+            return executeFlinkSql(studioExecuteDTO);
+        }
+    }
+
+    private JobResult executeFlinkSql(StudioExecuteDTO studioExecuteDTO) {
         JobConfig config = studioExecuteDTO.getJobConfig();
         // If you are using a shared session, configure the current jobmanager address
         if(!config.isUseSession()) {
@@ -80,6 +88,36 @@ public class StudioServiceImpl implements StudioService {
         return jobResult;
     }
 
+    private JobResult executeCommonSql(StudioExecuteDTO studioExecuteDTO) {
+        JobResult result = new JobResult();
+        result.setStatement(studioExecuteDTO.getStatement());
+        result.setStartTime(LocalDateTime.now());
+        if(Asserts.isNull(studioExecuteDTO.getDatabaseId())){
+            result.setSuccess(false);
+            result.setError("请指定数据源");
+            result.setEndTime(LocalDateTime.now());
+            return result;
+        }else{
+            DataBase dataBase = dataBaseService.getById(studioExecuteDTO.getDatabaseId());
+            if(Asserts.isNull(dataBase)){
+                result.setSuccess(false);
+                result.setError("数据源不存在");
+                result.setEndTime(LocalDateTime.now());
+                return result;
+            }
+            try {
+                com.dlink.metadata.result.SelectResult selectResult = Driver.build(dataBase.getDriverConfig()).connect().query(studioExecuteDTO.getStatement(),studioExecuteDTO.getMaxRowNum());
+                result.setResult(selectResult);
+                result.setSuccess(true);
+            }catch (Exception e){
+                result.setSuccess(false);
+                result.setError(e.getMessage());
+            }
+            result.setEndTime(LocalDateTime.now());
+            return result;
+        }
+    }
+
     @Override
     public IResult executeDDL(StudioDDLDTO studioDDLDTO) {
         JobConfig config = studioDDLDTO.getJobConfig();
@@ -110,9 +148,16 @@ public class StudioServiceImpl implements StudioService {
 
     private List<SqlExplainResult> explainCommonSql(StudioExecuteDTO studioExecuteDTO) {
         if(Asserts.isNull(studioExecuteDTO.getDatabaseId())){
-            return new ArrayList<>();
+            return new ArrayList<SqlExplainResult>(){{
+                add(SqlExplainResult.fail(studioExecuteDTO.getStatement(),"请指定数据源"));
+            }};
         }else{
             DataBase dataBase = dataBaseService.getById(studioExecuteDTO.getDatabaseId());
+            if(Asserts.isNull(dataBase)){
+                return new ArrayList<SqlExplainResult>(){{
+                    add(SqlExplainResult.fail(studioExecuteDTO.getStatement(),"数据源不存在"));
+                }};
+            }
             SqlExplainResult explainResult = Driver.build(dataBase.getDriverConfig()).connect().explain(studioExecuteDTO.getStatement());
             return new ArrayList<SqlExplainResult>(){{
                 add(explainResult);

dlink-core/src/main/java/com/dlink/job/JobResult.java

@@ -29,7 +29,10 @@ public class JobResult {
     private LocalDateTime startTime;
     private LocalDateTime endTime;
 
-    public JobResult(Integer id, JobConfig jobConfig, String jobManagerAddress, Job.JobStatus status, String statement, String jobId, String error, IResult result,  LocalDateTime startTime, LocalDateTime endTime) {
+    public JobResult() {
+    }
+
+    public JobResult(Integer id, JobConfig jobConfig, String jobManagerAddress, Job.JobStatus status, String statement, String jobId, String error, IResult result, LocalDateTime startTime, LocalDateTime endTime) {
         this.id = id;
         this.jobConfig = jobConfig;
         this.jobManagerAddress = jobManagerAddress;

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

@@ -2,6 +2,7 @@ package com.dlink.metadata.driver;
 
 import com.dlink.assertion.Asserts;
 import com.dlink.constant.CommonConstant;
+import com.dlink.metadata.result.SelectResult;
 import com.dlink.model.Column;
 import com.dlink.model.Schema;
 import com.dlink.model.Table;
@@ -241,35 +242,45 @@ public abstract class AbstractJdbcDriver extends AbstractDriver {
     }
 
     @Override
-    public List<HashMap<String, Object>> query(String sql) {
+    public SelectResult query(String sql,Integer limit) {
+        SelectResult result = new SelectResult();
         List<HashMap<String, Object>> datas = new ArrayList<>();
         List<Column> columns = new ArrayList<>();
+        List<String> columnNameList = new ArrayList<>();
         PreparedStatement preparedStatement = null;
         ResultSet results = null;
+        int count = 0;
         try {
             preparedStatement = conn.prepareStatement(sql);
             results = preparedStatement.executeQuery();
             ResultSetMetaData metaData = results.getMetaData();
             for (int i = 1; i <= metaData.getColumnCount(); i++) {
+                columnNameList.add(metaData.getColumnLabel(i));
                 Column column = new Column();
-                column.setName(metaData.getColumnName(i));
+                column.setName(metaData.getColumnLabel(i));
                 column.setType(metaData.getColumnTypeName(i));
                 column.setJavaType(getTypeConvert().convert(metaData.getColumnTypeName(i)).getType());
                 columns.add(column);
             }
+            result.setColumns(columnNameList);
             while (results.next()) {
                 HashMap<String, Object> data = new HashMap<>();
                 for (int i = 0; i < columns.size(); i++) {
                     data.put(columns.get(i).getName(), getTypeConvert().convertValue(results, columns.get(i).getName(), columns.get(i).getType()));
                 }
                 datas.add(data);
+                count ++;
+                if(count >= limit){
+                    break;
+                }
             }
         } catch (SQLException e) {
             e.printStackTrace();
         } finally {
             close(preparedStatement, results);
         }
-        return datas;
+        result.setRowData(datas);
+        return result;
     }
 
     @Override

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

@@ -90,7 +90,7 @@ public interface Driver {
 
     boolean execute(String sql);
 
-    List query(String sql);
+    SelectResult query(String sql,Integer limit);
 
     SqlExplainResult explain(String sql);
 

dlink-metadata/dlink-metadata-base/src/main/java/com/dlink/metadata/result/SelectResult.java

 package com.dlink.metadata.result;
 
+import com.dlink.result.AbstractResult;
+import com.dlink.result.IResult;
 import lombok.Getter;
 import lombok.Setter;
 
@@ -14,10 +16,15 @@ import java.util.List;
  */
 @Setter
 @Getter
-public class SelectResult {
+public class SelectResult extends AbstractResult implements IResult {
     private List<String> columns;
-    private List<HashMap<String,Object>> datas;
+    private List<HashMap<String,Object>> rowData;
     private Integer total;
     private Integer page;
     private Integer limit;
+
+    @Override
+    public String getJobId() {
+        return null;
+    }
 }

dlink-metadata/dlink-metadata-mysql/src/test/java/com/dlink/metadata/MysqlTest.java

@@ -2,6 +2,7 @@ package com.dlink.metadata;
 
 import com.dlink.metadata.driver.Driver;
 import com.dlink.metadata.driver.DriverConfig;
+import com.dlink.metadata.result.SelectResult;
 import com.dlink.model.Column;
 import com.dlink.model.Schema;
 import org.junit.Test;
@@ -16,14 +17,15 @@ import java.util.List;
  **/
 public class MysqlTest {
 
+    private static final String IP = "127.0.0.1";
     public Driver getDriver(){
         DriverConfig config = new DriverConfig();
         config.setType("Mysql");
-        config.setIp("10.1.51.25");
+        config.setIp(IP);
         config.setPort(3306);
         config.setUsername("dca");
         config.setPassword("dca");
-        config.setUrl("jdbc:mysql://10.1.51.25:3306/dca?zeroDateTimeBehavior=convertToNull&useUnicode=true&characterEncoding=UTF-8&serverTimezone=UTC&autoReconnect=true");
+        config.setUrl("jdbc:mysql://"+IP+":3306/dca?zeroDateTimeBehavior=convertToNull&useUnicode=true&characterEncoding=UTF-8&serverTimezone=UTC&autoReconnect=true");
         return Driver.build(config).connect();
     }
 
@@ -31,11 +33,11 @@ public class MysqlTest {
     public void connectTest(){
         DriverConfig config = new DriverConfig();
         config.setType("Mysql");
-        config.setIp("10.1.51.25");
+        config.setIp(IP);
         config.setPort(3306);
         config.setUsername("dca");
         config.setPassword("dca");
-        config.setUrl("jdbc:mysql://10.1.51.25:3306/dca?zeroDateTimeBehavior=convertToNull&useUnicode=true&characterEncoding=UTF-8&serverTimezone=UTC&autoReconnect=true");
+        config.setUrl("jdbc:mysql://"+IP+":3306/dca?zeroDateTimeBehavior=convertToNull&useUnicode=true&characterEncoding=UTF-8&serverTimezone=UTC&autoReconnect=true");
         String test = Driver.build(config).test();
         System.out.println(test);
         System.out.println("end...");
@@ -58,7 +60,7 @@ public class MysqlTest {
     @Test
     public void queryTest(){
         Driver driver = getDriver();
-        List query = driver.query("select * from MENU");
+        SelectResult query = driver.query("select * from MENU",10);
         System.out.println("end...");
     }
 }

dlink-web/src/components/Studio/StudioConsole/StudioMsg/index.tsx

 import {Typography, Divider, Badge, Empty,Tag} from "antd";
 import {StateType} from "@/pages/FlinkSqlStudio/model";
 import {connect} from "umi";
-import { FireOutlined } from '@ant-design/icons';
+import {FireOutlined, ScheduleOutlined} from '@ant-design/icons';
+import StudioSqlConfig from "@/components/Studio/StudioRightTool/StudioSqlConfig";
+import {DIALECT} from "@/components/Studio/conf";
 
 const { Title, Paragraph, Text, Link } = Typography;
 
@@ -9,29 +11,53 @@ const StudioMsg = (props:any) => {
 
   const {current} = props;
 
-  return (
-    <Typography>
-      {current.console.result.jobConfig?(<Paragraph>
-        <blockquote><Link href={`http://${current.console.result.jobConfig.address}`} target="_blank">
-          [{current.console.result.jobConfig.session}:{current.console.result.jobConfig.address}]
+  const renderCommonSqlContent = () => {
+    return (<>
+      <Paragraph>
+        <blockquote> <Divider type="vertical"/>{current.console.result.startTime}
+          <Divider type="vertical"/>{current.console.result.endTime}
+          <Divider type="vertical"/>
+          {!(current.console.result.success) ? <><Badge status="error"/><Text type="danger">Error</Text></> :
+            <><Badge status="success"/><Text type="success">Success</Text></>}
+          <Divider type="vertical"/>
+        </blockquote>
+        {current.console.result.statement && (<pre style={{height: '100px'}}>{current.console.result.statement}</pre>)}
+        {current.console.result.error && (<pre style={{height: '100px'}}>{current.console.result.error}</pre>)}
+      </Paragraph>
+    </>)
+  };
+
+  const renderFlinkSqlContent = () => {
+    return (<>
+      <Paragraph>
+        <blockquote><Link href={`http://${current.console.result.jobConfig?.address}`} target="_blank">
+          [{current.console.result.jobConfig?.session}:{current.console.result.jobConfig?.address}]
         </Link> <Divider type="vertical"/>{current.console.result.startTime}
           <Divider type="vertical"/>{current.console.result.endTime}
           <Divider type="vertical"/>
-          {!(current.console.result.status=='SUCCESS') ? <><Badge status="error"/><Text type="danger">Error</Text></> :
+          {!(current.console.result.status==='SUCCESS') ? <><Badge status="error"/><Text type="danger">Error</Text></> :
             <><Badge status="success"/><Text type="success">Success</Text></>}
           <Divider type="vertical"/>
-          {current.console.result.jobConfig.jobName&&<Text code>{current.console.result.jobConfig.jobName}</Text>}
+          {current.console.result.jobConfig?.jobName&&<Text code>{current.console.result.jobConfig?.jobName}</Text>}
           {current.console.result.jobId&&
-          (<>
-            <Divider type="vertical"/>
-            <Tag color="blue" key={current.console.result.jobId}>
-            <FireOutlined /> {current.console.result.jobId}
-          </Tag>
+            (<>
+              <Divider type="vertical"/>
+              <Tag color="blue" key={current.console.result.jobId}>
+                <FireOutlined /> {current.console.result.jobId}
+              </Tag>
             </>)}
         </blockquote>
         {current.console.result.statement && (<pre style={{height: '100px'}}>{current.console.result.statement}</pre>)}
         {current.console.result.error && (<pre style={{height: '100px'}}>{current.console.result.error}</pre>)}
-      </Paragraph>):<Empty image={Empty.PRESENTED_IMAGE_SIMPLE} />
+      </Paragraph>
+    </>)
+  };
+
+
+  return (
+    <Typography>
+      {current.console.result.success?(current.task.dialect === DIALECT.SQL ? renderCommonSqlContent():
+        renderFlinkSqlContent() ):<Empty image={Empty.PRESENTED_IMAGE_SIMPLE} />
       }
     </Typography>
   );

dlink-web/src/components/Studio/StudioConsole/StudioTable/index.tsx

@@ -6,6 +6,7 @@ import {useState} from "react";
 import { SearchOutlined } from '@ant-design/icons';
 import {showJobData} from "@/components/Studio/StudioEvent/DQL";
 import ProTable from '@ant-design/pro-table';
+import {DIALECT} from "@/components/Studio/conf";
 
 const { Option } = Select;
 const { Title, Paragraph, Text, Link } = Typography;
@@ -101,11 +102,13 @@ const StudioTable = (props:any) => {
   };
   return (
     <div style={{width: '100%'}}>
-      {current.console&&current.console.result.jobId?
+      {current.console&&current.console.result.success?
         (<>
-          <Button type="primary" onClick={showDetail} icon={<SearchOutlined />}>
-            获取最新数据
-          </Button>
+          {current.task.dialect === DIALECT.FLINKSQL ?
+            (<Button type="primary" onClick={showDetail} icon={<SearchOutlined/>}>
+              获取最新数据
+            </Button>):undefined
+          }
           {result.rowData&&result.columns?
             <ProTable dataSource={result.rowData} columns={getColumns(result.columns)} search={false}
                       options={{

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

@@ -18,6 +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";
 
 const menu = (
   <Menu>
@@ -82,6 +83,12 @@ const StudioMenu = (props: any) => {
         type: "Studio/saveTabs",
         payload: newTabs,
       });
+      if(current.task.dialect === DIALECT.SQL){
+        dispatch && dispatch({
+          type: "Studio/saveResult",
+          payload: res.datas.result,
+        });
+      }
       useSession && showTables(currentSession.session, dispatch);
     })
   };

dlink-web/src/pages/Welcome.tsx

@@ -481,6 +481,9 @@ export default (): React.ReactNode => {
                 <li>
                   <Link>新增数据源的 Sql 作业语法校验</Link>
                 </li>
+                <li>
+                  <Link>新增数据源的 Sql 作业语句执行</Link>
+                </li>
               </ul>
             </Paragraph>
           </Timeline.Item>