Skip to content

Commit

Permalink
[Refactor][Flink] Refactor flink sql parse (DataLinkDC#2509)
Browse files Browse the repository at this point in the history 
* spotless

* spotless

* spotless

* spotless

* spotless

* spotless

* spotless
  • Loading branch information
@zackyoungh
zackyoungh authored Nov 8, 2023
1 parent 0fa9655 commit 4cad347
Show file tree
Hide file tree
Showing 45 changed files with 892 additions and 446 deletions.
31 changes: 20 additions & 11 deletions ...client/dinky-client-1.14/src/main/java/org/dinky/executor/CustomTableEnvironmentImpl.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -70,13 +70,16 @@
import org.apache.flink.table.typeutils.FieldInfoUtils;
import org.apache.flink.types.Row;

import java.io.File;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.stream.Collectors;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
Expand Down Expand Up @@ -124,13 +127,7 @@ public static CustomTableEnvironmentImpl createBatch(StreamExecutionEnvironment
configuration.set(ExecutionOptions.RUNTIME_MODE, RuntimeExecutionMode.BATCH);
TableConfig tableConfig = new TableConfig();
tableConfig.addConfiguration(configuration);
return create(
executionEnvironment,
EnvironmentSettings.newInstance()
.useBlinkPlanner()
.inBatchMode()
.build(),
tableConfig);
return create(executionEnvironment, EnvironmentSettings.inBatchMode(), tableConfig);
}

public static CustomTableEnvironmentImpl create(
Expand Down Expand Up @@ -295,15 +292,14 @@ public SqlExplainResult explainSqlRecord(String statement, ExplainDetail... extr
return record;
}

public boolean parseAndLoadConfiguration(
String statement, StreamExecutionEnvironment environment, Map<String, Object> setMap) {
public boolean parseAndLoadConfiguration(String statement, Map<String, Object> setMap) {
List<Operation> operations = getParser().parse(statement);
for (Operation operation : operations) {
if (operation instanceof SetOperation) {
callSet((SetOperation) operation, environment, setMap);
callSet((SetOperation) operation, getStreamExecutionEnvironment(), setMap);
return true;
} else if (operation instanceof ResetOperation) {
callReset((ResetOperation) operation, environment, setMap);
callReset((ResetOperation) operation, getStreamExecutionEnvironment(), setMap);
return true;
}
}
Expand Down Expand Up @@ -378,6 +374,19 @@ public void executeCTAS(Operation operation) {
}
}

@Override
public void addJar(File... jarPath) {
Configuration configuration = this.getRootConfiguration();
List<String> jars = configuration.get(PipelineOptions.JARS);
if (jars == null) {
configuration.set(
PipelineOptions.JARS,
Arrays.stream(jarPath).map(File::getAbsolutePath).collect(Collectors.toList()));
} else {
CollUtil.addAll(jars, jarPath);
}
}

@Override
public <T> void createTemporaryView(String path, DataStream<T> dataStream, Expression... fields) {
createTemporaryView(path, fromDataStream(dataStream, fields));
Expand Down
23 changes: 19 additions & 4 deletions ...client/dinky-client-1.15/src/main/java/org/dinky/executor/CustomTableEnvironmentImpl.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -65,10 +65,13 @@
import org.apache.flink.table.operations.ddl.CreateTableOperation;
import org.apache.flink.types.Row;

import java.io.File;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
Expand Down Expand Up @@ -192,6 +195,19 @@ public ObjectNode getStreamGraph(String statement) {
}
}

@Override
public void addJar(File... jarPath) {
Configuration configuration = this.getRootConfiguration();
List<String> jars = configuration.get(PipelineOptions.JARS);
if (jars == null) {
configuration.set(
PipelineOptions.JARS,
Arrays.stream(jarPath).map(File::getAbsolutePath).collect(Collectors.toList()));
} else {
CollUtil.addAll(jars, jarPath);
}
}

@Override
public JobPlanInfo getJobPlanInfo(List<String> statements) {
return new JobPlanInfo(JsonPlanGenerator.generatePlan(getJobGraphFromInserts(statements)));
Expand Down Expand Up @@ -254,15 +270,14 @@ public SqlExplainResult explainSqlRecord(String statement, ExplainDetail... extr
return record;
}

public boolean parseAndLoadConfiguration(
String statement, StreamExecutionEnvironment environment, Map<String, Object> setMap) {
public boolean parseAndLoadConfiguration(String statement, Map<String, Object> setMap) {
List<Operation> operations = getParser().parse(statement);
for (Operation operation : operations) {
if (operation instanceof SetOperation) {
callSet((SetOperation) operation, environment, setMap);
callSet((SetOperation) operation, getStreamExecutionEnvironment(), setMap);
return true;
} else if (operation instanceof ResetOperation) {
callReset((ResetOperation) operation, environment, setMap);
callReset((ResetOperation) operation, getStreamExecutionEnvironment(), setMap);
return true;
}
}
Expand Down
43 changes: 29 additions & 14 deletions ...client/dinky-client-1.16/src/main/java/org/dinky/executor/CustomTableEnvironmentImpl.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -50,7 +50,9 @@
import org.apache.flink.table.operations.ddl.CreateTableOperation;
import org.apache.flink.types.Row;

import java.io.File;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.Map;
Expand Down Expand Up @@ -102,6 +104,33 @@ public static CustomTableEnvironmentImpl create(
return new CustomTableEnvironmentImpl(streamTableEnvironment);
}

public boolean parseAndLoadConfiguration(String statement, Map<String, Object> setMap) {
List<Operation> operations = getParser().parse(statement);
for (Operation operation : operations) {
if (operation instanceof SetOperation) {
callSet((SetOperation) operation, getStreamExecutionEnvironment(), setMap);
return true;
} else if (operation instanceof ResetOperation) {
callReset((ResetOperation) operation, getStreamExecutionEnvironment(), setMap);
return true;
}
}
return false;
}

@Override
public void addJar(File... jarPath) {
Configuration configuration = this.getRootConfiguration();
List<String> jars = configuration.get(PipelineOptions.JARS);
if (jars == null) {
configuration.set(
PipelineOptions.JARS,
Arrays.stream(jarPath).map(File::getAbsolutePath).collect(Collectors.toList()));
} else {
CollUtil.addAll(jars, jarPath);
}
}

public ObjectNode getStreamGraph(String statement) {
List<Operation> operations = super.getParser().parse(statement);
if (operations.size() != 1) {
Expand Down Expand Up @@ -198,20 +227,6 @@ public SqlExplainResult explainSqlRecord(String statement, ExplainDetail... extr
return record;
}

public boolean parseAndLoadConfiguration(
String statement, StreamExecutionEnvironment environment, Map<String, Object> setMap) {
for (Operation operation : getParser().parse(statement)) {
if (operation instanceof SetOperation) {
callSet((SetOperation) operation, environment, setMap);
return true;
} else if (operation instanceof ResetOperation) {
callReset((ResetOperation) operation, environment, setMap);
return true;
}
}
return false;
}

private void callSet(
SetOperation setOperation, StreamExecutionEnvironment environment, Map<String, Object> setMap) {
if (!setOperation.getKey().isPresent() || !setOperation.getValue().isPresent()) {
Expand Down
48 changes: 34 additions & 14 deletions ...client/dinky-client-1.17/src/main/java/org/dinky/executor/CustomTableEnvironmentImpl.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -51,7 +51,9 @@
import org.apache.flink.table.operations.ddl.CreateTableOperation;
import org.apache.flink.types.Row;

import java.io.File;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.Map;
Expand Down Expand Up @@ -103,6 +105,35 @@ public static CustomTableEnvironmentImpl create(
return new CustomTableEnvironmentImpl(streamTableEnvironment);
}

@Override
public void addJar(File... jarPath) {
Configuration configuration = this.getRootConfiguration();
List<String> jars = configuration.get(PipelineOptions.JARS);
if (jars == null) {
configuration.set(
PipelineOptions.JARS,
Arrays.stream(jarPath).map(File::getAbsolutePath).collect(Collectors.toList()));
} else {
CollUtil.addAll(jars, jarPath);
}
}

@Override
public boolean parseAndLoadConfiguration(String statement, Map<String, Object> setMap) {
List<Operation> operations = getParser().parse(statement);
for (Operation operation : operations) {
if (operation instanceof SetOperation) {
callSet((SetOperation) operation, getStreamExecutionEnvironment(), setMap);
return true;
} else if (operation instanceof ResetOperation) {
callReset((ResetOperation) operation, getStreamExecutionEnvironment(), setMap);
return true;
}
}
return false;
}

@Override
public ObjectNode getStreamGraph(String statement) {
List<Operation> operations = super.getParser().parse(statement);
if (operations.size() != 1) {
Expand Down Expand Up @@ -142,6 +173,7 @@ public JobPlanInfo getJobPlanInfo(List<String> statements) {
return new JobPlanInfo(JsonPlanGenerator.generatePlan(getJobGraphFromInserts(statements)));
}

@Override
public StreamGraph getStreamGraphFromInserts(List<String> statements) {
List<ModifyOperation> modifyOperations = new ArrayList<>();
statements.stream().map(statement -> getParser().parse(statement)).forEach(operations -> {
Expand All @@ -161,10 +193,12 @@ public StreamGraph getStreamGraphFromInserts(List<String> statements) {
return transOperatoinsToStreamGraph(modifyOperations);
}

@Override
public JobGraph getJobGraphFromInserts(List<String> statements) {
return getStreamGraphFromInserts(statements).getJobGraph();
}

@Override
public SqlExplainResult explainSqlRecord(String statement, ExplainDetail... extraDetails) {
List<Operation> operations = getParser().parse(statement);
if (operations.size() != 1) {
Expand Down Expand Up @@ -194,20 +228,6 @@ public SqlExplainResult explainSqlRecord(String statement, ExplainDetail... extr
return data;
}

public boolean parseAndLoadConfiguration(
String statement, StreamExecutionEnvironment environment, Map<String, Object> setMap) {
for (Operation operation : getParser().parse(statement)) {
if (operation instanceof SetOperation) {
callSet((SetOperation) operation, environment, setMap);
return true;
} else if (operation instanceof ResetOperation) {
callReset((ResetOperation) operation, environment, setMap);
return true;
}
}
return false;
}

private void callSet(
SetOperation setOperation, StreamExecutionEnvironment environment, Map<String, Object> setMap) {
if (!setOperation.getKey().isPresent() || !setOperation.getValue().isPresent()) {
Expand Down
Loading

0 comments on commit 4cad347

Please sign in to comment.