Unverified Commit e8a87da6 authored by aiwenmo's avatar aiwenmo Committed by GitHub

[Feature-733][admin,web,client] Add Flink meta store info and column details (#734)

Co-authored-by: 's avatarwenmo <32723967+wenmo@users.noreply.github.com>
parent 56a8db9a
......@@ -177,10 +177,23 @@ public class StudioController {
}
/**
* 获取 Meta Store Table
* 获取 Meta Store Schema/Database 信息
*/
@PostMapping("/getMSTables")
public Result getMSTables(@RequestBody StudioMetaStoreDTO studioMetaStoreDTO) {
return Result.succeed(studioService.getMSTables(studioMetaStoreDTO), "获取成功");
@PostMapping("/getMSSchemaInfo")
public Result getMSSchemaInfo(@RequestBody StudioMetaStoreDTO studioMetaStoreDTO) {
return Result.succeed(studioService.getMSSchemaInfo(studioMetaStoreDTO), "获取成功");
}
/**
* 获取 Meta Store Flink Column 信息
*/
@GetMapping("/getMSFlinkColumns")
public Result getMSFlinkColumns(@RequestParam Integer envId, @RequestParam String catalog, @RequestParam String database, @RequestParam String table) {
StudioMetaStoreDTO studioMetaStoreDTO = new StudioMetaStoreDTO();
studioMetaStoreDTO.setEnvId(envId);
studioMetaStoreDTO.setCatalog(catalog);
studioMetaStoreDTO.setDatabase(database);
studioMetaStoreDTO.setTable(table);
return Result.succeed(studioService.getMSFlinkColumns(studioMetaStoreDTO), "获取成功");
}
}
package com.dlink.dto;
import lombok.Getter;
import java.util.HashMap;
import java.util.Map;
import com.dlink.assertion.Asserts;
import com.dlink.gateway.GatewayType;
import com.dlink.job.JobConfig;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.JsonNode;
import lombok.Getter;
import lombok.Setter;
/**
* StudioMetaStoreDTO
......@@ -18,9 +13,11 @@ import com.fasterxml.jackson.databind.JsonNode;
* @since 2022/7/16 23:18
*/
@Getter
@Setter
public class StudioMetaStoreDTO extends AbstractStatementDTO {
private String catalog;
private String database;
private String table;
private String dialect;
private Integer databaseId;
......
......@@ -11,7 +11,8 @@ import com.dlink.dto.StudioMetaStoreDTO;
import com.dlink.explainer.lineage.LineageResult;
import com.dlink.job.JobResult;
import com.dlink.model.Catalog;
import com.dlink.model.Table;
import com.dlink.model.FlinkColumn;
import com.dlink.model.Schema;
import com.dlink.result.IResult;
import com.dlink.result.SelectResult;
import com.dlink.result.SqlExplainResult;
......@@ -57,5 +58,7 @@ public interface StudioService {
List<Catalog> getMSCatalogs(StudioMetaStoreDTO studioMetaStoreDTO);
List<Table> getMSTables(StudioMetaStoreDTO studioMetaStoreDTO);
Schema getMSSchemaInfo(StudioMetaStoreDTO studioMetaStoreDTO);
List<FlinkColumn> getMSFlinkColumns(StudioMetaStoreDTO studioMetaStoreDTO);
}
......@@ -34,6 +34,7 @@ import com.dlink.metadata.result.JdbcSelectResult;
import com.dlink.model.Catalog;
import com.dlink.model.Cluster;
import com.dlink.model.DataBase;
import com.dlink.model.FlinkColumn;
import com.dlink.model.Savepoints;
import com.dlink.model.Schema;
import com.dlink.model.Table;
......@@ -51,6 +52,7 @@ import com.dlink.service.TaskService;
import com.dlink.session.SessionConfig;
import com.dlink.session.SessionInfo;
import com.dlink.session.SessionPool;
import com.dlink.sql.FlinkQuery;
import com.dlink.utils.RunTimeUtil;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.JsonNode;
......@@ -369,13 +371,13 @@ public class StudioServiceImpl implements StudioService {
if (Dialect.isSql(studioMetaStoreDTO.getDialect())) {
DataBase dataBase = dataBaseService.getById(studioMetaStoreDTO.getDatabaseId());
if (!Asserts.isNull(dataBase)) {
Catalog defaultCatalog = Catalog.build("default_catalog");
Catalog defaultCatalog = Catalog.build(FlinkQuery.defaultCatalog());
Driver driver = Driver.build(dataBase.getDriverConfig());
defaultCatalog.setSchemas(driver.listSchemas());
catalogs.add(defaultCatalog);
}
} else {
studioMetaStoreDTO.setStatement("SHOW CATALOGS");
studioMetaStoreDTO.setStatement(FlinkQuery.showCatalogs());
IResult result = executeMSFlinkSql(studioMetaStoreDTO);
if (result instanceof DDLResult) {
DDLResult ddlResult = (DDLResult) result;
......@@ -388,7 +390,7 @@ public class StudioServiceImpl implements StudioService {
}
}
for (Catalog catalog : catalogs) {
String statement = "USE CATALOG " + catalog.getName() + ";\r\nSHOW DATABASES";
String statement = FlinkQuery.useCatalog(catalog.getName()) + FlinkQuery.separator() + FlinkQuery.showDatabases();
studioMetaStoreDTO.setStatement(statement);
IResult tableResult = executeMSFlinkSql(studioMetaStoreDTO);
if (result instanceof DDLResult) {
......@@ -411,7 +413,8 @@ public class StudioServiceImpl implements StudioService {
}
@Override
public List<Table> getMSTables(StudioMetaStoreDTO studioMetaStoreDTO) {
public Schema getMSSchemaInfo(StudioMetaStoreDTO studioMetaStoreDTO) {
Schema schema = Schema.build(studioMetaStoreDTO.getDatabase());
List<Table> tables = new ArrayList<>();
if (Dialect.isSql(studioMetaStoreDTO.getDialect())) {
DataBase dataBase = dataBaseService.getById(studioMetaStoreDTO.getDatabaseId());
......@@ -420,9 +423,11 @@ public class StudioServiceImpl implements StudioService {
tables.addAll(driver.listTables(studioMetaStoreDTO.getDatabase()));
}
} else {
String statement = "USE CATALOG " + studioMetaStoreDTO.getCatalog() + ";\r\n" +
"USE " + studioMetaStoreDTO.getDatabase() + ";\r\nSHOW TABLES";
studioMetaStoreDTO.setStatement(statement);
String baseStatement = FlinkQuery.useCatalog(studioMetaStoreDTO.getCatalog()) + FlinkQuery.separator() +
FlinkQuery.useDatabase(studioMetaStoreDTO.getDatabase()) + FlinkQuery.separator();
// show tables
String tableStatement = baseStatement + FlinkQuery.showTables();
studioMetaStoreDTO.setStatement(tableStatement);
IResult result = executeMSFlinkSql(studioMetaStoreDTO);
if (result instanceof DDLResult) {
DDLResult ddlResult = (DDLResult) result;
......@@ -437,8 +442,69 @@ public class StudioServiceImpl implements StudioService {
}
}
}
// show views
schema.setViews(showInfo(studioMetaStoreDTO, baseStatement, FlinkQuery.showViews()));
// show functions
schema.setFunctions(showInfo(studioMetaStoreDTO, baseStatement, FlinkQuery.showFunctions()));
// show user functions
schema.setUserFunctions(showInfo(studioMetaStoreDTO, baseStatement, FlinkQuery.showUserFunctions()));
// show modules
schema.setModules(showInfo(studioMetaStoreDTO, baseStatement, FlinkQuery.showModules()));
}
return tables;
schema.setTables(tables);
return schema;
}
@Override
public List<FlinkColumn> getMSFlinkColumns(StudioMetaStoreDTO studioMetaStoreDTO) {
List<FlinkColumn> columns = new ArrayList<>();
if (Dialect.isSql(studioMetaStoreDTO.getDialect())) {
// nothing to do
} else {
String baseStatement = FlinkQuery.useCatalog(studioMetaStoreDTO.getCatalog()) + FlinkQuery.separator() +
FlinkQuery.useDatabase(studioMetaStoreDTO.getDatabase()) + FlinkQuery.separator();
// desc tables
String tableStatement = baseStatement + FlinkQuery.descTable(studioMetaStoreDTO.getTable());
studioMetaStoreDTO.setStatement(tableStatement);
IResult result = executeMSFlinkSql(studioMetaStoreDTO);
if (result instanceof DDLResult) {
DDLResult ddlResult = (DDLResult) result;
List<Map<String, Object>> rowData = ddlResult.getRowData();
int i = 1;
for (Map<String, Object> item : rowData) {
FlinkColumn column = FlinkColumn.build(i,
item.get(FlinkQuery.columnName()).toString(),
item.get(FlinkQuery.columnType()).toString(),
item.get(FlinkQuery.columnKey()).toString(),
item.get(FlinkQuery.columnNull()).toString(),
item.get(FlinkQuery.columnExtras()).toString(),
item.get(FlinkQuery.columnWatermark()).toString()
);
columns.add(column);
i++;
}
}
}
return columns;
}
private List<String> showInfo(StudioMetaStoreDTO studioMetaStoreDTO, String baseStatement, String statement) {
List<String> infos = new ArrayList<>();
String tableStatement = baseStatement + statement;
studioMetaStoreDTO.setStatement(tableStatement);
IResult result = executeMSFlinkSql(studioMetaStoreDTO);
if (result instanceof DDLResult) {
DDLResult ddlResult = (DDLResult) result;
Iterator<String> iterator = ddlResult.getColumns().iterator();
if (iterator.hasNext()) {
String key = iterator.next();
List<Map<String, Object>> rowData = ddlResult.getRowData();
for (Map<String, Object> item : rowData) {
infos.add(item.get(key).toString());
}
}
}
return infos;
}
private void initUDF(JobConfig config, String statement) {
......
package com.dlink.sql;
/**
* FlinkQuery
*
* @author wenmo
* @since 2022/7/18 19:10
**/
public class FlinkQuery extends AbstractFlinkQuery {
}
package com.dlink.sql;
/**
* FlinkQuery
*
* @author wenmo
* @since 2022/7/18 19:10
**/
public class FlinkQuery extends AbstractFlinkQuery {
}
package com.dlink.sql;
/**
* FlinkQuery
*
* @author wenmo
* @since 2022/7/18 19:10
**/
public class FlinkQuery extends AbstractFlinkQuery {
}
package com.dlink.sql;
/**
* FlinkQuery
*
* @author wenmo
* @since 2022/7/18 19:10
**/
public class FlinkQuery extends AbstractFlinkQuery {
}
package com.dlink.sql;
/**
* FlinkQuery
*
* @author wenmo
* @since 2022/7/18 19:10
**/
public class FlinkQuery extends AbstractFlinkQuery {
}
package com.dlink.sql;
/**
* AbstractFlinkQuery
*
* @author wenmo
* @since 2022/7/18 18:43
**/
public abstract class AbstractFlinkQuery {
public static String separator() {
return ";\r\n";
}
public static String defaultCatalog() {
return "default_catalog";
}
public static String defaultDatabase() {
return "default_database";
}
public static String showCatalogs() {
return "SHOW CATALOGS";
}
public static String useCatalog(String catalog) {
return String.format("USE CATALOG %s", catalog);
}
public static String showDatabases() {
return "SHOW DATABASES";
}
public static String useDatabase(String database) {
return String.format("USE %s", database);
}
public static String showTables() {
return "SHOW TABLES";
}
public static String showViews() {
return "SHOW VIEWS";
}
public static String showFunctions() {
return "SHOW FUNCTIONS";
}
public static String showUserFunctions() {
return "SHOW USER FUNCTIONS";
}
public static String showModules() {
return "SHOW MODULES";
}
public static String descTable(String table) {
return String.format("DESC %s", table);
}
public static String columnName() {
return "name";
}
public static String columnType() {
return "type";
}
public static String columnNull() {
return "null";
}
public static String columnKey() {
return "key";
}
public static String columnExtras() {
return "extras";
}
public static String columnWatermark() {
return "watermark";
}
}
package com.dlink.model;
import java.io.Serializable;
import lombok.Getter;
import lombok.Setter;
/**
* FlinkColumn
*
* @author wenmo
* @since 2022/7/18 19:55
**/
@Getter
@Setter
public class FlinkColumn implements Serializable {
private static final long serialVersionUID = 4820196727157711974L;
private int position;
private String name;
private String type;
private String key;
private String nullable;
private String extras;
private String watermark;
public FlinkColumn() {
}
public FlinkColumn(int position, String name, String type, String key, String nullable, String extras, String watermark) {
this.position = position;
this.name = name;
this.type = type;
this.key = key;
this.nullable = nullable;
this.extras = extras;
this.watermark = watermark;
}
public static FlinkColumn build(int position, String name, String type, String key, String nullable, String extras, String watermark) {
return new FlinkColumn(position, name, type, key, nullable, extras, watermark);
}
}
......@@ -21,6 +21,10 @@ public class Schema implements Serializable, Comparable<Schema> {
private String name;
private List<Table> tables = new ArrayList<>();
private List<String> views = new ArrayList<>();
private List<String> functions = new ArrayList<>();
private List<String> userFunctions = new ArrayList<>();
private List<String> modules = new ArrayList<>();
public Schema(String name) {
this.name = name;
......
import {Button, Col, Empty, message, Modal, Row, Select, Tabs, Tooltip, Tree} from "antd";
import {MetaStoreTableType, StateType} from "@/pages/DataStudio/model";
import {MetaStoreDataBaseType, MetaStoreTableType, StateType} from "@/pages/DataStudio/model";
import {connect} from "umi";
import React, {useState} from "react";
import {
AppstoreOutlined,
BlockOutlined,
CodepenOutlined,
DownOutlined,
FunctionOutlined,
OrderedListOutlined,
ReloadOutlined,
TableOutlined
TableOutlined,
} from '@ant-design/icons';
import {Scrollbars} from 'react-custom-scrollbars';
import Columns from "@/pages/DataBase/Columns";
import Tables from "@/pages/DataBase/Tables";
import {TreeDataNode} from "@/components/Studio/StudioTree/Function";
import Generation from "@/pages/DataBase/Generation";
import {getMSTables} from "@/pages/DataStudio/service";
import {getMSSchemaInfo} from "@/pages/DataStudio/service";
import {Dispatch} from "@@/plugin-dva/connect";
import {DIALECT} from "@/components/Studio/conf";
import FlinkColumns from "@/pages/Flink/FlinkColumns";
const {DirectoryTree} = Tree;
const {Option, OptGroup} = Select;
......@@ -53,17 +58,18 @@ const StudioMetaStore = (props: any) => {
catalog: catalogTmp,
database: databaseTmp,
};
const result = getMSTables(param);
const result = getMSSchemaInfo(param);
result.then(res => {
const tables: MetaStoreTableType[] = [];
if (res.datas) {
for (let i = 0; i < res.datas.length; i++) {
if (res.datas.tables) {
for (let i = 0; i < res.datas.tables.length; i++) {
tables.push({
name: res.datas[i].name,
columns: res.datas[i].columns,
name: res.datas.tables[i].name,
columns: res.datas.tables[i].columns,
});
}
}
const treeDataTmp: [] = [];
const tablesData: [] = [];
for (let i = 0; i < tables.length; i++) {
tablesData.push({
......@@ -74,9 +80,111 @@ const StudioMetaStore = (props: any) => {
isLeaf: true,
catalog: catalogTmp,
database: databaseTmp,
isTable: true
})
}
setTreeData(tablesData);
treeDataTmp.push({
name: 'tables',
title: 'tables',
key: 'tables',
catalog: catalogTmp,
database: databaseTmp,
children: tablesData,
});
const viewsData: [] = [];
if (res.datas.views) {
for (let i = 0; i < res.datas.views.length; i++) {
viewsData.push({
name: res.datas.views[i],
title: res.datas.views[i],
key: res.datas.views[i],
icon: <BlockOutlined/>,
isLeaf: true,
catalog: catalogTmp,
database: databaseTmp,
});
}
}
treeDataTmp.push({
name: 'views',
title: 'views',
key: 'views',
catalog: catalogTmp,
database: databaseTmp,
children: viewsData,
});
const functionsData: [] = [];
if (res.datas.functions) {
for (let i = 0; i < res.datas.functions.length; i++) {
functionsData.push({
name: res.datas.functions[i],
title: res.datas.functions[i],
key: res.datas.functions[i],
icon: <FunctionOutlined/>,
isLeaf: true,
catalog: catalogTmp,
database: databaseTmp,
});
}
}
treeDataTmp.push({
name: 'functions',
title: 'functions',
key: 'functions',
catalog: catalogTmp,
database: databaseTmp,
children: functionsData,
});
const userFunctionsData: [] = [];
if (res.datas.userFunctions) {
for (let i = 0; i < res.datas.userFunctions.length; i++) {
userFunctionsData.push({
name: res.datas.userFunctions[i],
title: res.datas.userFunctions[i],
key: res.datas.userFunctions[i],
icon: <FunctionOutlined/>,
isLeaf: true,
catalog: catalogTmp,
database: databaseTmp,
});
}
}
treeDataTmp.push({
name: 'userFunctions',
title: 'user functions',
key: 'userFunctions',
catalog: catalogTmp,
database: databaseTmp,
children: userFunctionsData,
});
const modulesData: [] = [];
if (res.datas.modules) {
for (let i = 0; i < res.datas.modules.length; i++) {
modulesData.push({
name: res.datas.modules[i],
title: res.datas.modules[i],
key: res.datas.modules[i],
icon: <AppstoreOutlined/>,
isLeaf: true,
catalog: catalogTmp,
database: databaseTmp,
});
}
}
treeDataTmp.push({
name: 'modules',
title: 'modules',
key: 'modules',
catalog: catalogTmp,
database: databaseTmp,
children: modulesData,
});
setTreeData(treeDataTmp);
props.saveMetaStoreTable(current.key, catalogTmp, databaseTmp, tables);
message.success(`刷新 Catalog 成功`);
})
......@@ -105,16 +213,16 @@ const StudioMetaStore = (props: any) => {
};
const openColumnInfo = (e: React.MouseEvent, node: TreeDataNode) => {
if (node.isLeaf) {
if (node.isLeaf && node.isTable) {
setRow(node);
setModalVisit(true);
}
}
};
const cancelHandle = () => {
setRow(undefined);
setModalVisit(false);
}
};
return (
<>
......@@ -184,8 +292,12 @@ const StudioMetaStore = (props: any) => {
}
key="columnInfo"
>
{row ? <Columns dbId={current.task.databaseId} schema={row.database} table={row.name}/> :
<Empty image={Empty.PRESENTED_IMAGE_SIMPLE}/>}
{row ?
(current.task.dialect === DIALECT.FLINKSQL ?
<FlinkColumns envId={current.task.envId} catalog={row.catalog} database={row.database}
table={row.name}/>
: <Columns dbId={current.task.databaseId} schema={row.database} table={row.name}/>
) : <Empty image={Empty.PRESENTED_IMAGE_SIMPLE}/>}
</TabPane>
<TabPane
tab={
......@@ -196,8 +308,11 @@ const StudioMetaStore = (props: any) => {
}
key="sqlGeneration"
>
{row ? <Generation dbId={current.task.databaseId} schema={row.database} table={row.name}/> :
<Empty image={Empty.PRESENTED_IMAGE_SIMPLE}/>}
{row ?
(current.task.dialect === DIALECT.FLINKSQL ?
undefined
: <Generation dbId={current.task.databaseId} schema={row.database} table={row.name}/>
) : <Empty image={Empty.PRESENTED_IMAGE_SIMPLE}/>}
</TabPane>
</Tabs>
</Modal>
......
......@@ -144,6 +144,10 @@ export type MetaStoreCatalogType = {
export type MetaStoreDataBaseType = {
name: string;
tables: MetaStoreTableType[];
views: string[];
functions: string[];
userFunctions: string[];
modules: string[];
}
export type MetaStoreTableType = {
......
......@@ -82,8 +82,8 @@ export async function getMSCatalogs(params: StudioMetaStoreParam) {
});
}
export async function getMSTables(params: StudioMetaStoreParam) {
return request<API.Result>('/api/studio/getMSTables', {
export async function getMSSchemaInfo(params: StudioMetaStoreParam) {
return request<API.Result>('/api/studio/getMSSchemaInfo', {
method: 'POST',
data: {
...params,
......
import React from "react";
import {CheckSquareOutlined, KeyOutlined} from '@ant-design/icons';
import DTable from "@/components/Common/DTable";
import {DIALECT} from "@/components/Studio/conf";
const FlinkColumns = (props: any) => {
const {envId, catalog, database, table} = props;
const cols = [{
title: '序号',
dataIndex: 'position',
isString: false,
},
{
title: '列名',
dataIndex: 'name',
copyable: true,
},
{
title: '类型',
dataIndex: 'type',
},
{
title: '主键',
dataIndex: 'key',
render: (_, record) => (
<>
{record.key ? <KeyOutlined style={{color: '#FAA100'}}/> : undefined}
</>
),
filters: [
{
text: '主键',
value: true,
},
{
text: '其他',
value: '',
},
],
openSearch: 'dict',
}, {
title: '可为空',
dataIndex: 'nullable',
render: (_, record) => (
<>
{record.nullable ? <CheckSquareOutlined style={{color: '#1296db'}}/> : undefined}
</>
),
filters: [
{
text: '可为空',
value: true,
},
{
text: '非空',
value: '',
},
],
openSearch: 'dict',
}, {
title: '扩展',
dataIndex: 'extras',
}, {
title: '水印',
dataIndex: 'watermark',
},];
return (
<DTable columns={cols}
dataSource={{
url: 'api/studio/getMSFlinkColumns', params: {
envId,
dialect: DIALECT.FLINKSQL,
catalog,
database,
table
}
}}/>
)
};
export default FlinkColumns;
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment