[fix] fix 多级Join Table Not Found

Author: FlechazoW

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

@@ -490,7 +490,7 @@ private Set<String> extractSelectFieldFromJoinCondition(Set<Tuple2<String, Strin
             }
 
             if(tableRef.containsKey(field.f0)){
-                if(fromTableNameSet.contains(tableRef.get(field.f0))){
+                if(checkContainIterationTableName(fromTableNameSet, field.f0, tableRef)){
                     extractFieldList.add(tableRef.get(field.f0) + "." + field.f1);
                 }
             }
@@ -499,6 +499,20 @@ private Set<String> extractSelectFieldFromJoinCondition(Set<Tuple2<String, Strin
         return extractFieldList;
     }
 
+    private boolean checkContainIterationTableName(Set<String> fromTableNameSet, String checkTableName, Map<String, String> mappingTableName) {
+        for (int i = 0; i < mappingTableName.size() + 1; i++) {
+            if (fromTableNameSet.contains(checkTableName)) {
+                return true;
+            }
+
+            checkTableName = mappingTableName.get(checkTableName);
+            if (checkTableName == null) {
+                return false;
+            }
+        }
+        return true;
+    }
+
     private Set<String> extractFieldFromGroupByList(SqlNodeList parentGroupByList,
                                                     Set<String> fromTableNameSet,
                                                     Map<String, String> tableRef){