Merge branch '1.10_release_4.0.x' into 1.10_release_4.1.x

Author: gituser

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

@@ -194,7 +194,7 @@ public void asyncInvoke(Row row, ResultFuture<BaseRow> resultFuture) throws Exce
     }
 
     private Map<String, Object> parseInputParam(Row input) {
-        Map<String, Object> inputParams = Maps.newHashMap();
+        Map<String, Object> inputParams = Maps.newLinkedHashMap();
         for (int i = 0; i < sideInfo.getEqualValIndex().size(); i++) {
             Integer conValIndex = sideInfo.getEqualValIndex().get(i);
             Object equalObj = input.getField(conValIndex);

hbase/hbase-side/hbase-async-side/src/main/java/com/dtstack/flink/sql/side/hbase/rowkeydealer/AbstractRowKeyModeDealer.java

@@ -94,9 +94,6 @@ protected Row fillData(Row input, Object sideInput){
         Row row = new Row(outFieldInfoList.size());
         for(Map.Entry<Integer, Integer> entry : inFieldIndex.entrySet()){
             Object obj = input.getField(entry.getValue());
-            if(obj instanceof Timestamp){
-                obj = ((Timestamp)obj).getTime();
-            }
             row.setField(entry.getKey(), obj);
         }
 

rdb/rdb-side/src/main/java/com/dtstack/flink/sql/side/rdb/all/AbstractRdbAllReqRow.java

@@ -42,8 +42,10 @@
 import java.time.LocalDateTime;
 import java.util.ArrayList;
 import java.util.Calendar;
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.stream.Collectors;
 
@@ -103,7 +105,7 @@ public void flatMap(Row value, Collector<BaseRow> out) throws Exception {
         List<Integer> equalValIndex = sideInfo.getEqualValIndex();
         ArrayList<Object> inputParams = equalValIndex.stream()
                 .map(value::getField)
-                .filter(object -> null != object)
+                .filter(Objects::nonNull)
                 .collect(Collectors.toCollection(ArrayList::new));
 
         if (inputParams.size() != equalValIndex.size() && sideInfo.getJoinType() == JoinType.LEFT) {
@@ -121,7 +123,7 @@ public void flatMap(Row value, Collector<BaseRow> out) throws Exception {
             Row row = fillData(value, null);
             RowDataComplete.collectRow(out, row);
         } else if (!CollectionUtils.isEmpty(cacheList)) {
-            cacheList.stream().forEach(one -> out.collect(RowDataConvert.convertToBaseRow(fillData(value, one))));
+            cacheList.forEach(one -> out.collect(RowDataConvert.convertToBaseRow(fillData(value, one))));
         }
     }
 
@@ -166,13 +168,17 @@ private void queryAndFillData(Map<String, List<Map<String, Object>>> tmpCache, C
         ResultSet resultSet = statement.executeQuery(sql);
 
         String[] sideFieldNames = StringUtils.split(sideInfo.getSideSelectFields(), ",");
-        String[] fields = sideInfo.getSideTableInfo().getFieldTypes();
+        String[] sideFieldTypes = sideInfo.getSideTableInfo().getFieldTypes();
+        Map<String, String> sideFieldNamesAndTypes = Maps.newHashMap();
+        for (int i = 0; i < sideFieldNames.length; i++) {
+            sideFieldNamesAndTypes.put(sideFieldNames[i], sideFieldTypes[i]);
+        }
+
         while (resultSet.next()) {
             Map<String, Object> oneRow = Maps.newHashMap();
             for (String fieldName : sideFieldNames) {
                 Object object = resultSet.getObject(fieldName.trim());
-                int fieldIndex = sideInfo.getSideTableInfo().getFieldList().indexOf(fieldName.trim());
-                object = SwitchUtil.getTarget(object, fields[fieldIndex]);
+                object = SwitchUtil.getTarget(object, sideFieldNamesAndTypes.get(fieldName));
                 oneRow.put(fieldName.trim(), object);
             }
 

rdb/rdb-side/src/main/java/com/dtstack/flink/sql/side/rdb/all/RdbAllSideInfo.java

@@ -18,24 +18,24 @@
 
 package com.dtstack.flink.sql.side.rdb.all;
 
-import org.apache.flink.api.java.typeutils.RowTypeInfo;
-
+import com.dtstack.flink.sql.side.AbstractSideTableInfo;
+import com.dtstack.flink.sql.side.BaseSideInfo;
 import com.dtstack.flink.sql.side.FieldInfo;
 import com.dtstack.flink.sql.side.JoinInfo;
 import com.dtstack.flink.sql.side.PredicateInfo;
-import com.dtstack.flink.sql.side.BaseSideInfo;
-import com.dtstack.flink.sql.side.AbstractSideTableInfo;
 import com.dtstack.flink.sql.side.rdb.table.RdbSideTableInfo;
 import com.dtstack.flink.sql.util.ParseUtils;
 import com.google.common.collect.Lists;
 import org.apache.calcite.sql.SqlNode;
 import org.apache.commons.collections.CollectionUtils;
 import org.apache.commons.lang3.StringUtils;
+import org.apache.flink.api.java.typeutils.RowTypeInfo;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import java.util.Arrays;
 import java.util.List;
+import java.util.Map;
+import java.util.Objects;
 import java.util.stream.Collectors;
 
 /**
@@ -59,7 +59,18 @@ public RdbAllSideInfo(RowTypeInfo rowTypeInfo, JoinInfo joinInfo, List<FieldInfo
     @Override
     public void buildEqualInfo(JoinInfo joinInfo, AbstractSideTableInfo sideTableInfo) {
         RdbSideTableInfo rdbSideTableInfo = (RdbSideTableInfo) sideTableInfo;
-        sqlCondition = getSelectFromStatement(getTableName(rdbSideTableInfo), Arrays.asList(StringUtils.split(sideSelectFields, ",")), sideTableInfo.getPredicateInfoes());
+        List<String> selectFields = Lists.newArrayList();
+        Map<String, String> physicalFields = rdbSideTableInfo.getPhysicalFields();
+        physicalFields.keySet().forEach(
+                item -> {
+                    if (Objects.isNull(physicalFields.get(item))) {
+                        selectFields.add(quoteIdentifier(item));
+                    } else {
+                        selectFields.add(quoteIdentifier(physicalFields.get(item)) + " AS " + quoteIdentifier(item));
+                    }
+                }
+        );
+        sqlCondition = getSelectFromStatement(getTableName(rdbSideTableInfo), selectFields, sideTableInfo.getPredicateInfoes());
         LOG.info("--------dimension sql query-------\n{}" + sqlCondition);
     }
 
@@ -68,7 +79,7 @@ public String getAdditionalWhereClause() {
     }
 
     private String getSelectFromStatement(String tableName, List<String> selectFields, List<PredicateInfo> predicateInfoes) {
-        String fromClause = selectFields.stream().map(this::quoteIdentifier).collect(Collectors.joining(", "));
+        String fromClause = String.join(", ", selectFields);
         String predicateClause = predicateInfoes.stream().map(this::buildFilterCondition).collect(Collectors.joining(" AND "));
         String whereClause = buildWhereClause(predicateClause);
         return "SELECT " + fromClause + " FROM " + tableName + whereClause;

rdb/rdb-side/src/main/java/com/dtstack/flink/sql/side/rdb/async/RdbAsyncReqRow.java

@@ -338,7 +338,7 @@ private void handleQuery(SQLConnection connection, Map<String, Object> inputPara
     }
 
     private Map<String, Object> formatInputParam(Map<String, Object> inputParam) {
-        Map<String, Object> result = Maps.newHashMap();
+        Map<String, Object> result = Maps.newLinkedHashMap();
         inputParam.forEach((k, v) -> {
             result.put(k, convertDataType(v));
         });