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

Author: gituser

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

@@ -118,7 +118,7 @@ public void parseSelectFields(JoinInfo joinInfo){
     public String getTargetFieldType(String fieldName){
         int fieldIndex = sideTableInfo.getFieldList().indexOf(fieldName);
         if(fieldIndex == -1){
-            throw new RuntimeException(sideTableInfo.getName() + "can't find field: " + fieldName);
+            throw new RuntimeException(sideTableInfo.getName() + " can't find field: " + fieldName);
         }
 
         return sideTableInfo.getFieldTypes()[fieldIndex];

core/src/main/java/com/dtstack/flink/sql/table/AbstractTableParser.java

@@ -19,12 +19,15 @@
 
 package com.dtstack.flink.sql.table;
 
+import com.dtstack.flink.sql.side.AbstractSideTableInfo;
 import com.dtstack.flink.sql.util.ClassUtil;
 import com.dtstack.flink.sql.util.DtStringUtil;
 import com.google.common.base.Preconditions;
 import com.google.common.collect.Maps;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.flink.api.java.tuple.Tuple2;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import java.util.Arrays;
 import java.util.List;
@@ -43,6 +46,8 @@
 
 public abstract class AbstractTableParser {
 
+    private static final Logger LOG = LoggerFactory.getLogger(AbstractTableParser.class);
+
     private static final String PRIMARY_KEY = "primaryKey";
     private static final String NEST_JSON_FIELD_KEY = "nestFieldKey";
     private static final String CHAR_TYPE_NO_LENGTH = "CHAR";
@@ -105,30 +110,50 @@ public void parseFieldsInfo(String fieldsInfo, AbstractTableInfo tableInfo) {
                 continue;
             }
 
-            Tuple2<String, String> t = extractType(fieldRow, tableInfo.getName());
-            String fieldName = t.f0;
-            String fieldType = t.f1;
+            handleKeyNotHaveAlias(fieldRow, tableInfo);
+        }
 
-            Class fieldClass;
-            AbstractTableInfo.FieldExtraInfo fieldExtraInfo = null;
+        /*
+         *  check whether filed list contains pks and then add pks into field list.
+         *  because some no-sql database is not primary key. eg :redis、hbase etc...
+         */
+        if (tableInfo instanceof AbstractSideTableInfo) {
+            tableInfo.getPrimaryKeys().stream()
+                    .filter(pk -> !tableInfo.getFieldList().contains(pk))
+                    .forEach(pk -> {
+                        try {
+                            handleKeyNotHaveAlias(String.format("%s varchar", pk), tableInfo);
+                        } catch (Exception e) {
+                            LOG.error(String.format("Handle primary key failed. Reason: %s", e.getMessage()));
+                        }
+                    });
+        }
 
-            Matcher matcher = charTypePattern.matcher(fieldType);
-            if (matcher.find()) {
-                fieldClass = dbTypeConvertToJavaType(CHAR_TYPE_NO_LENGTH);
-                fieldExtraInfo = new AbstractTableInfo.FieldExtraInfo();
-                fieldExtraInfo.setLength(Integer.parseInt(matcher.group(1)));
-            } else {
-                fieldClass = dbTypeConvertToJavaType(fieldType);
-            }
+        tableInfo.finish();
+    }
 
-            tableInfo.addPhysicalMappings(fieldName, fieldName);
-            tableInfo.addField(fieldName);
-            tableInfo.addFieldClass(fieldClass);
-            tableInfo.addFieldType(fieldType);
-            tableInfo.addFieldExtraInfo(fieldExtraInfo);
+    private void handleKeyNotHaveAlias(String fieldRow, AbstractTableInfo tableInfo) {
+        Tuple2<String, String> t = extractType(fieldRow, tableInfo.getName());
+        String fieldName = t.f0;
+        String fieldType = t.f1;
+
+        Class fieldClass;
+        AbstractTableInfo.FieldExtraInfo fieldExtraInfo = null;
+
+        Matcher matcher = charTypePattern.matcher(fieldType);
+        if (matcher.find()) {
+            fieldClass = dbTypeConvertToJavaType(CHAR_TYPE_NO_LENGTH);
+            fieldExtraInfo = new AbstractTableInfo.FieldExtraInfo();
+            fieldExtraInfo.setLength(Integer.parseInt(matcher.group(1)));
+        } else {
+            fieldClass = dbTypeConvertToJavaType(fieldType);
         }
 
-        tableInfo.finish();
+        tableInfo.addPhysicalMappings(fieldName, fieldName);
+        tableInfo.addField(fieldName);
+        tableInfo.addFieldClass(fieldClass);
+        tableInfo.addFieldType(fieldType);
+        tableInfo.addFieldExtraInfo(fieldExtraInfo);
     }
 
     private Tuple2<String, String> extractType(String fieldRow, String tableName) {

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

@@ -55,6 +55,8 @@ public abstract class AbstractRowKeyModeDealer {
 
     protected List<FieldInfo> outFieldInfoList;
 
+    protected static final String ROWKEY = "rowkey";
+
     //key:Returns the value of the position, returns the index values ​​in the input data
     protected Map<Integer, Integer> inFieldIndex = Maps.newHashMap();
 

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

@@ -110,6 +110,12 @@ private String dealOneRow(ArrayList<ArrayList<KeyValue>> args, String rowKeyStr,
                         //The order of the fields defined in the data conversion table
                         List<Object> sideVal = Lists.newArrayList();
                         for (String key : colNames) {
+
+                            if (ROWKEY.equalsIgnoreCase(key)) {
+                                sideVal.add(rowKeyStr);
+                                continue;
+                            }
+
                             Object val = sideMap.get(key);
                             if (val == null) {
                                 LOG.error("can't get data with column {}", key);

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

@@ -84,6 +84,12 @@ public void asyncGetData(String tableName, String rowKeyStr, BaseRow input, Resu
                         //The order of the fields defined in the data conversion table
                         List<Object> sideVal = Lists.newArrayList();
                         for(String key : colNames){
+
+                            if (ROWKEY.equalsIgnoreCase(key)) {
+                                sideVal.add(rowKeyStr);
+                                continue;
+                            }
+
                             Object val = sideMap.get(key);
                             if(val == null){
                                 LOG.error("can't get data with column {}", key);