Skip to content

Commit bf07e2c

Browse files
todd5167todd5167
committed
1.10 first commit
1 parent b259aa5 commit bf07e2c

64 files changed

Lines changed: 275 additions & 264 deletions

File tree

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

‎cassandra/cassandra-side/cassandra-all-side/pom.xml‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -70,14 +70,14 @@
7070
</goals>
7171
<configuration>
7272
<tasks>
73-
<copy todir="${basedir}/../../../plugins/cassandraallside">
73+
<copy todir="${basedir}/../../../sqlplugins/cassandraallside">
7474
<fileset dir="target/">
7575
<include name="${project.artifactId}-${project.version}.jar" />
7676
</fileset>
7777
</copy>
7878

79-
<move file="${basedir}/../../../plugins/cassandraallside/${project.artifactId}-${project.version}.jar"
80-
tofile="${basedir}/../../../plugins/cassandraallside/${project.name}-${git.branch}.jar" />
79+
<move file="${basedir}/../../../sqlplugins/cassandraallside/${project.artifactId}-${project.version}.jar"
80+
tofile="${basedir}/../../../sqlplugins/cassandraallside/${project.name}-${git.branch}.jar" />
8181
</tasks>
8282
</configuration>
8383
</execution>

‎cassandra/cassandra-side/cassandra-async-side/pom.xml‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -86,14 +86,14 @@
8686
</goals>
8787
<configuration>
8888
<tasks>
89-
<copy todir="${basedir}/../../../plugins/cassandraasyncside">
89+
<copy todir="${basedir}/../../../sqlplugins/cassandraasyncside">
9090
<fileset dir="target/">
9191
<include name="${project.artifactId}-${project.version}.jar" />
9292
</fileset>
9393
</copy>
9494

95-
<move file="${basedir}/../../../plugins/cassandraasyncside/${project.artifactId}-${project.version}.jar"
96-
tofile="${basedir}/../../../plugins/cassandraasyncside/${project.name}-${git.branch}.jar" />
95+
<move file="${basedir}/../../../sqlplugins/cassandraasyncside/${project.artifactId}-${project.version}.jar"
96+
tofile="${basedir}/../../../sqlplugins/cassandraasyncside/${project.name}-${git.branch}.jar" />
9797
</tasks>
9898
</configuration>
9999
</execution>

‎cassandra/cassandra-sink/pom.xml‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -64,14 +64,14 @@
6464
</goals>
6565
<configuration>
6666
<tasks>
67-
<copy todir="${basedir}/../../plugins/cassandrasink">
67+
<copy todir="${basedir}/../../sqlplugins/cassandrasink">
6868
<fileset dir="target/">
6969
<include name="${project.artifactId}-${project.version}.jar" />
7070
</fileset>
7171
</copy>
7272

73-
<move file="${basedir}/../../plugins/cassandrasink/${project.artifactId}-${project.version}.jar"
74-
tofile="${basedir}/../../plugins/cassandrasink/${project.name}-${git.branch}.jar" />
73+
<move file="${basedir}/../../sqlplugins/cassandrasink/${project.artifactId}-${project.version}.jar"
74+
tofile="${basedir}/../../sqlplugins/cassandrasink/${project.name}-${git.branch}.jar" />
7575
</tasks>
7676
</configuration>
7777
</execution>

‎clickhouse/clickhouse-side/clickhouse-all-side/pom.xml‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -72,14 +72,14 @@
7272
</goals>
7373
<configuration>
7474
<tasks>
75-
<copy todir="${basedir}/../../../plugins/clickhouseallside">
75+
<copy todir="${basedir}/../../../sqlplugins/clickhouseallside">
7676
<fileset dir="target/">
7777
<include name="${project.artifactId}-${project.version}.jar"/>
7878
</fileset>
7979
</copy>
8080

81-
<move file="${basedir}/../../../plugins/clickhouseallside/${project.artifactId}-${project.version}.jar"
82-
tofile="${basedir}/../../../plugins/clickhouseallside/${project.name}-${git.branch}.jar"/>
81+
<move file="${basedir}/../../../sqlplugins/clickhouseallside/${project.artifactId}-${project.version}.jar"
82+
tofile="${basedir}/../../../sqlplugins/clickhouseallside/${project.name}-${git.branch}.jar"/>
8383
</tasks>
8484
</configuration>
8585
</execution>

‎clickhouse/clickhouse-side/clickhouse-async-side/pom.xml‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -72,14 +72,14 @@
7272
</goals>
7373
<configuration>
7474
<tasks>
75-
<copy todir="${basedir}/../../../plugins/clickhouseasyncside">
75+
<copy todir="${basedir}/../../../sqlplugins/clickhouseasyncside">
7676
<fileset dir="target/">
7777
<include name="${project.artifactId}-${project.version}.jar"/>
7878
</fileset>
7979
</copy>
8080

81-
<move file="${basedir}/../../../plugins/clickhouseasyncside/${project.artifactId}-${project.version}.jar"
82-
tofile="${basedir}/../../../plugins/clickhouseasyncside/${project.name}-${git.branch}.jar"/>
81+
<move file="${basedir}/../../../sqlplugins/clickhouseasyncside/${project.artifactId}-${project.version}.jar"
82+
tofile="${basedir}/../../../sqlplugins/clickhouseasyncside/${project.name}-${git.branch}.jar"/>
8383
</tasks>
8484
</configuration>
8585
</execution>

‎clickhouse/clickhouse-sink/pom.xml‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -71,14 +71,14 @@
7171
</goals>
7272
<configuration>
7373
<tasks>
74-
<copy todir="${basedir}/../../plugins/clickhousesink">
74+
<copy todir="${basedir}/../../sqlplugins/clickhousesink">
7575
<fileset dir="target/">
7676
<include name="${project.artifactId}-${project.version}.jar"/>
7777
</fileset>
7878
</copy>
7979

80-
<move file="${basedir}/../../plugins/clickhousesink/${project.artifactId}-${project.version}.jar"
81-
tofile="${basedir}/../../plugins/clickhousesink/${project.name}-${git.branch}.jar"/>
80+
<move file="${basedir}/../../sqlplugins/clickhousesink/${project.artifactId}-${project.version}.jar"
81+
tofile="${basedir}/../../sqlplugins/clickhousesink/${project.name}-${git.branch}.jar"/>
8282
</tasks>
8383
</configuration>
8484
</execution>

‎console/console-sink/pom.xml‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -60,14 +60,14 @@
6060
</goals>
6161
<configuration>
6262
<tasks>
63-
<copy todir="${basedir}/../../plugins/consolesink">
63+
<copy todir="${basedir}/../../sqlplugins/consolesink">
6464
<fileset dir="target/">
6565
<include name="${project.artifactId}-${project.version}.jar" />
6666
</fileset>
6767
</copy>
6868

69-
<move file="${basedir}/../../plugins/consolesink/${project.artifactId}-${project.version}.jar"
70-
tofile="${basedir}/../../plugins/consolesink/${project.name}-${git.branch}.jar" />
69+
<move file="${basedir}/../../sqlplugins/consolesink/${project.artifactId}-${project.version}.jar"
70+
tofile="${basedir}/../../sqlplugins/consolesink/${project.name}-${git.branch}.jar" />
7171
</tasks>
7272
</configuration>
7373
</execution>

‎core/pom.xml‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -181,13 +181,13 @@
181181
</goals>
182182
<configuration>
183183
<tasks>
184-
<copy todir="${basedir}/../plugins">
184+
<copy todir="${basedir}/../sqlplugins">
185185
<fileset dir="target/">
186186
<include name="${project.artifactId}-${project.version}.jar" />
187187
</fileset>
188188
</copy>
189-
<move file="${basedir}/../plugins/${project.artifactId}-${project.version}.jar"
190-
tofile="${basedir}/../plugins/${project.name}-${git.branch}.jar" />
189+
<move file="${basedir}/../sqlplugins/${project.artifactId}-${project.version}.jar"
190+
tofile="${basedir}/../sqlplugins/${project.name}-${git.branch}.jar" />
191191
</tasks>
192192
</configuration>
193193
</execution>

‎core/src/main/java/com/dtstack/flink/sql/environment/MyLocalStreamEnvironment.java‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -104,8 +104,8 @@ public JobExecutionResult execute(StreamGraph streamGraph) throws Exception {
104104
Configuration configuration = new Configuration();
105105
configuration.addAll(jobGraph.getJobConfiguration());
106106

107-
configuration.setString(TaskManagerOptions.MANAGED_MEMORY_SIZE, "512M");
108-
configuration.setInteger(TaskManagerOptions.NUM_TASK_SLOTS, jobGraph.getMaximumParallelism());
107+
configuration.setString(TaskManagerOptions.MANAGED_MEMORY_SIZE.key(), "512M");
108+
configuration.setInteger(TaskManagerOptions.NUM_TASK_SLOTS.key(), jobGraph.getMaximumParallelism());
109109

110110
// add (and override) the settings with what the user defined
111111
configuration.addAll(this.conf);

‎core/src/main/java/com/dtstack/flink/sql/exec/FlinkSQLExec.java‎

Lines changed: 22 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -27,21 +27,23 @@
2727
import org.apache.flink.table.api.internal.TableImpl;
2828
import org.apache.flink.table.api.java.StreamTableEnvironment;
2929
import org.apache.flink.table.api.java.internal.StreamTableEnvironmentImpl;
30+
import org.apache.flink.table.catalog.CatalogManager;
31+
import org.apache.flink.table.catalog.ObjectIdentifier;
32+
import org.apache.flink.table.operations.Operation;
3033
import org.apache.flink.table.operations.QueryOperation;
3134
import org.apache.flink.table.planner.calcite.FlinkPlannerImpl;
3235
import org.apache.flink.table.planner.delegation.PlannerBase;
3336
import org.apache.flink.table.planner.delegation.StreamPlanner;
34-
import org.apache.flink.table.planner.operations.PlannerQueryOperation;
3537
import org.apache.flink.table.planner.operations.SqlToOperationConverter;
3638
import org.apache.flink.table.sinks.TableSink;
3739
import org.slf4j.Logger;
3840
import org.slf4j.LoggerFactory;
3941
import scala.Option;
42+
import scala.Tuple2;
4043

44+
import java.lang.reflect.Constructor;
4145
import java.lang.reflect.InvocationTargetException;
4246
import java.lang.reflect.Method;
43-
import java.util.Arrays;
44-
import java.util.List;
4547

4648

4749
/**
@@ -51,12 +53,13 @@
5153
*/
5254
public class FlinkSQLExec {
5355
private static final Logger LOG = LoggerFactory.getLogger(FlinkSQLExec.class);
56+
5457
public static void sqlUpdate(StreamTableEnvironment tableEnv, String stmt) throws Exception {
5558
StreamTableEnvironmentImpl tableEnvImpl = ((StreamTableEnvironmentImpl) tableEnv);
5659
StreamPlanner streamPlanner = (StreamPlanner)tableEnvImpl.getPlanner();
5760
FlinkPlannerImpl flinkPlanner = streamPlanner.createFlinkPlanner();
5861

59-
RichSqlInsert insert = (RichSqlInsert)flinkPlanner.parse(stmt);
62+
RichSqlInsert insert = (RichSqlInsert) flinkPlanner.validate(flinkPlanner.parser().parse(stmt));
6063
TableImpl queryResult = extractQueryTableFromInsertCaluse(tableEnvImpl, flinkPlanner, insert);
6164

6265
String targetTableName = ((SqlIdentifier) ((SqlInsert) insert).getTargetTable()).names.get(0);
@@ -93,19 +96,28 @@ public static void sqlUpdate(StreamTableEnvironment tableEnv, String stmt) throw
9396

9497
}
9598

99+
private static SqlToOperationConverter createSqlToOperationConverter(FlinkPlannerImpl flinkPlanner, CatalogManager catalogManager)
100+
throws NoSuchMethodException, IllegalAccessException, InvocationTargetException, InstantiationException {
101+
102+
Constructor<SqlToOperationConverter> declaredConstructor = SqlToOperationConverter.class.getDeclaredConstructor(FlinkPlannerImpl.class, CatalogManager.class);
103+
declaredConstructor.setAccessible(true);
104+
SqlToOperationConverter sqlToOperationConverter = declaredConstructor.newInstance(flinkPlanner, catalogManager);
105+
return sqlToOperationConverter;
106+
}
107+
96108
private static TableSink getTableSinkByPlanner(StreamPlanner streamPlanner, String targetTableName)
97109
throws NoSuchMethodException, IllegalAccessException, InvocationTargetException {
98-
Method getTableSink = PlannerBase.class.getDeclaredMethod("getTableSink", List.class);
110+
Method getTableSink = PlannerBase.class.getDeclaredMethod("getTableSink", ObjectIdentifier.class);
99111
getTableSink.setAccessible(true);
100-
Option tableSinkOption = (Option) getTableSink.invoke(streamPlanner, Arrays.asList(targetTableName));
101-
return (TableSink) tableSinkOption.get();
112+
ObjectIdentifier objectIdentifier = ObjectIdentifier.of(streamPlanner.catalogManager().getCurrentCatalog(), streamPlanner.catalogManager().getCurrentDatabase(), targetTableName);
113+
Option tableSinkOption = (Option) getTableSink.invoke(streamPlanner, objectIdentifier);
114+
return (TableSink) ((Tuple2) tableSinkOption.get())._2;
102115
}
103116

104117
private static TableImpl extractQueryTableFromInsertCaluse(StreamTableEnvironmentImpl tableEnvImpl, FlinkPlannerImpl flinkPlanner, RichSqlInsert insert)
105118
throws NoSuchMethodException, IllegalAccessException, InvocationTargetException {
106-
107-
PlannerQueryOperation queryOperation = (PlannerQueryOperation) SqlToOperationConverter.convert(flinkPlanner,
108-
insert.getSource());
119+
StreamPlanner streamPlanner = (StreamPlanner) tableEnvImpl.getPlanner();
120+
Operation queryOperation = SqlToOperationConverter.convert(flinkPlanner, streamPlanner.catalogManager(), insert.getSource()).get();
109121
Method createTableMethod = TableEnvironmentImpl.class.getDeclaredMethod("createTable", QueryOperation.class);
110122
createTableMethod.setAccessible(true);
111123
return (TableImpl) createTableMethod.invoke(tableEnvImpl, queryOperation);

0 commit comments

Comments
 (0)