insert sink table by name

Author: todd5167

core/src/main/java/com/dtstack/flink/sql/Main.java

@@ -333,7 +333,7 @@ private static StreamExecutionEnvironment getStreamExeEnv(Properties confPropert
         StreamExecutionEnvironment env = !ClusterMode.local.name().equals(deployMode) ?
                 StreamExecutionEnvironment.getExecutionEnvironment() :
                 new MyLocalStreamEnvironment();
-
+        env.getConfig().disableClosureCleaner();
         env.setParallelism(FlinkUtil.getEnvParallelism(confProperties));
         Configuration globalJobParameters = new Configuration();
         Method method = Configuration.class.getDeclaredMethod("setValueInternal", String.class, Object.class);

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

@@ -11,6 +11,8 @@
 import org.apache.flink.table.calcite.FlinkPlannerImpl;
 import org.apache.flink.table.plan.logical.LogicalRelNode;
 import org.apache.flink.table.plan.schema.TableSinkTable;
+import org.apache.flink.table.plan.schema.TableSourceSinkTable;
+import scala.Option;
 
 import java.lang.reflect.Method;
 
@@ -39,12 +41,17 @@ public static void sqlUpdate(StreamTableEnvironment tableEnv, String stmt) throw
 
         Method method = TableEnvironment.class.getDeclaredMethod("getTable", String.class);
         method.setAccessible(true);
+        Option sinkTab = (Option)method.invoke(tableEnv, targetTableName);
 
-        TableSinkTable targetTable = (TableSinkTable) method.invoke(tableEnv, targetTableName);
-        String[] fieldNames = targetTable.tableSink().getFieldNames();
+        if (sinkTab.isEmpty()) {
+            throw  new ValidationException("Sink table " + targetTableName + "not found in flink");
+        }
 
-        Table newTable = null;
+        TableSourceSinkTable targetTable = (TableSourceSinkTable) sinkTab.get();
+        TableSinkTable tableSinkTable = (TableSinkTable)targetTable.tableSinkTable().get();
+        String[] fieldNames = tableSinkTable.tableSink().getFieldNames();
 
+        Table newTable = null;
         try {
             newTable = queryResult.select(String.join(",", fieldNames));
         } catch (Exception e) {

core/src/main/java/com/dtstack/flink/sql/side/SideSqlExec.java

@@ -116,7 +116,7 @@ public void exec(String sql, Map<String, SideTableInfo> sideTableMap, StreamTabl
                     }
                 }else if(pollSqlNode.getKind() == AS){
                     AliasInfo aliasInfo = parseASNode(pollSqlNode);
-                    Table table = tableEnv.sql(aliasInfo.getName());
+                    Table table = tableEnv.sqlQuery(aliasInfo.getName());
                     tableEnv.registerTable(aliasInfo.getAlias(), table);
                     localTableCache.put(aliasInfo.getAlias(), table);
                 }
@@ -613,7 +613,7 @@ public void registerTmpTable(CreateTmpTableParser.SqlParserResult result,
                     tableEnv.sqlUpdate(pollSqlNode.toString());
                 }else if(pollSqlNode.getKind() == AS){
                     AliasInfo aliasInfo = parseASNode(pollSqlNode);
-                    Table table = tableEnv.sql(aliasInfo.getName());
+                    Table table = tableEnv.sqlQuery(aliasInfo.getName());
                     tableEnv.registerTable(aliasInfo.getAlias(), table);
                     if(LOG.isInfoEnabled()){
                         LOG.info("Register Table {} by {}", aliasInfo.getAlias(), aliasInfo.getName());
@@ -740,7 +740,7 @@ private boolean checkFieldsInfo(CreateTmpTableParser.SqlParserResult result, Tab
                 fieldNames.add(fieldName);
                 String fieldType = filed[filed.length - 1 ].trim();
                 Class fieldClass = ClassUtil.stringConvertClass(fieldType);
-                Class tableField = table.getSchema().getType(i).get().getTypeClass();
+                Class tableField = table.getSchema().getFieldType(i).get().getTypeClass();
                 if (fieldClass == tableField){
                     continue;
                 } else {