[fix-32711][core] fix nosql(hbase/redis) side join that can not find primary key.

Author: a49a

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,7 +319,7 @@ private List<String> convertPrimaryAlias(AbstractSideTableInfo sideTableInfo) {
      * @param conditionNode
      * @param joinScope
      */
-    public void checkConditionFieldsInTable(SqlNode conditionNode, JoinScope joinScope) {
+    public void checkConditionFieldsInTable(SqlNode conditionNode, JoinScope joinScope, Map<String, AbstractSideTableInfo> sideTableMap) {
         List<SqlNode> sqlNodeList = Lists.newArrayList();
         ParseUtils.parseAnd(conditionNode, sqlNodeList);
         for (SqlNode sqlNode : sqlNodeList) {
@@ -329,11 +331,11 @@ public void checkConditionFieldsInTable(SqlNode conditionNode, JoinScope joinSco
             SqlNode rightNode = ((SqlBasicCall) sqlNode).getOperands()[1];
 
             if (leftNode.getKind() == SqlKind.IDENTIFIER) {
-                checkFieldInTable((SqlIdentifier) leftNode, joinScope, conditionNode);
+                checkFieldInTable((SqlIdentifier) leftNode, joinScope, conditionNode, sideTableMap);
             }
 
             if (rightNode.getKind() == SqlKind.IDENTIFIER) {
-                checkFieldInTable((SqlIdentifier) rightNode, joinScope, conditionNode);
+                checkFieldInTable((SqlIdentifier) rightNode, joinScope, conditionNode, sideTableMap);
             }
 
         }
@@ -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, Map<String, AbstractSideTableInfo> sideTableMap) {
         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,22 @@ 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)
+        );
+        AbstractSideTableInfo sideTableInfo = sideTableMap.get(tableName);
+        // 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 +497,7 @@ private void joinFun(Object pollObj,
         joinScope.addScope(rightScopeChild);
 
         HashBasedTable<String, String, String> mappingTable = ((JoinInfo) pollObj).getTableFieldRef();
-        checkConditionFieldsInTable(joinInfo.getCondition(), joinScope);
+        checkConditionFieldsInTable(joinInfo.getCondition(), joinScope, sideTableMap);
 
         //获取两个表的所有字段
         List<FieldInfo> sideJoinFieldInfo = ParserJoinField.getRowTypeInfo(joinInfo.getSelectNode(), joinScope, true);