Merge branch 'hotfix_1.10_4.0.x_31674' into '1.10_release_4.0.x'

[fix] decimal type transform error when table join

See merge request dt-insight-engine/flinkStreamSQL!173

Author: zoudaokoulife

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

@@ -43,6 +43,7 @@
 import org.apache.calcite.sql.parser.SqlParseException;
 import org.apache.commons.collections.CollectionUtils;
 import org.apache.commons.lang3.StringUtils;
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
 import org.apache.flink.api.common.typeinfo.TypeInformation;
 import org.apache.flink.api.java.typeutils.RowTypeInfo;
 import org.apache.flink.streaming.api.datastream.DataStream;
@@ -52,16 +53,27 @@
 import org.apache.flink.table.api.java.StreamTableEnvironment;
 import org.apache.flink.table.catalog.ObjectIdentifier;
 import org.apache.flink.table.runtime.typeutils.BaseRowTypeInfo;
+import org.apache.flink.table.runtime.typeutils.BigDecimalTypeInfo;
+import org.apache.flink.table.types.logical.DecimalType;
+import org.apache.flink.table.types.logical.LegacyTypeInformationType;
 import org.apache.flink.table.types.logical.LogicalType;
 import org.apache.flink.table.typeutils.TimeIndicatorTypeInfo;
 import org.apache.flink.types.Row;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.time.LocalDateTime;
-import java.util.*;
-
-import static org.apache.calcite.sql.SqlKind.*;
+import java.util.Arrays;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.Queue;
+import java.util.Set;
+
+import static org.apache.calcite.sql.SqlKind.AS;
+import static org.apache.calcite.sql.SqlKind.INSERT;
+import static org.apache.calcite.sql.SqlKind.SELECT;
+import static org.apache.calcite.sql.SqlKind.WITH_ITEM;
 
 /**
  * Reason:
@@ -284,10 +296,7 @@ private Table getTableFromCache(Map<String, Table> localTableCache, String table
      */
     private boolean checkJoinCondition(SqlNode conditionNode, String sideTableAlias, AbstractSideTableInfo sideTableInfo) {
         List<String> conditionFields = getConditionFields(conditionNode, sideTableAlias, sideTableInfo);
-        if (CollectionUtils.isEqualCollection(conditionFields, convertPrimaryAlias(sideTableInfo))) {
-            return true;
-        }
-        return false;
+        return CollectionUtils.isEqualCollection(conditionFields, convertPrimaryAlias(sideTableInfo));
     }
 
     private List<String> convertPrimaryAlias(AbstractSideTableInfo sideTableInfo) {
@@ -372,8 +381,12 @@ private void joinFun(Object pollObj,
 
         int length = leftTable.getSchema().getFieldDataTypes().length;
         LogicalType[] logicalTypes = new LogicalType[length];
-        for(int i=0; i<length; i++){
+        for (int i = 0; i < length; i++) {
             logicalTypes[i] = leftTable.getSchema().getFieldDataTypes()[i].getLogicalType();
+            if (logicalTypes[i] instanceof LegacyTypeInformationType &&
+                    ((LegacyTypeInformationType<?>) logicalTypes[i]).getTypeInformation().getClass().equals(BigDecimalTypeInfo.class)) {
+                logicalTypes[i] = new DecimalType(38, 18);
+            }
         }
 
         BaseRowTypeInfo leftBaseTypeInfo = new BaseRowTypeInfo(logicalTypes, leftTable.getSchema().getFieldNames());
@@ -412,7 +425,14 @@ private void joinFun(Object pollObj,
             targetTable = localTableCache.get(joinInfo.getLeftTableName());
         }
 
-        RowTypeInfo typeInfo = new RowTypeInfo(targetTable.getSchema().getFieldTypes(), targetTable.getSchema().getFieldNames());
+        TypeInformation[] fieldDataTypes = targetTable.getSchema().getFieldTypes();
+        for (int i = 0; i < fieldDataTypes.length; i++) {
+            if (fieldDataTypes[i].getClass().equals(BigDecimalTypeInfo.class)) {
+                fieldDataTypes[i] = BasicTypeInfo.BIG_DEC_TYPE_INFO;
+            }
+        }
+
+        RowTypeInfo typeInfo = new RowTypeInfo(fieldDataTypes, targetTable.getSchema().getFieldNames());
 
         DataStream adaptStream = tableEnv.toRetractStream(targetTable, typeInfo)
                 .filter(f -> f.f0)
@@ -484,9 +504,7 @@ private boolean checkFieldsInfo(CreateTmpTableParser.SqlParserResult result, Tab
                 String fieldType = filed[filed.length - 1].trim();
                 Class fieldClass = ClassUtil.stringConvertClass(fieldType);
                 Class tableField = table.getSchema().getFieldType(i).get().getTypeClass();
-                if (fieldClass == tableField) {
-                    continue;
-                } else {
+                if (fieldClass != tableField) {
                     return false;
                 }
             }