实时计算谓词下推,关系型数据库异步方式

Author: simenliuxing

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

@@ -39,16 +39,70 @@
 import org.apache.calcite.sql.SqlBasicCall;
 import org.apache.calcite.sql.SqlKind;
 import org.apache.calcite.sql.SqlNode;
+import org.apache.calcite.sql.*;
+import org.apache.calcite.sql.parser.SqlParserPos;
 import org.apache.commons.lang3.StringUtils;
+import org.apache.flink.calcite.shaded.com.google.common.collect.Lists;
 
 import java.util.List;
+import java.util.Map;
 
 /**
  * @Auther: jiangjunjie
  * @Date: 2019-06-30 14:57
  * @Description:
  */
 public class ParseUtils {
+    public static void parseSideWhere(SqlNode whereNode, Map<String, String> physicalFields, List<String> whereConditionList) {
+        SqlKind sqlKind = whereNode.getKind();
+        if ((sqlKind == SqlKind.OR || sqlKind == SqlKind.AND) && ((SqlBasicCall) whereNode).getOperandList().size() == 2) {
+            SqlNode[] sqlOperandsList = ((SqlBasicCall) whereNode).getOperands();
+            // whereNode是一颗先解析or再解析and的二叉树。二叉树中序遍历，先左子树，其次中间节点，最后右子树
+            parseSideWhere(sqlOperandsList[0], physicalFields, whereConditionList);
+            whereConditionList.add(sqlKind.name());
+            parseSideWhere(sqlOperandsList[1], physicalFields, whereConditionList);
+        } else {
+            SqlIdentifier sqlIdentifier = (SqlIdentifier) ((SqlBasicCall) whereNode).getOperands()[0];
+            String fieldName = null;
+            if (sqlIdentifier.names.size() == 1) {
+                fieldName = sqlIdentifier.getComponent(0).getSimple();
+            } else {
+                fieldName = sqlIdentifier.getComponent(1).getSimple();
+            }
+            if (physicalFields.containsKey(fieldName)) {
+                String sideFieldName = physicalFields.get(fieldName);
+                // clone SqlIdentifier node
+                SqlParserPos sqlParserPos = new SqlParserPos(0, 0);
+                SqlIdentifier sqlIdentifierClone = new SqlIdentifier("", null, sqlParserPos);
+                List<String> namesClone = Lists.newArrayList();
+                for(String name :sqlIdentifier.names){
+                    namesClone.add(name);
+                }
+                sqlIdentifierClone.setNames(namesClone,null);
+                // clone SqlBasicCall node
+                SqlBasicCall sqlBasicCall = (SqlBasicCall)whereNode;
+                SqlNode[] sqlNodes =  sqlBasicCall.getOperands();
+                SqlNode[] sqlNodesClone = new SqlNode[sqlNodes.length];
+                for (int i = 0; i < sqlNodes.length; i++) {
+                    sqlNodesClone[i] = sqlNodes[i];
+                }
+                SqlBasicCall sqlBasicCallClone = new SqlBasicCall(sqlBasicCall.getOperator(), sqlNodesClone, sqlParserPos);
+                // 替换维表中真实字段名
+                List<String> names = Lists.newArrayList();
+                names.add(sideFieldName);
+                sqlIdentifierClone.setNames(names, null);
+
+                sqlBasicCallClone.setOperand(0, sqlIdentifierClone);
+                whereConditionList.add(sqlBasicCallClone.toString());
+            } else {
+                // 如果字段不是维表中字段，删除字段前的链接符
+                if (whereConditionList.size() >= 1) {
+                    whereConditionList.remove(whereConditionList.size() - 1);
+                }
+            }
+        }
+    }
+
     public static void parseAnd(SqlNode conditionNode, List<SqlNode> sqlNodeList){
         if(conditionNode.getKind() == SqlKind.AND && ((SqlBasicCall)conditionNode).getOperandList().size()==2){
             parseAnd(((SqlBasicCall)conditionNode).getOperands()[0], sqlNodeList);
@@ -78,4 +132,4 @@ public static String parseOperator(SqlKind sqlKind) {
         return sqlKind.sql;
     }
 
-}
+}

rdb/rdb-side/src/main/java/com/dtstack/flink/sql/side/rdb/async/RdbAsyncSideInfo.java

@@ -28,10 +28,14 @@
 import org.apache.calcite.sql.SqlIdentifier;
 import org.apache.calcite.sql.SqlKind;
 import org.apache.calcite.sql.SqlNode;
+import org.apache.calcite.sql.*;
 import org.apache.flink.api.java.typeutils.RowTypeInfo;
 import org.apache.flink.calcite.shaded.com.google.common.collect.Lists;
 
+import java.util.Arrays;
 import java.util.List;
+import java.util.Map;
+
 
 /**
  * Reason:
@@ -68,6 +72,14 @@ public void buildEqualInfo(JoinInfo joinInfo, SideTableInfo sideTableInfo) {
             dealOneEqualCon(sqlNode, sideTableName);
         }
 
+        List<String> whereConditionList = Lists.newArrayList();;
+        Map<String, String> physicalFields = rdbSideTableInfo.getPhysicalFields();
+        SqlNode whereNode = ((SqlSelect) joinInfo.getSelectNode()).getWhere();
+        if (whereNode != null) {
+            // 解析维表中的过滤条件
+            ParseUtils.parseSideWhere(whereNode, physicalFields, whereConditionList);
+        }
+
         sqlCondition = "select ${selectField} from ${tableName} where ";
         for (int i = 0; i < equalFieldList.size(); i++) {
             String equalField = sideTableInfo.getPhysicalFields().getOrDefault(equalFieldList.get(i), equalFieldList.get(i));
@@ -77,6 +89,14 @@ public void buildEqualInfo(JoinInfo joinInfo, SideTableInfo sideTableInfo) {
                 sqlCondition += " and ";
             }
         }
+        if (0 != whereConditionList.size()) {
+            // 如果where条件中第一个符合条件的是维表中的条件
+            String firstCondition = whereConditionList.get(0);
+            if (!"and".equalsIgnoreCase(firstCondition) && !"or".equalsIgnoreCase(firstCondition)) {
+                sqlCondition += " and ";
+            }
+            sqlCondition += String.join(" ", whereConditionList);
+        }
 
         sqlCondition = sqlCondition.replace("${tableName}", rdbSideTableInfo.getTableName()).replace("${selectField}", sideSelectFields);
     }
@@ -134,4 +154,4 @@ public void dealOneEqualCon(SqlNode sqlNode, String sideTableName) {
 
     }
 
-}
+}