Skip to content

D
dlink
Unverified Commit f8c751af authored by Kerwin's avatar Kerwin Committed by GitHub
Browse files
Options

[Optimization][Style] Added dlink-core module code style (#917) 

* Added dlink-core module code style.

* fix AbstractTrans error
parent a95ce812
Show whitespace changes
Inline Side-by-side
Showing with 415 additions and 383 deletions
dlink-core/src/main/java/com/dlink/api/FlinkAPI.java
View file @ f8c751af
......@@ -17,11 +17,8 @@
*
*/
package com.dlink.api;
import cn.hutool.http.HttpUtil;
import cn.hutool.http.Method;
import com.dlink.assertion.Asserts;
import com.dlink.constant.FlinkRestAPIConstant;
import com.dlink.constant.NetConstant;
......@@ -29,11 +26,19 @@ import com.dlink.gateway.GatewayType;
import com.dlink.gateway.config.SavePointType;
import com.dlink.gateway.model.JobInfo;
import com.dlink.gateway.result.SavePointResult;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import java.util.*;
import cn.hutool.http.HttpUtil;
import cn.hutool.http.Method;
/**
* FlinkAPI
......@@ -238,6 +243,7 @@ public class FlinkAPI {
public String getJobManagerStdOut() {
return getResult(FlinkRestAPIConstant.JOB_MANAGER + FlinkRestAPIConstant.STDOUT);
}
/**
* @Author: zhumingye
* @date: 2022/6/24
......@@ -247,6 +253,7 @@ public class FlinkAPI {
public JsonNode getJobManagerLogList() {
return get(FlinkRestAPIConstant.JOB_MANAGER + FlinkRestAPIConstant.LOGS);
}
/**
* @Author: zhumingye
* @date: 2022/6/24
......@@ -257,6 +264,7 @@ public class FlinkAPI {
public String getJobManagerLogFileDetail(String logName) {
return getResult(FlinkRestAPIConstant.JOB_MANAGER + FlinkRestAPIConstant.LOGS + logName);
}
/**
* @Author: zhumingye
* @date: 2022/6/24
......@@ -278,12 +286,13 @@ public class FlinkAPI {
JsonNode jsonNode = get(type + FlinkRestAPIConstant.METRICS);
StringBuilder sb = new StringBuilder();
Iterator<JsonNode> jsonNodeIterator = jsonNode.elements();
while(jsonNodeIterator.hasNext()) {
while (jsonNodeIterator.hasNext()) {
JsonNode node = jsonNodeIterator.next();
sb.append(node.get("id").asText()).append(",");
}
return sb.deleteCharAt(sb.length() - 1).toString();
}
/**
* @Author: zhumingye
* @date: 2022/6/24
......@@ -291,9 +300,9 @@ public class FlinkAPI {
* @return JsonNode
*/
public JsonNode getTaskManagerMetrics(String containerId) {
JsonNode TaskManagerMetricsJsonNode = get(FlinkRestAPIConstant.TASK_MANAGER + containerId + FlinkRestAPIConstant.METRICS + FlinkRestAPIConstant.GET + buildMetricsParms(FlinkRestAPIConstant.JOB_MANAGER));
return TaskManagerMetricsJsonNode;
return get(FlinkRestAPIConstant.TASK_MANAGER + containerId + FlinkRestAPIConstant.METRICS + FlinkRestAPIConstant.GET + buildMetricsParms(FlinkRestAPIConstant.JOB_MANAGER));
}
/**
* @Author: zhumingye
* @date: 2022/6/24
......@@ -304,6 +313,7 @@ public class FlinkAPI {
public String getTaskManagerLog(String containerId) {
return getResult(FlinkRestAPIConstant.TASK_MANAGER + containerId + FlinkRestAPIConstant.LOG);
}
/**
* @Author: zhumingye
* @date: 2022/6/24
......@@ -314,6 +324,7 @@ public class FlinkAPI {
public String getTaskManagerStdOut(String containerId) {
return getResult(FlinkRestAPIConstant.TASK_MANAGER + containerId + FlinkRestAPIConstant.STDOUT);
}
/**
* @Author: zhumingye
* @date: 2022/6/24
......
dlink-core/src/main/java/com/dlink/cluster/FlinkCluster.java
View file @ f8c751af
......@@ -17,15 +17,16 @@
*
*/
package com.dlink.cluster;
import cn.hutool.core.io.IORuntimeException;
import com.dlink.api.FlinkAPI;
import com.dlink.assertion.Asserts;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import cn.hutool.core.io.IORuntimeException;
/**
* FlinkCluster
*
......
dlink-core/src/main/java/com/dlink/cluster/FlinkClusterInfo.java
View file @ f8c751af
......@@ -17,7 +17,6 @@
*
*/
package com.dlink.cluster;
import lombok.Getter;
......
dlink-core/src/main/java/com/dlink/config/Dialect.java
View file @ f8c751af
......@@ -17,7 +17,6 @@
*
*/
package com.dlink.config;
import com.dlink.assertion.Asserts;
......
dlink-core/src/main/java/com/dlink/constant/FlinkHistoryConstant.java
View file @ f8c751af
......@@ -17,7 +17,6 @@
*
*/
package com.dlink.constant;
public interface FlinkHistoryConstant {
......
dlink-core/src/main/java/com/dlink/constant/FlinkRestAPIConstant.java
View file @ f8c751af
......@@ -17,7 +17,6 @@
*
*/
package com.dlink.constant;
/**
......
dlink-core/src/main/java/com/dlink/constant/FlinkRestResultConstant.java
View file @ f8c751af
......@@ -17,7 +17,6 @@
*
*/
package com.dlink.constant;
/**
......
dlink-core/src/main/java/com/dlink/explainer/Explainer.java
View file @ f8c751af
......@@ -17,13 +17,17 @@
*
*/
package com.dlink.explainer;
import com.dlink.assertion.Asserts;
import com.dlink.constant.FlinkSQLConstant;
import com.dlink.executor.Executor;
import com.dlink.explainer.ca.*;
import com.dlink.explainer.ca.ColumnCA;
import com.dlink.explainer.ca.ColumnCAResult;
import com.dlink.explainer.ca.NodeRel;
import com.dlink.explainer.ca.TableCA;
import com.dlink.explainer.ca.TableCAGenerator;
import com.dlink.explainer.ca.TableCAResult;
import com.dlink.explainer.lineage.LineageColumnGenerator;
import com.dlink.explainer.lineage.LineageTableGenerator;
import com.dlink.explainer.trans.Trans;
......@@ -39,8 +43,7 @@ import com.dlink.trans.Operations;
import com.dlink.utils.FlinkUtil;
import com.dlink.utils.LogUtil;
import com.dlink.utils.SqlUtil;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.node.ObjectNode;
import org.apache.flink.runtime.rest.messages.JobPlanInfo;
import org.apache.flink.table.catalog.CatalogManager;
......@@ -49,6 +52,9 @@ import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.node.ObjectNode;
/**
* Explainer
*
......@@ -276,7 +282,7 @@ public class Explainer {
public ObjectNode getStreamGraph(String statement) {
JobParam jobParam = pretreatStatements(SqlUtil.getStatements(statement, sqlSeparator));
if (jobParam.getDdl().size() > 0) {
for(StatementParam statementParam: jobParam.getDdl()){
for (StatementParam statementParam: jobParam.getDdl()) {
executor.executeSql(statementParam.getValue());
}
}
......@@ -296,7 +302,7 @@ public class Explainer {
public JobPlanInfo getJobPlanInfo(String statement) {
JobParam jobParam = pretreatStatements(SqlUtil.getStatements(statement, sqlSeparator));
if (jobParam.getDdl().size() > 0) {
for(StatementParam statementParam: jobParam.getDdl()){
for (StatementParam statementParam: jobParam.getDdl()) {
executor.executeSql(statementParam.getValue());
}
}
......@@ -453,11 +459,12 @@ public class Explainer {
for (NodeRel nodeRel : columnCAResult.getColumnCASRelChain()) {
if (nodeRel.getPreId().equals(item.getValue().getId())) {
for (NodeRel nodeRel2 : columnCAResult.getColumnCASRelChain()) {
if (columnCAResult.getColumnCASMaps().containsKey(nodeRel2.getSufId()) && columnCAResult.getColumnCASMaps().containsKey(nodeRel2.getPreId()) &&
columnCAResult.getColumnCASMaps().containsKey(nodeRel.getSufId()) &&
columnCAResult.getColumnCASMaps().get(nodeRel2.getSufId()).getTableId().equals(columnCAResult.getColumnCASMaps().get(nodeRel.getSufId()).getTableId()) &&
columnCAResult.getColumnCASMaps().get(nodeRel2.getSufId()).getName().equals(columnCAResult.getColumnCASMaps().get(nodeRel.getSufId()).getName()) &&
!columnCAResult.getColumnCASMaps().get(nodeRel2.getPreId()).getType().equals("Data Sink")) {
if (columnCAResult.getColumnCASMaps().containsKey(nodeRel2.getSufId())
&& columnCAResult.getColumnCASMaps().containsKey(nodeRel2.getPreId())
&& columnCAResult.getColumnCASMaps().containsKey(nodeRel.getSufId())
&& columnCAResult.getColumnCASMaps().get(nodeRel2.getSufId()).getTableId().equals(columnCAResult.getColumnCASMaps().get(nodeRel.getSufId()).getTableId())
&& columnCAResult.getColumnCASMaps().get(nodeRel2.getSufId()).getName().equals(columnCAResult.getColumnCASMaps().get(nodeRel.getSufId()).getName())
&& !columnCAResult.getColumnCASMaps().get(nodeRel2.getPreId()).getType().equals("Data Sink")) {
addNodeRels.add(new NodeRel(nodeRel2.getPreId(), nodeRel.getPreId()));
}
}
......
dlink-core/src/main/java/com/dlink/explainer/ca/CABuilder.java
View file @ f8c751af
......@@ -17,7 +17,6 @@
*
*/
package com.dlink.explainer.ca;
import com.dlink.plus.FlinkSqlPlus;
......@@ -104,7 +103,7 @@ public class CABuilder {
for (NodeRel nodeRel : columnCASRel) {
if (columnId.equals(nodeRel.getSufId())) {
ColumnCA childca = (ColumnCA) result.getColumnCASMaps().get(nodeRel.getPreId());
// operation = operation.replaceAll(childca.getAlias().replaceAll("\\$","\\\\$"),childca.getOperation());
//operation = operation.replaceAll(childca.getAlias().replaceAll("\\$","\\\\$"),childca.getOperation());
operation = operation.replaceAll(childca.getAlias()
.replaceAll("\\)", ""), childca.getOperation());
buildColumnCANodeChildren(children, result, nodeRel.getPreId(), operation);
......
dlink-core/src/main/java/com/dlink/explainer/ca/CAGenerator.java
View file @ f8c751af
......@@ -17,7 +17,6 @@
*
*/
package com.dlink.explainer.ca;
public interface CAGenerator {
......
dlink-core/src/main/java/com/dlink/explainer/ca/ColumnCA.java
View file @ f8c751af
......@@ -17,7 +17,6 @@
*
*/
package com.dlink.explainer.ca;
import java.util.List;
......@@ -55,17 +54,6 @@ public class ColumnCA implements ICA {
this.type = tableCA.getType();
}
/* public ColumnCA(Integer id, List<Integer> parentId, String name, String alias, String columnName, String familyName, String type, TableCA tableCA) {
this.id = id;
this.parentId = parentId;
this.name = name;
this.alias = alias;
this.columnName = columnName;
this.familyName = familyName;
this.type = type;
this.tableCA = tableCA;
}*/
public Integer getId() {
return id;
}
......
dlink-core/src/main/java/com/dlink/explainer/ca/ColumnCAGenerator.java
View file @ f8c751af
......@@ -17,16 +17,25 @@
*
*/
package com.dlink.explainer.ca;
import com.dlink.assertion.Asserts;
import com.dlink.explainer.trans.*;
import com.dlink.explainer.trans.Field;
import com.dlink.explainer.trans.OperatorTrans;
import com.dlink.explainer.trans.SinkTrans;
import com.dlink.explainer.trans.SourceTrans;
import com.dlink.explainer.trans.Trans;
import com.dlink.utils.MapParseUtils;
import org.apache.commons.collections.CollectionUtils;
import java.util.*;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
/**
* ColumnCAGenerator
......@@ -176,7 +185,7 @@ public class ColumnCAGenerator implements CAGenerator {
}
if (!isHad) {
cid = index++;
// String columnOperation = MapParseUtils.replaceField(operation,columnCA.getAlias(),columnCA.getOperation());
//String columnOperation = MapParseUtils.replaceField(operation,columnCA.getAlias(),columnCA.getOperation());
ColumnCA columnCA2 = new ColumnCA(cid, alias, alias, alias, alias, operation, tableCA);
this.columnCASMaps.put(cid, columnCA2);
this.columnCAS.add(columnCA2);
......
dlink-core/src/main/java/com/dlink/explainer/ca/ColumnCANode.java
View file @ f8c751af
......@@ -17,15 +17,14 @@
*
*/
package com.dlink.explainer.ca;
import lombok.Getter;
import lombok.Setter;
import java.io.Serializable;
import java.util.List;
import lombok.Getter;
import lombok.Setter;
/**
* ColumnCANode
*
......@@ -43,12 +42,9 @@ public class ColumnCANode implements Serializable {
private String value;
private String type;
private String operation;
// private Tables tables;
// private Columns columns;
private List<ColumnCANode> children;
public ColumnCANode() {
}
}
dlink-core/src/main/java/com/dlink/explainer/ca/ColumnCAResult.java
View file @ f8c751af
......@@ -17,7 +17,6 @@
*
*/
package com.dlink.explainer.ca;
import com.dlink.explainer.lineage.LineageColumnGenerator;
......
dlink-core/src/main/java/com/dlink/explainer/ca/ICA.java
View file @ f8c751af
......@@ -17,7 +17,6 @@
*
*/
package com.dlink.explainer.ca;
public interface ICA {
......
dlink-core/src/main/java/com/dlink/explainer/ca/NodeRel.java
View file @ f8c751af
......@@ -17,7 +17,6 @@
*
*/
package com.dlink.explainer.ca;
import java.util.Objects;
......@@ -55,11 +54,15 @@ public class NodeRel {
@Override
public boolean equals(Object o) {
if (this == o) return true;
if (o == null || getClass() != o.getClass()) return false;
if (this == o) {
return true;
}
if (o == null || getClass() != o.getClass()) {
return false;
}
NodeRel nodeRel = (NodeRel) o;
return Objects.equals(preId, nodeRel.preId) &&
Objects.equals(sufId, nodeRel.sufId);
return Objects.equals(preId, nodeRel.preId)
&& Objects.equals(sufId, nodeRel.sufId);
}
@Override
......
dlink-core/src/main/java/com/dlink/explainer/ca/TableCA.java
View file @ f8c751af
......@@ -17,19 +17,23 @@
*
*/
package com.dlink.explainer.ca;
import com.dlink.assertion.Asserts;
import com.dlink.explainer.trans.*;
import lombok.Getter;
import lombok.Setter;
import com.dlink.explainer.trans.Field;
import com.dlink.explainer.trans.OperatorTrans;
import com.dlink.explainer.trans.SinkTrans;
import com.dlink.explainer.trans.SourceTrans;
import com.dlink.explainer.trans.Trans;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import lombok.Getter;
import lombok.Setter;
/**
* TableCA
*
......@@ -135,17 +139,17 @@ public class TableCA implements ICA {
@Override
public String toString() {
return "TableCA{" +
"id=" + id +
", parentId=" + parentId +
", name='" + name + '\'' +
", catalog='" + catalog + '\'' +
", database='" + database + '\'' +
", table='" + table + '\'' +
", fields=" + fields +
", useFields=" + useFields +
", parallelism=" + parallelism +
'}';
return "TableCA{"
+ "id=" + id
+ ", parentId=" + parentId
+ ", name='" + name + '\''
+ ", catalog='" + catalog + '\''
+ ", database='" + database + '\''
+ ", table='" + table + '\''
+ ", fields=" + fields
+ ", useFields=" + useFields
+ ", parallelism=" + parallelism
+ '}';
}
@Override
......
dlink-core/src/main/java/com/dlink/explainer/ca/TableCAGenerator.java
View file @ f8c751af
......@@ -17,14 +17,24 @@
*
*/
package com.dlink.explainer.ca;
import com.dlink.assertion.Asserts;
import com.dlink.explainer.trans.*;
import com.dlink.explainer.trans.Field;
import com.dlink.explainer.trans.OperatorTrans;
import com.dlink.explainer.trans.SinkTrans;
import com.dlink.explainer.trans.SourceTrans;
import com.dlink.explainer.trans.Trans;
import org.apache.commons.collections.CollectionUtils;
import java.util.*;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
......
dlink-core/src/main/java/com/dlink/explainer/ca/TableCANode.java
View file @ f8c751af
......@@ -17,15 +17,14 @@
*
*/
package com.dlink.explainer.ca;
import lombok.Getter;
import lombok.Setter;
import java.io.Serializable;
import java.util.List;
import lombok.Getter;
import lombok.Setter;
/**
* TableCANode
*
......
dlink-core/src/main/java/com/dlink/explainer/ca/TableCAResult.java
View file @ f8c751af
......@@ -17,7 +17,6 @@
*
*/
package com.dlink.explainer.ca;
import java.util.List;
......
dlink-core/src/main/java/com/dlink/explainer/lineage/LineageBuilder.java
View file @ f8c751af
......@@ -17,7 +17,6 @@
*
*/
package com.dlink.explainer.lineage;
import com.dlink.explainer.ca.ColumnCAResult;
......@@ -25,7 +24,12 @@ import com.dlink.explainer.ca.NodeRel;
import com.dlink.explainer.ca.TableCA;
import com.dlink.plus.FlinkSqlPlus;
import java.util.*;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
/**
* LineageBuilder
......@@ -51,14 +55,15 @@ public class LineageBuilder {
}
Set<String> keySet = new HashSet<>();
for (NodeRel nodeRel : item.getColumnCASRelChain()) {
if( item.getColumnCASMaps().containsKey(nodeRel.getPreId())&&item.getColumnCASMaps().containsKey(nodeRel.getSufId())
if (item.getColumnCASMaps().containsKey(nodeRel.getPreId())
&& item.getColumnCASMaps().containsKey(nodeRel.getSufId())
&& !item.getColumnCASMaps().get(nodeRel.getPreId()).getTableId().equals(item.getColumnCASMaps().get(nodeRel.getSufId()).getTableId())) {
String key = item.getColumnCASMaps().get(nodeRel.getPreId()).getTableId().toString() + "@" +
item.getColumnCASMaps().get(nodeRel.getSufId()).getTableId().toString() + "@" +
item.getColumnCASMaps().get(nodeRel.getPreId()).getName() + "@" +
item.getColumnCASMaps().get(nodeRel.getSufId()).getName();
String key = item.getColumnCASMaps().get(nodeRel.getPreId()).getTableId().toString() + "@"
+ item.getColumnCASMaps().get(nodeRel.getSufId()).getTableId().toString() + "@"
+ item.getColumnCASMaps().get(nodeRel.getPreId()).getName() + "@"
+ item.getColumnCASMaps().get(nodeRel.getSufId()).getName();
//去重
if(!keySet.contains(key)){
if (!keySet.contains(key)) {
index++;
relations.add(LineageRelation.build(index + "",
item.getColumnCASMaps().get(nodeRel.getPreId()).getTableId().toString(),
......@@ -87,18 +92,18 @@ public class LineageBuilder {
}
//重复表合并
Map<String,String> correctTableIdMap = new HashMap<>();
for(List<LineageTable> tableList : repeatTablesList){
for (List<LineageTable> tableList : repeatTablesList) {
LineageTable newTable = new LineageTable();
Set<String> columnKeySet = new HashSet<>();
for(LineageTable table: tableList){
if(newTable.getId() == null || newTable.getName() == null){
for (LineageTable table: tableList) {
if (newTable.getId() == null || newTable.getName() == null) {
newTable.setId(table.getId());
newTable.setName(table.getName());
newTable.setColumns(new ArrayList<>());
}
for(LineageColumn column : table.getColumns()){
for (LineageColumn column : table.getColumns()) {
String key = column.getName() + "@&" + column.getTitle();
if(!columnKeySet.contains(key)){
if (!columnKeySet.contains(key)) {
newTable.getColumns().add(column);
columnKeySet.add(key);
}
......@@ -109,11 +114,11 @@ public class LineageBuilder {
tables.add(newTable);
}
//关系中id重新指向
for (LineageRelation relation : relations){
if(correctTableIdMap.containsKey(relation.getSrcTableId())){
for (LineageRelation relation : relations) {
if (correctTableIdMap.containsKey(relation.getSrcTableId())) {
relation.setSrcTableId(correctTableIdMap.get(relation.getSrcTableId()));
}
if(correctTableIdMap.containsKey(relation.getTgtTableId())){
if (correctTableIdMap.containsKey(relation.getTgtTableId())) {
relation.setTgtTableId(correctTableIdMap.get(relation.getTgtTableId()));
}
}
......
dlink-core/src/main/java/com/dlink/explainer/lineage/LineageColumn.java
View file @ f8c751af
......@@ -17,7 +17,6 @@
*
*/
package com.dlink.explainer.lineage;
/**
......
dlink-core/src/main/java/com/dlink/explainer/lineage/LineageColumnGenerator.java
View file @ f8c751af
......@@ -17,16 +17,8 @@
*
*/
package com.dlink.explainer.lineage;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import com.dlink.assertion.Asserts;
import com.dlink.explainer.ca.ColumnCA;
import com.dlink.explainer.ca.NodeRel;
......@@ -38,6 +30,13 @@ import com.dlink.explainer.trans.SinkTrans;
import com.dlink.explainer.trans.Trans;
import com.dlink.utils.MapParseUtils;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
/**
* LineageColumnGenerator
*
......
dlink-core/src/main/java/com/dlink/explainer/lineage/LineageRelation.java
View file @ f8c751af
......@@ -17,7 +17,6 @@
*
*/
package com.dlink.explainer.lineage;
/**
......
dlink-core/src/main/java/com/dlink/explainer/lineage/LineageResult.java
View file @ f8c751af
......@@ -17,7 +17,6 @@
*
*/
package com.dlink.explainer.lineage;
import java.util.List;
......
dlink-core/src/main/java/com/dlink/explainer/lineage/LineageTable.java
View file @ f8c751af
......@@ -17,7 +17,6 @@
*
*/
package com.dlink.explainer.lineage;
import com.dlink.explainer.ca.TableCA;
......
dlink-core/src/main/java/com/dlink/explainer/lineage/LineageTableGenerator.java
View file @ f8c751af
......@@ -17,7 +17,6 @@
*
*/
package com.dlink.explainer.lineage;
import com.dlink.assertion.Asserts;
......
dlink-core/src/main/java/com/dlink/explainer/sqlLineage/LineageBuilder.java
View file @ f8c751af
......@@ -17,16 +17,8 @@
*
*/
package com.dlink.explainer.sqlLineage;
import com.alibaba.druid.sql.SQLUtils;
import com.alibaba.druid.sql.ast.SQLExpr;
import com.alibaba.druid.sql.ast.SQLStatement;
import com.alibaba.druid.sql.ast.expr.SQLIdentifierExpr;
import com.alibaba.druid.sql.ast.expr.SQLPropertyExpr;
import com.alibaba.druid.sql.ast.statement.SQLInsertStatement;
import com.alibaba.druid.stat.TableStat;
import com.dlink.assertion.Asserts;
import com.dlink.explainer.lineage.LineageRelation;
import com.dlink.explainer.lineage.LineageResult;
......@@ -35,10 +27,23 @@ import com.dlink.metadata.driver.Driver;
import com.dlink.metadata.driver.DriverConfig;
import com.dlink.model.Column;
import java.util.*;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import com.alibaba.druid.sql.SQLUtils;
import com.alibaba.druid.sql.ast.SQLExpr;
import com.alibaba.druid.sql.ast.SQLStatement;
import com.alibaba.druid.sql.ast.expr.SQLIdentifierExpr;
import com.alibaba.druid.sql.ast.expr.SQLPropertyExpr;
import com.alibaba.druid.sql.ast.statement.SQLInsertStatement;
import com.alibaba.druid.stat.TableStat;
public class LineageBuilder {
public static LineageResult getSqlLineageByOne(String statement ,String type) {
public static LineageResult getSqlLineageByOne(String statement, String type) {
List<LineageTable> tables = new ArrayList<>();
List<LineageRelation> relations = new ArrayList<>();
try {
......@@ -166,7 +171,7 @@ public class LineageBuilder {
//处理target表中字段
if (columns.size() <= 0 || sqls[n].contains("*")) {
Driver driver = Driver.build(driverConfig);
if(!targetTable.contains(".")){
if (!targetTable.contains(".")) {
return null;
}
List<Column> columns1 = driver.listColumns(targetTable.split("\\.")[0], targetTable.split("\\.")[1]);
......
dlink-core/src/main/java/com/dlink/explainer/sqlLineage/LineageColumn.java
View file @ f8c751af
......@@ -17,10 +17,10 @@
*
*/
package com.dlink.explainer.sqlLineage;
import com.dlink.assertion.Asserts;
import lombok.Data;
@Data
......@@ -83,16 +83,15 @@ public class LineageColumn implements Comparable<LineageColumn> {
private Boolean isEnd = false;
public void setSourceTableName(String sourceTableName) {
sourceTableName = Asserts.isNotNullString(sourceTableName) ? sourceTableName.replace("`",
"") : sourceTableName;
if(sourceTableName.contains(" ")){
sourceTableName = Asserts.isNotNullString(sourceTableName) ? sourceTableName.replace("`", "") : sourceTableName;
if (sourceTableName.contains(" ")) {
sourceTableName = sourceTableName.substring(0,sourceTableName.indexOf(" "));
}
if (sourceTableName.contains(".")) {
if(Asserts.isNullString(this.sourceDbName)){
if (Asserts.isNullString(this.sourceDbName)) {
this.sourceDbName = sourceTableName.substring(0, sourceTableName.indexOf("."));
}
// this.sourceDbName = sourceTableName.substring(0, sourceTableName.indexOf("."));
//this.sourceDbName = sourceTableName.substring(0, sourceTableName.indexOf("."));
this.sourceTableName = sourceTableName.substring(sourceTableName.indexOf(".") + 1);
} else {
this.sourceTableName = sourceTableName;
......@@ -100,12 +99,16 @@ public class LineageColumn implements Comparable<LineageColumn> {
}
public int compareTo(LineageColumn o) {
if(Asserts.isNotNullString(this.getSourceDbName())&& Asserts.isNotNullString(this.getSourceTableName())){
if(this.getSourceDbName().equals(o.getSourceDbName())&&this.getSourceTableName().equals(o.getSourceTableName())&&this.getTargetColumnName().equals(o.getTargetColumnName())){
if (Asserts.isNotNullString(this.getSourceDbName())
&& Asserts.isNotNullString(this.getSourceTableName())) {
if (this.getSourceDbName().equals(o.getSourceDbName())
&& this.getSourceTableName().equals(o.getSourceTableName())
&& this.getTargetColumnName().equals(o.getTargetColumnName())) {
return 0;
}
} else if(Asserts.isNotNullString(this.getSourceTableName())){
if(this.getSourceTableName().equals(o.getSourceTableName())&&this.getTargetColumnName().equals(o.getTargetColumnName())){
} else if (Asserts.isNotNullString(this.getSourceTableName())) {
if (this.getSourceTableName().equals(o.getSourceTableName())
&& this.getTargetColumnName().equals(o.getTargetColumnName())) {
return 0;
}
} else {
......
dlink-core/src/main/java/com/dlink/explainer/sqlLineage/LineageUtils.java
View file @ f8c751af
......@@ -17,20 +17,37 @@
*
*/
package com.dlink.explainer.sqlLineage;
import com.alibaba.druid.sql.SQLUtils;
import com.alibaba.druid.sql.ast.SQLExpr;
import com.alibaba.druid.sql.ast.SQLStatement;
import com.alibaba.druid.sql.ast.expr.*;
import com.alibaba.druid.sql.ast.statement.*;
import com.dlink.assertion.Asserts;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.atomic.AtomicReference;
import com.alibaba.druid.sql.SQLUtils;
import com.alibaba.druid.sql.ast.SQLExpr;
import com.alibaba.druid.sql.ast.SQLStatement;
import com.alibaba.druid.sql.ast.expr.SQLAggregateExpr;
import com.alibaba.druid.sql.ast.expr.SQLBinaryOpExpr;
import com.alibaba.druid.sql.ast.expr.SQLCaseExpr;
import com.alibaba.druid.sql.ast.expr.SQLCharExpr;
import com.alibaba.druid.sql.ast.expr.SQLIdentifierExpr;
import com.alibaba.druid.sql.ast.expr.SQLIntegerExpr;
import com.alibaba.druid.sql.ast.expr.SQLMethodInvokeExpr;
import com.alibaba.druid.sql.ast.expr.SQLNumberExpr;
import com.alibaba.druid.sql.ast.expr.SQLPropertyExpr;
import com.alibaba.druid.sql.ast.statement.SQLExprTableSource;
import com.alibaba.druid.sql.ast.statement.SQLJoinTableSource;
import com.alibaba.druid.sql.ast.statement.SQLSelectItem;
import com.alibaba.druid.sql.ast.statement.SQLSelectQuery;
import com.alibaba.druid.sql.ast.statement.SQLSelectQueryBlock;
import com.alibaba.druid.sql.ast.statement.SQLSelectStatement;
import com.alibaba.druid.sql.ast.statement.SQLSubqueryTableSource;
import com.alibaba.druid.sql.ast.statement.SQLTableSource;
import com.alibaba.druid.sql.ast.statement.SQLUnionQuery;
import com.alibaba.druid.sql.ast.statement.SQLUnionQueryTableSource;
public class LineageUtils {
public static void columnLineageAnalyzer(String sql, String type, TreeNode<LineageColumn> node) {
......@@ -59,8 +76,7 @@ public class LineageUtils {
SQLSelectQueryBlock sqlSelectQueryBlock = (SQLSelectQueryBlock) sqlSelectQuery;
// 获取字段列表
List<SQLSelectItem> selectItems = sqlSelectQueryBlock.getSelectList();
selectItems.forEach(x ->
{
selectItems.forEach(x -> {
// 处理---------------------
String column = Asserts.isNullString(x.getAlias()) ? x.toString() : x.getAlias();
......@@ -122,8 +138,7 @@ public class LineageUtils {
* @param table
*/
private static void handlerSQLUnionQueryTableSource(TreeNode<LineageColumn> node, SQLUnionQueryTableSource table, String type) {
node.getAllLeafs().stream().filter(e -> !e.getData().getIsEnd()).forEach(e ->
{
node.getAllLeafs().stream().filter(e -> !e.getData().getIsEnd()).forEach(e -> {
columnLineageAnalyzer(table.getUnion().toString(), type, e);
});
}
......@@ -135,8 +150,7 @@ public class LineageUtils {
* @param table
*/
private static void handlerSQLSubqueryTableSource(TreeNode<LineageColumn> node, SQLTableSource table, String type) {
node.getAllLeafs().stream().filter(e -> !e.getData().getIsEnd()).forEach(e ->
{
node.getAllLeafs().stream().filter(e -> !e.getData().getIsEnd()).forEach(e -> {
if (Asserts.isNotNullString(e.getData().getSourceTableName())) {
if (e.getData().getSourceTableName().equals(table.getAlias())) {
columnLineageAnalyzer(((SQLSubqueryTableSource) table).getSelect().toString(), type, e);
......@@ -157,8 +171,7 @@ public class LineageUtils {
SQLJoinTableSource table, String type) {
// 处理---------------------
// 子查询作为表
node.getAllLeafs().stream().filter(e -> !e.getData().getIsEnd()).forEach(e ->
{
node.getAllLeafs().stream().filter(e -> !e.getData().getIsEnd()).forEach(e -> {
if (table.getLeft() instanceof SQLJoinTableSource) {
handlerSQLJoinTableSource(node, (SQLJoinTableSource) table.getLeft(), type);
} else if (table.getLeft() instanceof SQLExprTableSource) {
......@@ -171,8 +184,7 @@ public class LineageUtils {
handlerSQLUnionQueryTableSource(node, (SQLUnionQueryTableSource) table.getLeft(), type);
}
});
node.getAllLeafs().stream().filter(e -> !e.getData().getIsEnd()).forEach(e ->
{
node.getAllLeafs().stream().filter(e -> !e.getData().getIsEnd()).forEach(e -> {
if (table.getRight() instanceof SQLJoinTableSource) {
handlerSQLJoinTableSource(node, (SQLJoinTableSource) table.getRight(), type);
} else if (table.getRight() instanceof SQLExprTableSource) {
......@@ -199,8 +211,7 @@ public class LineageUtils {
String tableName = tableSource.getExpr() instanceof SQLPropertyExpr ? ((
SQLPropertyExpr) tableSource.getExpr()).getName().replace("`", "").replace("\"", "") : "";
String alias = Asserts.isNotNullString(tableSource.getAlias()) ? tableSource.getAlias().replace("`", "").replace("\"", "") : "";
node.getChildren().forEach(e ->
{
node.getChildren().forEach(e -> {
e.getChildren().forEach(f -> {
if (!f.getData().getIsEnd() && (f.getData().getSourceTableName() == null || f.getData().getSourceTableName().equals(tableName) || f
.getData().getSourceTableName().equals(alias))) {
......@@ -271,8 +282,7 @@ public class LineageUtils {
node.getData().setIsEnd(true);
}
} else {
expr.getArguments().forEach(expr1 ->
{
expr.getArguments().forEach(expr1 -> {
handlerExpr(expr1, node);
});
}
......@@ -285,8 +295,7 @@ public class LineageUtils {
* @param node
*/
public static void visitSQLAggregateExpr(SQLAggregateExpr expr, TreeNode<LineageColumn> node) {
expr.getArguments().forEach(expr1 ->
{
expr.getArguments().forEach(expr1 -> {
handlerExpr(expr1, node);
});
}
......@@ -299,8 +308,7 @@ public class LineageUtils {
*/
public static void visitSQLCaseExpr(SQLCaseExpr expr, TreeNode<LineageColumn> node) {
handlerExpr(expr.getValueExpr(), node);
expr.getItems().forEach(expr1 ->
{
expr.getItems().forEach(expr1 -> {
handlerExpr(expr1.getValueExpr(), node);
});
handlerExpr(expr.getElseExpr(), node);
......
dlink-core/src/main/java/com/dlink/explainer/sqlLineage/TreeNode.java
View file @ f8c751af
......@@ -17,10 +17,13 @@
*
*/
package com.dlink.explainer.sqlLineage;
import java.util.*;
import java.util.HashSet;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Set;
public class TreeNode<T> implements Iterable<TreeNode<T>> {
......
dlink-core/src/main/java/com/dlink/explainer/sqlLineage/TreeNodeIterator.java
View file @ f8c751af
......@@ -17,9 +17,10 @@
*
*/
package com.dlink.explainer.sqlLineage;
import java.util.Iterator;
public class TreeNodeIterator<T> implements Iterator<TreeNode<T>> {
private ProcessStages doNext;
......
dlink-core/src/main/java/com/dlink/explainer/trans/AbstractTrans.java
View file @ f8c751af
......@@ -17,16 +17,15 @@
*
*/
package com.dlink.explainer.trans;
import com.fasterxml.jackson.databind.JsonNode;
import java.util.ArrayList;
import java.util.List;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import com.fasterxml.jackson.databind.JsonNode;
/**
* AbstractTrans
*
......@@ -45,7 +44,6 @@ public abstract class AbstractTrans {
protected Integer parallelism;
protected List<Predecessor> predecessors;
public void build(JsonNode node) {
id = node.get("id").asInt();
text = node.toPrettyString();
......
dlink-core/src/main/java/com/dlink/explainer/trans/Field.java
View file @ f8c751af
......@@ -17,7 +17,6 @@
*
*/
package com.dlink.explainer.trans;
/**
......
dlink-core/src/main/java/com/dlink/explainer/trans/OperatorTrans.java
View file @ f8c751af
......@@ -17,10 +17,10 @@
*
*/
package com.dlink.explainer.trans;
import com.dlink.utils.MapParseUtils;
import org.apache.commons.lang3.StringUtils;
import java.util.ArrayList;
......@@ -44,8 +44,8 @@ public class OperatorTrans extends AbstractTrans implements Trans {
private List<String> leftInputSpec;
private List<String> rightInputSpec;
public final static String TRANS_TYPE = "Operator";
private final static String FIELD_AS = " AS ";
public static final String TRANS_TYPE = "Operator";
private static final String FIELD_AS = " AS ";
public List<Field> getSelect() {
return select;
......@@ -89,7 +89,6 @@ public class OperatorTrans extends AbstractTrans implements Trans {
return TRANS_TYPE.equals(pact);
}
@Override
public void translate() {
name = pact;
......
dlink-core/src/main/java/com/dlink/explainer/trans/Predecessor.java
View file @ f8c751af
......@@ -17,7 +17,6 @@
*
*/
package com.dlink.explainer.trans;
import lombok.Getter;
......@@ -44,10 +43,10 @@ public class Predecessor {
@Override
public String toString() {
return "Predecessor{" +
"id=" + id +
", shipStrategy='" + shipStrategy + '\'' +
", side='" + side + '\'' +
'}';
return "Predecessor{"
+ "id=" + id
+ ", shipStrategy='" + shipStrategy + '\''
+ ", side='" + side + '\''
+ '}';
}
}
dlink-core/src/main/java/com/dlink/explainer/trans/SinkTrans.java
View file @ f8c751af
......@@ -17,10 +17,8 @@
*
*/
package com.dlink.explainer.trans;
import com.dlink.utils.MapParseUtils;
import java.util.ArrayList;
......@@ -40,7 +38,7 @@ public class SinkTrans extends AbstractTrans implements Trans {
private String table;
private List<String> fields;
public final static String TRANS_TYPE = "Data Sink";
public static final String TRANS_TYPE = "Data Sink";
public SinkTrans() {
}
......
dlink-core/src/main/java/com/dlink/explainer/trans/SourceTrans.java
View file @ f8c751af
......@@ -17,15 +17,19 @@
*
*/
package com.dlink.explainer.trans;
import com.dlink.utils.MapParseUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.table.operations.Operation;
import org.apache.flink.table.operations.OperationUtils;
import java.util.*;
import java.util.ArrayList;
import java.util.Collections;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
/**
* SourceTrans
......@@ -41,7 +45,7 @@ public class SourceTrans extends AbstractTrans implements Trans {
private List<String> project;
private List<String> fields;
public final static String TRANS_TYPE = "Data Source";
public static final String TRANS_TYPE = "Data Source";
public SourceTrans() {
}
......
dlink-core/src/main/java/com/dlink/explainer/trans/Trans.java
View file @ f8c751af
......@@ -17,13 +17,12 @@
*
*/
package com.dlink.explainer.trans;
import com.fasterxml.jackson.databind.JsonNode;
import java.util.List;
import com.fasterxml.jackson.databind.JsonNode;
/**
* Trans
*
......
dlink-core/src/main/java/com/dlink/explainer/trans/TransGenerator.java
View file @ f8c751af
......@@ -17,18 +17,18 @@
*
*/
package com.dlink.explainer.trans;
import com.dlink.assertion.Asserts;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.node.ObjectNode;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.node.ObjectNode;
/**
* TransGenerator
*
......
dlink-core/src/main/java/com/dlink/job/Job.java
View file @ f8c751af
......@@ -17,19 +17,19 @@
*
*/
package com.dlink.job;
import com.dlink.executor.Executor;
import com.dlink.executor.ExecutorSetting;
import com.dlink.gateway.GatewayType;
import com.dlink.result.IResult;
import lombok.Getter;
import lombok.Setter;
import java.time.LocalDateTime;
import java.util.List;
import lombok.Getter;
import lombok.Setter;
/**
* Job
*
......
dlink-core/src/main/java/com/dlink/job/JobConfig.java
View file @ f8c751af
......@@ -17,12 +17,8 @@
*
*/
package com.dlink.job;
import java.util.List;
import java.util.Map;
import com.dlink.assertion.Asserts;
import com.dlink.executor.ExecutorSetting;
import com.dlink.gateway.GatewayType;
......@@ -33,6 +29,9 @@ import com.dlink.gateway.config.GatewayConfig;
import com.dlink.gateway.config.SavePointStrategy;
import com.dlink.session.SessionConfig;
import java.util.List;
import java.util.Map;
import lombok.Getter;
import lombok.Setter;
......
dlink-core/src/main/java/com/dlink/job/JobContextHolder.java
View file @ f8c751af
......@@ -17,7 +17,6 @@
*
*/
package com.dlink.job;
/**
......
dlink-core/src/main/java/com/dlink/job/JobHandler.java
View file @ f8c751af
......@@ -17,7 +17,6 @@
*
*/
package com.dlink.job;
import com.dlink.exception.JobException;
......
dlink-core/src/main/java/com/dlink/job/JobManager.java
View file @ f8c751af
......@@ -17,32 +17,8 @@
*
*/
package com.dlink.job;
import org.apache.flink.configuration.CoreOptions;
import org.apache.flink.configuration.DeploymentOptions;
import org.apache.flink.configuration.PipelineOptions;
import org.apache.flink.core.execution.JobClient;
import org.apache.flink.runtime.jobgraph.JobGraph;
import org.apache.flink.runtime.jobgraph.SavepointConfigOptions;
import org.apache.flink.runtime.jobgraph.SavepointRestoreSettings;
import org.apache.flink.streaming.api.environment.ExecutionCheckpointingOptions;
import org.apache.flink.streaming.api.graph.StreamGraph;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.yarn.configuration.YarnConfigOptions;
import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.dlink.api.FlinkAPI;
import com.dlink.assertion.Asserts;
import com.dlink.constant.FlinkSQLConstant;
......@@ -79,6 +55,30 @@ import com.dlink.trans.Operations;
import com.dlink.utils.LogUtil;
import com.dlink.utils.SqlUtil;
import com.dlink.utils.UDFUtil;
import org.apache.flink.configuration.CoreOptions;
import org.apache.flink.configuration.DeploymentOptions;
import org.apache.flink.configuration.PipelineOptions;
import org.apache.flink.core.execution.JobClient;
import org.apache.flink.runtime.jobgraph.JobGraph;
import org.apache.flink.runtime.jobgraph.SavepointConfigOptions;
import org.apache.flink.runtime.jobgraph.SavepointRestoreSettings;
import org.apache.flink.streaming.api.environment.ExecutionCheckpointingOptions;
import org.apache.flink.streaming.api.graph.StreamGraph;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.yarn.configuration.YarnConfigOptions;
import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.fasterxml.jackson.databind.node.ObjectNode;
/**
......@@ -300,9 +300,11 @@ public class JobManager {
TableResult tableResult = executor.executeStatementSet(inserts);
if (tableResult.getJobClient().isPresent()) {
job.setJobId(tableResult.getJobClient().get().getJobID().toHexString());
job.setJids(new ArrayList<String>() {{
job.setJids(new ArrayList<String>() {
{
add(job.getJobId());
}});
}
});
}
if (config.isUseResult()) {
// Build insert result.
......@@ -339,9 +341,11 @@ public class JobManager {
TableResult tableResult = executor.executeSql(item.getValue());
if (tableResult.getJobClient().isPresent()) {
job.setJobId(tableResult.getJobClient().get().getJobID().toHexString());
job.setJids(new ArrayList<String>() {{
job.setJids(new ArrayList<String>() {
{
add(job.getJobId());
}});
}
});
}
if (config.isUseResult()) {
IResult result =
......@@ -390,9 +394,11 @@ public class JobManager {
JobClient jobClient = executor.executeAsync(config.getJobName());
if (Asserts.isNotNull(jobClient)) {
job.setJobId(jobClient.getJobID().toHexString());
job.setJids(new ArrayList<String>() {{
job.setJids(new ArrayList<String>() {
{
add(job.getJobId());
}});
}
});
}
if (config.isUseResult()) {
IResult result = ResultBuilder.build(SqlType.EXECUTE, config.getMaxRowNum(), config.isUseChangeLog(), config.isUseAutoCancel(), executor.getTimeZone()).getResult(null);
......
dlink-core/src/main/java/com/dlink/job/JobParam.java
View file @ f8c751af
......@@ -17,7 +17,6 @@
*
*/
package com.dlink.job;
import java.util.ArrayList;
......@@ -69,7 +68,7 @@ public class JobParam {
public List<String> getTransStatement() {
List<String> statementList = new ArrayList<>();
for(StatementParam statementParam: trans){
for (StatementParam statementParam: trans) {
statementList.add(statementParam.getValue());
}
return statementList;
......
dlink-core/src/main/java/com/dlink/job/JobResult.java
View file @ f8c751af
......@@ -17,15 +17,15 @@
*
*/
package com.dlink.job;
import com.dlink.result.IResult;
import lombok.Getter;
import lombok.Setter;
import java.time.LocalDateTime;
import lombok.Getter;
import lombok.Setter;
/**
* JobResult
*
......@@ -51,7 +51,8 @@ public class JobResult {
public JobResult() {
}
public JobResult(Integer id, Integer jobInstanceId, JobConfig jobConfig, String jobManagerAddress, Job.JobStatus status, String statement, String jobId, String error, IResult result, LocalDateTime startTime, LocalDateTime endTime) {
public JobResult(Integer id, Integer jobInstanceId, JobConfig jobConfig, String jobManagerAddress, Job.JobStatus status,
String statement, String jobId, String error, IResult result, LocalDateTime startTime, LocalDateTime endTime) {
this.id = id;
this.jobInstanceId = jobInstanceId;
this.jobConfig = jobConfig;
......
dlink-core/src/main/java/com/dlink/job/RunTime.java
View file @ f8c751af
......@@ -17,7 +17,6 @@
*
*/
package com.dlink.job;
/**
......
dlink-core/src/main/java/com/dlink/job/StatementParam.java
View file @ f8c751af
......@@ -17,7 +17,6 @@
*
*/
package com.dlink.job;
import com.dlink.parser.SqlType;
......
dlink-core/src/main/java/com/dlink/plus/FlinkSqlPlus.java
View file @ f8c751af
......@@ -17,7 +17,6 @@
*
*/
package com.dlink.plus;
import com.dlink.executor.Executor;
......@@ -25,12 +24,14 @@ import com.dlink.explainer.Explainer;
import com.dlink.explainer.ca.ColumnCAResult;
import com.dlink.explainer.ca.TableCAResult;
import com.dlink.result.SqlExplainResult;
import com.fasterxml.jackson.databind.node.ObjectNode;
import org.apache.flink.runtime.rest.messages.JobPlanInfo;
import java.util.ArrayList;
import java.util.List;
import com.fasterxml.jackson.databind.node.ObjectNode;
/**
* FlinkSqlPlus
*
......
dlink-core/src/main/java/com/dlink/plus/SqlResult.java
View file @ f8c751af
......@@ -17,7 +17,6 @@
*
*/
package com.dlink.plus;
import org.apache.flink.table.api.TableResult;
......
dlink-core/src/main/java/com/dlink/result/DDLResult.java
View file @ f8c751af
......@@ -17,17 +17,16 @@
*
*/
package com.dlink.result;
import lombok.Getter;
import lombok.Setter;
import java.time.LocalDateTime;
import java.util.List;
import java.util.Map;
import java.util.Set;
import lombok.Getter;
import lombok.Setter;
/**
* DDLResult
*
......
dlink-core/src/main/java/com/dlink/result/DDLResultBuilder.java
View file @ f8c751af
......@@ -17,7 +17,6 @@
*
*/
package com.dlink.result;
import org.apache.flink.table.api.TableResult;
......
dlink-core/src/main/java/com/dlink/result/ErrorResult.java
View file @ f8c751af
......@@ -17,7 +17,6 @@
*
*/
package com.dlink.result;
import java.time.LocalDateTime;
......
dlink-core/src/main/java/com/dlink/result/InsertResult.java
View file @ f8c751af
......@@ -17,14 +17,13 @@
*
*/
package com.dlink.result;
import java.time.LocalDateTime;
import lombok.Getter;
import lombok.Setter;
import java.time.LocalDateTime;
/**
* InsertResult
*
......
dlink-core/src/main/java/com/dlink/result/InsertResultBuilder.java
View file @ f8c751af
......@@ -17,7 +17,6 @@
*
*/
package com.dlink.result;
import org.apache.flink.table.api.TableResult;
......
dlink-core/src/main/java/com/dlink/result/JobSubmitResult.java
View file @ f8c751af
......@@ -17,7 +17,6 @@
*
*/
package com.dlink.result;
/**
......
dlink-core/src/main/java/com/dlink/result/ResultBuilder.java
View file @ f8c751af
......@@ -17,13 +17,12 @@
*
*/
package com.dlink.result;
import org.apache.flink.table.api.TableResult;
import com.dlink.parser.SqlType;
import org.apache.flink.table.api.TableResult;
/**
* ResultBuilder
*
......
dlink-core/src/main/java/com/dlink/result/ResultPool.java
View file @ f8c751af
......@@ -17,7 +17,6 @@
*
*/
package com.dlink.result;
import java.util.HashMap;
......
dlink-core/src/main/java/com/dlink/result/ResultRunnable.java
View file @ f8c751af
......@@ -17,9 +17,11 @@
*
*/
package com.dlink.result;
import com.dlink.constant.FlinkConstant;
import com.dlink.utils.FlinkUtil;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.types.Row;
import org.apache.flink.types.RowKind;
......@@ -34,9 +36,6 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
import com.dlink.constant.FlinkConstant;
import com.dlink.utils.FlinkUtil;
/**
* ResultRunnable
*
......
dlink-core/src/main/java/com/dlink/result/RunResult.java
View file @ f8c751af
......@@ -17,7 +17,6 @@
*
*/
package com.dlink.result;
import com.dlink.executor.ExecutorSetting;
......
dlink-core/src/main/java/com/dlink/result/SelectResult.java
View file @ f8c751af
......@@ -17,17 +17,16 @@
*
*/
package com.dlink.result;
import lombok.Getter;
import lombok.Setter;
import java.time.LocalDateTime;
import java.util.List;
import java.util.Map;
import java.util.Set;
import lombok.Getter;
import lombok.Setter;
/**
* SelectResult
*
......@@ -53,7 +52,7 @@ public class SelectResult extends AbstractResult implements IResult {
this.columns = columns;
this.jobID = jobID;
this.success = success;
// this.endTime = LocalDateTime.now();
//this.endTime = LocalDateTime.now();
this.isDestroyed = false;
}
......
dlink-core/src/main/java/com/dlink/result/SelectResultBuilder.java
View file @ f8c751af
......@@ -17,7 +17,6 @@
*
*/
package com.dlink.result;
import org.apache.flink.table.api.TableResult;
......
dlink-core/src/main/java/com/dlink/result/ShowResultBuilder.java
View file @ f8c751af
......@@ -17,14 +17,20 @@
*
*/
package com.dlink.result;
import com.dlink.utils.FlinkUtil;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.types.Row;
import java.util.*;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
/**
* ShowResultBuilder
......
dlink-core/src/main/java/com/dlink/result/SubmitResult.java
View file @ f8c751af
......@@ -17,7 +17,6 @@
*
*/
package com.dlink.result;
import java.time.LocalDateTime;
......
dlink-core/src/main/java/com/dlink/session/ExecutorEntity.java
View file @ f8c751af
......@@ -17,15 +17,15 @@
*
*/
package com.dlink.session;
import com.dlink.executor.Executor;
import lombok.Getter;
import lombok.Setter;
import java.time.LocalDateTime;
import lombok.Getter;
import lombok.Setter;
/**
* FlinkEntity
*
......
dlink-core/src/main/java/com/dlink/session/SessionConfig.java
View file @ f8c751af
......@@ -17,10 +17,10 @@
*
*/
package com.dlink.session;
import com.dlink.executor.ExecutorSetting;
import lombok.Getter;
import lombok.Setter;
......
dlink-core/src/main/java/com/dlink/session/SessionInfo.java
View file @ f8c751af
......@@ -17,14 +17,13 @@
*
*/
package com.dlink.session;
import java.time.LocalDateTime;
import lombok.Getter;
import lombok.Setter;
import java.time.LocalDateTime;
/**
* SessionInfo
*
......
dlink-core/src/main/java/com/dlink/session/SessionPool.java
View file @ f8c751af
......@@ -17,7 +17,6 @@
*
*/
package com.dlink.session;
import com.dlink.constant.FlinkConstant;
......
dlink-core/src/main/java/com/dlink/utils/CustomStringJavaCompiler.java
View file @ f8c751af
......@@ -17,10 +17,8 @@
*
*/
package com.dlink.utils;
import javax.tools.*;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.OutputStream;
......@@ -32,6 +30,17 @@ import java.util.concurrent.ConcurrentHashMap;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import javax.tools.Diagnostic;
import javax.tools.DiagnosticCollector;
import javax.tools.FileObject;
import javax.tools.ForwardingJavaFileManager;
import javax.tools.JavaCompiler;
import javax.tools.JavaFileManager;
import javax.tools.JavaFileObject;
import javax.tools.SimpleJavaFileObject;
import javax.tools.StandardJavaFileManager;
import javax.tools.ToolProvider;
/**
* CustomStringJavaCompiler
*
......
dlink-core/src/main/java/com/dlink/utils/MapParseUtils.java
View file @ f8c751af
......@@ -17,10 +17,16 @@
*
*/
package com.dlink.utils;
import java.util.*;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Deque;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Stack;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import java.util.stream.Collectors;
......@@ -293,8 +299,6 @@ public class MapParseUtils {
* @date 2021/8/20 15:03
*/
public static Map parseForSelect(String inStr) {
// Map map = new HashMap();
// map.put(getMapKeyOnlySelectOrField(inStr), getSelectList(inStr));
return getKeyAndValues(inStr);
}
......
dlink-core/src/main/java/com/dlink/utils/UDFUtil.java
View file @ f8c751af
......@@ -17,14 +17,15 @@
*
*/
package com.dlink.utils;
import com.dlink.pool.ClassEntity;
import com.dlink.pool.ClassPool;
import groovy.lang.GroovyClassLoader;
import org.codehaus.groovy.control.CompilerConfiguration;
import groovy.lang.GroovyClassLoader;
/**
* UDFUtil
*
......@@ -58,6 +59,6 @@ public class UDFUtil {
groovyClassLoader.setShouldRecompile(true);
groovyClassLoader.defineClass(classEntity.getName(), classEntity.getClassByte());
Thread.currentThread().setContextClassLoader(groovyClassLoader);
// Class<?> clazz = groovyClassLoader.parseClass(codeSource,"com.dlink.ud.udf.SubstringFunction");
//Class<?> clazz = groovyClassLoader.parseClass(codeSource,"com.dlink.ud.udf.SubstringFunction");
}
}
dlink-core/src/test/java/com/dlink/core/BatchTest.java
View file @ f8c751af
......@@ -17,18 +17,12 @@
*
*/
package com.dlink.core;
import com.dlink.executor.CustomTableEnvironmentImpl;
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.ExecutionOptions;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.TableConfig;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.api.TableResult;
import org.junit.Test;
/**
......@@ -40,20 +34,20 @@ import org.junit.Test;
public class BatchTest {
@Test
public void batchTest() {
String source = "CREATE TABLE Orders (\n" +
" order_number BIGINT,\n" +
" price DECIMAL(32,2),\n" +
" buyer ROW<first_name STRING, last_name STRING>,\n" +
" order_time TIMESTAMP(3)\n" +
") WITH (\n" +
" 'connector' = 'datagen',\n" +
" 'number-of-rows' = '100'\n" +
")";
String source = "CREATE TABLE Orders (\n"
+ " order_number BIGINT,\n"
+ " price DECIMAL(32,2),\n"
+ " buyer ROW<first_name STRING, last_name STRING>,\n"
+ " order_time TIMESTAMP(3)\n"
+ ") WITH (\n"
+ " 'connector' = 'datagen',\n"
+ " 'number-of-rows' = '100'\n"
+ ")";
String select = "select order_number,price,order_time from Orders";
// LocalEnvironment environment = ExecutionEnvironment.createLocalEnvironment();
//LocalEnvironment environment = ExecutionEnvironment.createLocalEnvironment();
EnvironmentSettings settings = EnvironmentSettings
.newInstance()
// .inStreamingMode() // 声明为流任务
//.inStreamingMode() // 声明为流任务
.inBatchMode() // 声明为批任务
.build();
......
dlink-core/src/test/java/com/dlink/core/FlinkRestAPITest.java
View file @ f8c751af
......@@ -17,16 +17,17 @@
*
*/
package com.dlink.core;
import com.dlink.api.FlinkAPI;
import com.dlink.gateway.result.SavePointResult;
import com.fasterxml.jackson.databind.JsonNode;
import org.junit.Test;
import java.util.List;
import org.junit.Test;
import com.fasterxml.jackson.databind.JsonNode;
/**
* FlinkRestAPITest
*
......@@ -56,7 +57,6 @@ public class FlinkRestAPITest {
FlinkAPI.build(address).stop("0727f796fcf9e07d89e724f7e15598cf");
}
@Test
public void getCheckPointsDetailInfoTest() {
JsonNode checkPointsDetailInfo = FlinkAPI.build(address).getCheckPointsConfig("9b0910c865874430b98d3817a248eb24");
......@@ -87,14 +87,12 @@ public class FlinkRestAPITest {
System.out.println(jobManagerConfig.toString());
}
@Test
public void getJobManagerLogTest() {
String jobManagerLog = FlinkAPI.build(address).getJobManagerLog();
System.out.println(jobManagerLog);
}
@Test
public void getJobManagerStdOutTest() {
String jobManagerLogs = FlinkAPI.build(address).getJobManagerStdOut();
......@@ -106,6 +104,7 @@ public class FlinkRestAPITest {
JsonNode jobManagerLogList = FlinkAPI.build(address).getJobManagerLogList();
System.out.println(jobManagerLogList.toString());
}
@Test
public void getJobManagerLogListToDetailTest() {
String jobManagerLogList = FlinkAPI.build(address).getJobManagerLogFileDetail("jobmanager.log");
......@@ -124,7 +123,6 @@ public class FlinkRestAPITest {
System.out.println(taskManagerMetrics.toString());
}
@Test
public void getTaskManagerLogTest() {
String taskManagerLog = FlinkAPI.build(address).getTaskManagerLog("container_e46_1655948912029_0061_01_000002");
......@@ -142,6 +140,7 @@ public class FlinkRestAPITest {
JsonNode taskManagerLogList = FlinkAPI.build(address).getTaskManagerLogList("container_e46_1655948912029_0061_01_000002");
System.out.println(taskManagerLogList.toString());
}
@Test
public void getTaskManagerLogListToDetail() {
String taskManagerLogDetail = FlinkAPI.build(address).getTaskManagerLogFileDeatil("container_e46_1655948912029_0061_01_000002","taskmanager.log");
......
dlink-core/src/test/java/com/dlink/core/FlinkSqlPlusTest.java
View file @ f8c751af
......@@ -17,11 +17,12 @@
*
*/
package com.dlink.core;
import com.dlink.plus.FlinkSqlPlus;
import org.apache.flink.runtime.rest.messages.JobPlanInfo;
import org.junit.Test;
/**
......@@ -34,48 +35,48 @@ public class FlinkSqlPlusTest {
@Test
public void getJobPlanInfo() {
String sql = "jdbcconfig:='connector' = 'jdbc',\n" +
" 'url' = 'jdbc:mysql://127.0.0.1:3306/test?useUnicode=true&characterEncoding=UTF-8&autoReconnect=true&useSSL=false&zeroDateTimeBehavior=convertToNull&serverTimezone=Asia/Shanghai&allowPublicKeyRetrieval=true',\n" +
" 'username'='dlink',\n" +
" 'password'='dlink',;\n" +
"create temporary function TOP2 as 'com.dlink.ud.udtaf.Top2';\n" +
"CREATE TABLE student (\n" +
" sid INT,\n" +
" name STRING,\n" +
" PRIMARY KEY (sid) NOT ENFORCED\n" +
") WITH (\n" +
" ${jdbcconfig}\n" +
" 'table-name' = 'student'\n" +
");\n" +
"CREATE TABLE score (\n" +
" cid INT,\n" +
" sid INT,\n" +
" cls STRING,\n" +
" score INT,\n" +
" PRIMARY KEY (cid) NOT ENFORCED\n" +
") WITH (\n" +
" ${jdbcconfig}\n" +
" 'table-name' = 'score'\n" +
");\n" +
"CREATE TABLE scoretop2 (\n" +
" cls STRING,\n" +
" score INT,\n" +
" `rank` INT,\n" +
" PRIMARY KEY (cls,`rank`) NOT ENFORCED\n" +
") WITH (\n" +
" ${jdbcconfig}\n" +
" 'table-name' = 'scoretop2'\n" +
");\n" +
"CREATE AGGTABLE aggscore AS \n" +
"SELECT cls,score,rank\n" +
"FROM score\n" +
"GROUP BY cls\n" +
"AGG BY TOP2(score) as (score,rank);\n" +
"\n" +
"insert into scoretop2\n" +
"select \n" +
"b.cls,b.score,b.`rank`\n" +
"from aggscore b";
String sql = "jdbcconfig:='connector' = 'jdbc',\n"
+ " 'url' = 'jdbc:mysql://127.0.0.1:3306/test?useUnicode=true&characterEncoding=UTF-8&autoReconnect=true&useSSL=false"
+ "&zeroDateTimeBehavior=convertToNull&serverTimezone=Asia/Shanghai&allowPublicKeyRetrieval=true',\n"
+ " 'username'='dlink',\n"
+ " 'password'='dlink',;\n"
+ "create temporary function TOP2 as 'com.dlink.ud.udtaf.Top2';\n"
+ "CREATE TABLE student (\n"
+ " sid INT,\n"
+ " name STRING,\n"
+ " PRIMARY KEY (sid) NOT ENFORCED\n"
+ ") WITH (\n"
+ " ${jdbcconfig}\n"
+ " 'table-name' = 'student'\n"
+ ");\n"
+ "CREATE TABLE score (\n"
+ " cid INT,\n"
+ " sid INT,\n"
+ " cls STRING,\n"
+ " score INT,\n"
+ " PRIMARY KEY (cid) NOT ENFORCED\n"
+ ") WITH (\n"
+ " ${jdbcconfig}\n"
+ " 'table-name' = 'score'\n"
+ ");\n"
+ "CREATE TABLE scoretop2 (\n"
+ " cls STRING,\n"
+ " score INT,\n"
+ " `rank` INT,\n"
+ " PRIMARY KEY (cls,`rank`) NOT ENFORCED\n"
+ ") WITH (\n"
+ " ${jdbcconfig}\n"
+ " 'table-name' = 'scoretop2'\n"
+ ");\n"
+ "CREATE AGGTABLE aggscore AS \n"
+ "SELECT cls,score,rank\n"
+ "FROM score\n"
+ "GROUP BY cls\n"
+ "AGG BY TOP2(score) as (score,rank);\n"
+ "insert into scoretop2\n"
+ "select \n"
+ "b.cls,b.score,b.`rank`\n"
+ "from aggscore b";
FlinkSqlPlus plus = FlinkSqlPlus.build();
JobPlanInfo jobPlanInfo = plus.getJobPlanInfo(sql);
......
dlink-core/src/test/java/com/dlink/core/JobManagerTest.java
View file @ f8c751af
......@@ -17,19 +17,18 @@
*
*/
package com.dlink.core;
import java.util.HashMap;
import org.junit.Test;
import com.dlink.job.JobConfig;
import com.dlink.job.JobManager;
import com.dlink.job.JobResult;
import com.dlink.result.ResultPool;
import com.dlink.result.SelectResult;
import java.util.HashMap;
import org.junit.Test;
/**
* JobManagerTest
*
......@@ -48,14 +47,14 @@ public class JobManagerTest {
config.setAddress("192.168.123.157:8081");
}
JobManager jobManager = JobManager.build(config);
String sql1 = "CREATE TABLE Orders (\n" +
" order_number BIGINT,\n" +
" price DECIMAL(32,2),\n" +
" order_time TIMESTAMP(3)\n" +
") WITH (\n" +
" 'connector' = 'datagen',\n" +
" 'rows-per-second' = '1'\n" +
");";
String sql1 = "CREATE TABLE Orders (\n"
+ " order_number BIGINT,\n"
+ " price DECIMAL(32,2),\n"
+ " order_time TIMESTAMP(3)\n"
+ ") WITH (\n"
+ " 'connector' = 'datagen',\n"
+ " 'rows-per-second' = '1'\n"
+ ");";
String sql3 = "select order_number,price,order_time from Orders";
String sql = sql1 + sql3;
JobResult result = jobManager.executeSql(sql);
......
dlink-core/src/test/java/com/dlink/core/LineageTest.java
View file @ f8c751af
......@@ -17,11 +17,11 @@
*
*/
package com.dlink.core;
import com.dlink.explainer.lineage.LineageBuilder;
import com.dlink.explainer.lineage.LineageResult;
import org.junit.Test;
/**
......@@ -34,21 +34,21 @@ public class LineageTest {
@Test
public void sumTest() {
String sql = "CREATE TABLE ST (\n" +
" a STRING,\n" +
" b STRING,\n" +
" c STRING\n" +
") WITH (\n" +
" 'connector' = 'datagen',\n" +
" 'rows-per-second' = '1'\n" +
");\n" +
"CREATE TABLE TT (\n" +
" A STRING,\n" +
" B STRING\n" +
") WITH (\n" +
" 'connector' = 'print'\n" +
");\n" +
"insert into TT select a||c A ,b||c B from ST";
String sql = "CREATE TABLE ST (\n"
+ " a STRING,\n"
+ " b STRING,\n"
+ " c STRING\n"
+ ") WITH (\n"
+ " 'connector' = 'datagen',\n"
+ " 'rows-per-second' = '1'\n"
+ ");\n"
+ "CREATE TABLE TT (\n"
+ " A STRING,\n"
+ " B STRING\n"
+ ") WITH (\n"
+ " 'connector' = 'print'\n"
+ ");\n"
+ "insert into TT select a||c A ,b||c B from ST";
LineageResult result = LineageBuilder.getLineage(sql);
System.out.println("end");
}
......
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