Merge remote-tracking branch 'origin/1.8.0_dev_partitioned_join' into v1.8.0_dev

Author: simenliuxing

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

@@ -45,6 +45,7 @@
 import org.apache.calcite.sql.parser.SqlParseException;
 import org.apache.calcite.sql.parser.SqlParserPos;
 import org.apache.commons.collections.CollectionUtils;
+import org.apache.flink.api.common.typeinfo.SqlTimeTypeInfo;
 import org.apache.flink.api.common.typeinfo.TypeInformation;
 import org.apache.flink.api.java.functions.KeySelector;
 import org.apache.flink.api.java.tuple.Tuple2;
@@ -56,11 +57,13 @@
 import org.apache.flink.table.api.StreamQueryConfig;
 import org.apache.flink.table.api.Table;
 import org.apache.flink.table.api.java.StreamTableEnvironment;
+import org.apache.flink.table.typeutils.TimeIndicatorTypeInfo;
 import org.apache.flink.types.Row;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.io.Serializable;
+import java.sql.Timestamp;
 import java.util.*;
 import java.util.stream.Collectors;
 
@@ -242,6 +245,28 @@ public RowTypeInfo buildOutRowTypeInfo(List<FieldInfo> sideJoinFieldInfo, HashBa
         return new RowTypeInfo(sideOutTypes, sideOutNames);
     }
 
+    /**
+     *  对时间类型进行类型转换
+     * @param leftTypeInfo
+     * @return
+     */
+    private RowTypeInfo buildLeftTableOutType(RowTypeInfo leftTypeInfo) {
+        TypeInformation[] sideOutTypes = new TypeInformation[leftTypeInfo.getFieldNames().length];
+        TypeInformation<?>[] fieldTypes = leftTypeInfo.getFieldTypes();
+        for (int i = 0; i < sideOutTypes.length; i++) {
+            sideOutTypes[i] = convertTimeAttributeType(fieldTypes[i]);
+        }
+        RowTypeInfo rowTypeInfo = new RowTypeInfo(sideOutTypes, leftTypeInfo.getFieldNames());
+        return rowTypeInfo;
+    }
+
+    private TypeInformation convertTimeAttributeType(TypeInformation typeInformation) {
+        if (typeInformation instanceof TimeIndicatorTypeInfo) {
+            return TypeInformation.of(Timestamp.class);
+        }
+        return typeInformation;
+    }
+
     //需要考虑更多的情况
     private void replaceFieldName(SqlNode sqlNode, HashBasedTable<String, String, String> mappingTable, String targetTableName, String tableAlias) {
         SqlKind sqlKind = sqlNode.getKind();
@@ -747,10 +772,12 @@ private void joinFun(Object pollObj, Map<String, Table> localTableCache,
         DataStream adaptStream = tableEnv.toRetractStream(targetTable, org.apache.flink.types.Row.class)
                                 .map((Tuple2<Boolean, Row> f0) -> { return f0.f1; })
                                 .returns(Row.class);
-        //adaptStream.getTransformation().setOutputType(leftTypeInfo);
+
 
         //join side table before keyby ===> Reducing the size of each dimension table cache of async
         if(sideTableInfo.isPartitionedJoin()){
+            RowTypeInfo leftTableOutType = buildLeftTableOutType(leftTypeInfo);
+            adaptStream.getTransformation().setOutputType(leftTableOutType);
             List<String> leftJoinColList = getConditionFields(joinInfo.getCondition(), joinInfo.getLeftTableAlias(), sideTableInfo);
             String[] leftJoinColArr = leftJoinColList.toArray(new String[leftJoinColList.size()]);
             adaptStream = adaptStream.keyBy(leftJoinColArr);
@@ -781,6 +808,8 @@ private void joinFun(Object pollObj, Map<String, Table> localTableCache,
         }
     }
 
+
+
     private boolean checkFieldsInfo(CreateTmpTableParser.SqlParserResult result, Table table) {
         List<String> fieldNames = new LinkedList<>();
         String fieldsInfo = result.getFieldsInfoStr();