[hotfix-33089][core]add pks into sidetableinfo's fields.

Author: HiLany

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,13 +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 java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
 import java.util.Map;
@@ -105,30 +107,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) {
+            List<String> pks = new ArrayList<>();
+            tableInfo.getPrimaryKeys().stream().forEach(pk -> {
+                if (!tableInfo.getFieldList().contains(pk)) {
+                    pks.add(String.format("%s varchar", pk));
+                }
+            });
+            pks.stream().forEach(pk -> {
+                handleKeyNotHaveAlias(pk, tableInfo);
+            });
+        }
 
-            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) {