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

Author: gituser

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

@@ -310,6 +310,64 @@ private List<String> convertPrimaryAlias(AbstractSideTableInfo sideTableInfo) {
         return res;
     }
 
+    /**
+     * check whether all table fields exist in join condition.
+     * @param conditionNode
+     * @param joinScope
+     */
+    public void checkConditionFieldsInTable(SqlNode conditionNode, JoinScope joinScope) {
+        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.");
+            }
+
+            SqlNode leftNode = ((SqlBasicCall) sqlNode).getOperands()[0];
+            SqlNode rightNode = ((SqlBasicCall) sqlNode).getOperands()[1];
+
+            if (leftNode.getKind() == SqlKind.IDENTIFIER) {
+                checkFieldInTable((SqlIdentifier) leftNode, joinScope, conditionNode);
+            }
+
+            if (rightNode.getKind() == SqlKind.IDENTIFIER) {
+                checkFieldInTable((SqlIdentifier) rightNode, joinScope, conditionNode);
+            }
+
+        }
+    }
+
+    /**
+     * check whether table exists and whether field is in table.
+     * @param sqlNode
+     * @param joinScope
+     * @param conditionNode
+     */
+    private void checkFieldInTable(SqlIdentifier sqlNode, JoinScope joinScope, SqlNode conditionNode) {
+        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";
+        Preconditions.checkState(
+            scopeChild != null,
+            tableErrorMsg,
+            tableName,
+            conditionNode.toString(),
+            tableName
+        );
+
+        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]";
+        Preconditions.checkState(
+            hasField,
+            fieldErrorMsg,
+            tableName,
+            fieldName,
+            conditionNode.toString()
+        );
+    }
+
     public List<String> getConditionFields(SqlNode conditionNode, String specifyTableName, AbstractSideTableInfo sideTableInfo) {
         List<SqlNode> sqlNodeList = Lists.newArrayList();
         ParseUtils.parseAnd(conditionNode, sqlNodeList);
@@ -421,6 +479,7 @@ private void joinFun(Object pollObj,
         joinScope.addScope(rightScopeChild);
 
         HashBasedTable<String, String, String> mappingTable = ((JoinInfo) pollObj).getTableFieldRef();
+        checkConditionFieldsInTable(joinInfo.getCondition(), joinScope);
 
         //获取两个表的所有字段
         List<FieldInfo> sideJoinFieldInfo = ParserJoinField.getRowTypeInfo(joinInfo.getSelectNode(), joinScope, true);