Merge branch '1.10_release_4.1.x' into 1.10_release_4.2.x

Author: gituser

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

@@ -95,6 +95,8 @@ public abstract class AbstractSideTableInfo extends AbstractTableInfo implements
 
     private List<PredicateInfo>  predicateInfoes = Lists.newArrayList();
 
+    private List<PredicateInfo>  fullPredicateInfoes = Lists.newArrayList();
+
     private boolean fastCheck;
 
     public RowTypeInfo getRowTypeInfo(){
@@ -197,6 +199,14 @@ public void addPredicateInfo(PredicateInfo predicateInfo) {
         this.predicateInfoes.add(predicateInfo);
     }
 
+    public List<PredicateInfo> getFullPredicateInfoes() {
+        return fullPredicateInfoes;
+    }
+
+    public void addFullPredicateInfoes(PredicateInfo predicateInfo) {
+        this.fullPredicateInfoes.add(predicateInfo);
+    }
+
     public Long getAsyncFailMaxNum(Long defaultValue) {
         return Objects.isNull(asyncFailMaxNum) ? defaultValue : asyncFailMaxNum;
     }

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

@@ -194,6 +194,7 @@ private void evalConstantEquation(SqlLiteral literal, SqlIdentifier identifier,
             .setCondition(constant.toString())
             .build();
         sideTableInfo.addPredicateInfo(predicate);
+        sideTableInfo.addFullPredicateInfoes(predicate);
     }
 
     private void checkSupport(SqlIdentifier identifier) {

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

@@ -109,7 +109,7 @@ private void parseSql(SqlNode sqlNode, Map<String, AbstractSideTableInfo> sideTa
 
     private void fillToSideTableInfo(Map<String, AbstractSideTableInfo> sideTableMap, Map<String, String> tabMapping, List<PredicateInfo> predicateInfoList) {
         predicateInfoList.stream().filter(info -> sideTableMap.containsKey(tabMapping.getOrDefault(info.getOwnerTable(), info.getOwnerTable())))
-                .map(info -> sideTableMap.get(tabMapping.getOrDefault(info.getOwnerTable(), info.getOwnerTable())).getPredicateInfoes().add(info))
+                .map(info -> sideTableMap.get(tabMapping.getOrDefault(info.getOwnerTable(), info.getOwnerTable())).getFullPredicateInfoes().add(info))
                 .count();
     }
 

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

@@ -75,6 +75,7 @@
 import java.util.Map;
 import java.util.Queue;
 import java.util.Set;
+import java.util.stream.Collectors;
 
 import static org.apache.calcite.sql.SqlKind.AS;
 import static org.apache.calcite.sql.SqlKind.INSERT;
@@ -506,6 +507,8 @@ private void joinFun(Object pollObj,
             sideTableInfo = sideTableMap.get(joinInfo.getRightTableAlias());
         }
 
+        extractActualSidePredicateInfos(joinInfo, sideTableInfo);
+
         if (sideTableInfo == null) {
             throw new RuntimeException("can't not find side table:" + joinInfo.getRightTableName());
         }
@@ -592,6 +595,18 @@ private void joinFun(Object pollObj,
         }
     }
 
+    /**
+     * 抽取维表本次真正使用的谓词集合
+     * @param joinInfo 维表join信息
+     * @param sideTableInfo 维表实体信息
+     */
+    private void extractActualSidePredicateInfos(JoinInfo joinInfo, AbstractSideTableInfo sideTableInfo) {
+        sideTableInfo.setPredicateInfoes(sideTableInfo.getFullPredicateInfoes()
+                .stream()
+                .filter(e -> e.getOwnerTable().equals(joinInfo.getRightTableAlias()))
+                .collect(Collectors.toList()));
+    }
+
     private boolean checkFieldsInfo(CreateTmpTableParser.SqlParserResult result, Table table) {
         List<String> fieldNames = new LinkedList<>();
         String fieldsInfo = result.getFieldsInfoStr();

core/src/main/java/com/dtstack/flink/sql/util/TableUtils.java

@@ -758,8 +758,7 @@ public static void addConstant(Map<String, Object> keyMap, AbstractSideTableInfo
         List<PredicateInfo> predicateInfos = sideTableInfo.getPredicateInfoes();
         final String name = sideTableInfo.getName();
         for (PredicateInfo info : predicateInfos) {
-            if (info.getOwnerTable().equals(name)
-                && info.getOperatorName().equals("=")) {
+            if (info.getOperatorName().equals("=")) {
                 String condition = info.getCondition();
                 Matcher matcher = stringPattern.matcher(condition);
                 if (matcher.matches()) {