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/SideSqlExec.java

@@ -68,10 +68,12 @@
 
 import java.sql.Timestamp;
 import java.time.LocalDateTime;
+import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
 import java.util.Queue;
 import java.util.Set;
 
@@ -317,23 +319,23 @@ private List<String> convertPrimaryAlias(AbstractSideTableInfo sideTableInfo) {
      * @param conditionNode
      * @param joinScope
      */
-    public void checkConditionFieldsInTable(SqlNode conditionNode, JoinScope joinScope) {
+    public void checkConditionFieldsInTable(SqlNode conditionNode, JoinScope joinScope, AbstractSideTableInfo sideTableInfo) {
         List<SqlNode> sqlNodeList = Lists.newArrayList();
         ParseUtils.parseAnd(conditionNode, sqlNodeList);
         for (SqlNode sqlNode : sqlNodeList) {
             if (!SqlKind.COMPARISON.contains(sqlNode.getKind())) {
-                throw new RuntimeException("not compare operator.");
+                throw new RuntimeException("It is not comparison operator.");
             }
 
             SqlNode leftNode = ((SqlBasicCall) sqlNode).getOperands()[0];
             SqlNode rightNode = ((SqlBasicCall) sqlNode).getOperands()[1];
 
             if (leftNode.getKind() == SqlKind.IDENTIFIER) {
-                checkFieldInTable((SqlIdentifier) leftNode, joinScope, conditionNode);
+                checkFieldInTable((SqlIdentifier) leftNode, joinScope, conditionNode, sideTableInfo);
             }
 
             if (rightNode.getKind() == SqlKind.IDENTIFIER) {
-                checkFieldInTable((SqlIdentifier) rightNode, joinScope, conditionNode);
+                checkFieldInTable((SqlIdentifier) rightNode, joinScope, conditionNode, sideTableInfo);
             }
 
         }
@@ -345,11 +347,11 @@ public void checkConditionFieldsInTable(SqlNode conditionNode, JoinScope joinSco
      * @param joinScope
      * @param conditionNode
      */
-    private void checkFieldInTable(SqlIdentifier sqlNode, JoinScope joinScope, SqlNode conditionNode) {
+    private void checkFieldInTable(SqlIdentifier sqlNode, JoinScope joinScope, SqlNode conditionNode,  AbstractSideTableInfo sideTableInfo) {
         String tableName = sqlNode.getComponent(0).getSimple();
         String fieldName = sqlNode.getComponent(1).getSimple();
         JoinScope.ScopeChild scopeChild = joinScope.getScope(tableName);
-        String tableErrorMsg = "table [%s] is not exist. error condition is [%s]. if you find [%s] is exist, please check AS statement";
+        String tableErrorMsg = "Table [%s] is not exist. Error condition is [%s]. If you find [%s] is exist. Please check AS statement.";
         Preconditions.checkState(
             scopeChild != null,
             tableErrorMsg,
@@ -359,8 +361,21 @@ private void checkFieldInTable(SqlIdentifier sqlNode, JoinScope joinScope, SqlNo
         );
 
         String[] fieldNames = scopeChild.getRowTypeInfo().getFieldNames();
-        boolean hasField = Arrays.asList(fieldNames).contains(fieldName);
-        String fieldErrorMsg = "table [%s] has not [%s] field.\n error join condition is [%s]";
+        ArrayList<String> allFieldNames = new ArrayList(
+            Arrays.asList(fieldNames)
+        );
+        // HBase、Redis这种NoSQL Primary Key不在字段列表中，所以要加进去。
+        if (sideTableInfo != null) {
+            List<String> pks = sideTableInfo.getPrimaryKeys();
+            if (pks != null) {
+                pks.stream()
+                    .filter(pk -> !allFieldNames.contains(pk))
+                    .forEach(pk -> allFieldNames.add(pk));
+            }
+        }
+
+        boolean hasField = allFieldNames.contains(fieldName);
+        String fieldErrorMsg = "Table [%s] has not [%s] field. Error join condition is [%s]. If you find it is exist. Please check AS statement.";
         Preconditions.checkState(
             hasField,
             fieldErrorMsg,
@@ -481,7 +496,8 @@ private void joinFun(Object pollObj,
         joinScope.addScope(rightScopeChild);
 
         HashBasedTable<String, String, String> mappingTable = ((JoinInfo) pollObj).getTableFieldRef();
-        checkConditionFieldsInTable(joinInfo.getCondition(), joinScope);
+        // 检查JOIN等式字段是否在原表中
+        checkConditionFieldsInTable(joinInfo.getCondition(), joinScope, sideTableInfo);
 
         //获取两个表的所有字段
         List<FieldInfo> sideJoinFieldInfo = ParserJoinField.getRowTypeInfo(joinInfo.getSelectNode(), joinScope, true);

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

@@ -59,7 +59,7 @@ public abstract class AbstractTableInfo implements Serializable {
 
     private List<String> primaryKeys;
 
-    private Integer parallelism = 1;
+    private Integer parallelism = -1;
 
     public String[] getFieldTypes() {
         return fieldTypes;