Merge branch 'feat_keyPartition_mergeTest' into '1.8_test_3.10.x'

Feat key partition merge test



See merge request !268

Author: zoudaokoulife

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

@@ -187,12 +187,17 @@ public static List<URL> getExternalJarUrls(String addJarListStr) throws java.io.
         }
         return jarUrlList;
     }
-
-    public static void sqlTranslation(String localSqlPluginPath, StreamTableEnvironment tableEnv, SqlTree sqlTree, Map<String, SideTableInfo> sideTableMap, Map<String, Table> registerTableCache, StreamQueryConfig queryConfig) throws Exception {
-        SideSqlExec sideSqlExec = new SideSqlExec();
+    
+    private static void sqlTranslation(String localSqlPluginPath,
+            StreamTableEnvironment tableEnv,
+            SqlTree sqlTree,Map<String, SideTableInfo> sideTableMap,
+            Map<String, Table> registerTableCache,
+            StreamQueryConfig queryConfig) throws Exception {
+
+    	SideSqlExec sideSqlExec = new SideSqlExec();
         sideSqlExec.setLocalSqlPluginPath(localSqlPluginPath);
         for (CreateTmpTableParser.SqlParserResult result : sqlTree.getTmpSqlList()) {
-            sideSqlExec.registerTmpTable(result, sideTableMap, tableEnv, registerTableCache);
+            sideSqlExec.exec(result.getExecSql(), sideTableMap, tableEnv, registerTableCache, queryConfig, result);
         }
 
         for (InsertSqlParser.SqlParseResult result : sqlTree.getExecSqlList()) {
@@ -208,7 +213,7 @@ public static void sqlTranslation(String localSqlPluginPath, StreamTableEnvironm
                     SqlNode sqlNode = org.apache.calcite.sql.parser.SqlParser.create(realSql, CalciteConfig.MYSQL_LEX_CONFIG).parseStmt();
                     String tmpSql = ((SqlInsert) sqlNode).getSource().toString();
                     tmp.setExecSql(tmpSql);
-                    sideSqlExec.registerTmpTable(tmp, sideTableMap, tableEnv, registerTableCache);
+                    sideSqlExec.exec(tmp.getExecSql(), sideTableMap, tableEnv, registerTableCache, queryConfig, tmp);
                 } else {
                     for (String sourceTable : result.getSourceTableList()) {
                         if (sideTableMap.containsKey(sourceTable)) {
@@ -218,8 +223,11 @@ public static void sqlTranslation(String localSqlPluginPath, StreamTableEnvironm
                     }
                     if (isSide) {
                         //sql-dimensional table contains the dimension table of execution
-                        sideSqlExec.exec(result.getExecSql(), sideTableMap, tableEnv, registerTableCache, queryConfig);
+                        sideSqlExec.exec(result.getExecSql(), sideTableMap, tableEnv, registerTableCache, queryConfig, null);
                     } else {
+                        System.out.println("----------exec sql without dimension join-----------");
+                        System.out.println("----------real sql exec is--------------------------");
+                        System.out.println(result.getExecSql());
                         FlinkSQLExec.sqlUpdate(tableEnv, result.getExecSql(), queryConfig);
                         if (LOG.isInfoEnabled()) {
                             LOG.info("exec sql: " + result.getExecSql());

core/src/main/java/com/dtstack/flink/sql/format/SerializationMetricWrapper.java

@@ -36,6 +36,8 @@
  */
 public class SerializationMetricWrapper implements SerializationSchema<Row> {
 
+    private static final long serialVersionUID = 1L;
+
     private SerializationSchema<Row> serializationSchema;
 
     private transient RuntimeContext runtimeContext;
@@ -77,4 +79,8 @@ public void setRuntimeContext(RuntimeContext runtimeContext) {
         this.runtimeContext = runtimeContext;
     }
 
+    public SerializationSchema<Row> getSerializationSchema() {
+        return serializationSchema;
+    }
+
 }

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

@@ -202,4 +202,21 @@ public boolean isLeftIsTmpTable() {
     public void setLeftIsTmpTable(boolean leftIsTmpTable) {
         this.leftIsTmpTable = leftIsTmpTable;
     }
+
+    @Override
+    public String toString() {
+        return "JoinInfo{" +
+                "leftIsSideTable=" + leftIsSideTable +
+                ", leftIsTmpTable=" + leftIsTmpTable +
+                ", rightIsSideTable=" + rightIsSideTable +
+                ", leftTableName='" + leftTableName + '\'' +
+                ", leftTableAlias='" + leftTableAlias + '\'' +
+                ", rightTableName='" + rightTableName + '\'' +
+                ", rightTableAlias='" + rightTableAlias + '\'' +
+                ", condition=" + condition +
+                ", selectFields=" + selectFields +
+                ", selectNode=" + selectNode +
+                ", joinType=" + joinType +
+                '}';
+    }
 }