all cache predicate pushdown

Author: todd5167

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

@@ -39,7 +39,7 @@ public class StreamSideFactory {
 
     public static AbsTableParser getSqlParser(String pluginType, String sqlRootDir, String cacheType) throws Exception {
 
-        String sideOperator = ECacheType.ALL.name().equals(cacheType) ? "all" : "async";
+        String sideOperator = ECacheType.ALL.name().equalsIgnoreCase(cacheType) ? "all" : "async";
         String pluginJarPath = PluginUtil.getSideJarFileDirPath(pluginType, sideOperator, "side", sqlRootDir);
         String className = PluginUtil.getSqlParserClassName(pluginType, CURR_TYPE);
 

oracle/oracle-side/oracle-all-side/src/main/java/com/dtstack/flink/sql/side/oracle/OracleAllSideInfo.java

@@ -35,26 +35,12 @@ public OracleAllSideInfo(RowTypeInfo rowTypeInfo, JoinInfo joinInfo, List<FieldI
     }
 
     @Override
-    public void buildEqualInfo(JoinInfo joinInfo, SideTableInfo sideTableInfo) {
-        RdbSideTableInfo rdbSideTableInfo = (RdbSideTableInfo) sideTableInfo;
-
-        sqlCondition = "select ${selectField} from ${tableName} ";
-
-
-        sqlCondition = sqlCondition.replace("${tableName}", DtStringUtil.getTableFullPath(rdbSideTableInfo.getSchema(), rdbSideTableInfo.getTableName())).replace("${selectField}", dealLowerSelectFiled(sideSelectFields));
-        System.out.println("---------side_exe_sql-----\n" + sqlCondition);
+    public String getTableName(RdbSideTableInfo rdbSideTableInfo) {
+        return DtStringUtil.getTableFullPath(rdbSideTableInfo.getSchema(), rdbSideTableInfo.getTableName());
     }
 
-
-    private String dealLowerSelectFiled(String fieldsStr) {
-        StringBuilder sb = new StringBuilder();
-        String[] fields = fieldsStr.split(",");
-
-        for(String f : fields) {
-            sb.append("\"").append(f).append("\"").append(",");
-        }
-
-        sb.deleteCharAt(sb.lastIndexOf(","));
-        return  sb.toString();
+    @Override
+    public  String quoteIdentifier(String identifier) {
+        return "\"" + identifier + "\"";
     }
 }

rdb/rdb-side/src/main/java/com/dtstack/flink/sql/side/rdb/all/RdbAllSideInfo.java

@@ -20,6 +20,7 @@
 
 import com.dtstack.flink.sql.side.FieldInfo;
 import com.dtstack.flink.sql.side.JoinInfo;
+import com.dtstack.flink.sql.side.PredicateInfo;
 import com.dtstack.flink.sql.side.SideInfo;
 import com.dtstack.flink.sql.side.SideTableInfo;
 import com.dtstack.flink.sql.side.rdb.table.RdbSideTableInfo;
@@ -29,7 +30,9 @@
 import org.apache.flink.api.java.typeutils.RowTypeInfo;
 import com.google.common.collect.Lists;
 
+import java.util.Arrays;
 import java.util.List;
+import java.util.stream.Collectors;
 
 /**
  * Reason:
@@ -52,8 +55,15 @@ public RdbAllSideInfo(RowTypeInfo rowTypeInfo, JoinInfo joinInfo, List<FieldInfo
     public void buildEqualInfo(JoinInfo joinInfo, SideTableInfo sideTableInfo) {
         RdbSideTableInfo rdbSideTableInfo = (RdbSideTableInfo) sideTableInfo;
 
-        sqlCondition = "select ${selectField} from ${tableName} ";
-        sqlCondition = sqlCondition.replace("${tableName}", rdbSideTableInfo.getTableName()).replace("${selectField}", sideSelectFields);
+        sqlCondition = getSelectFromStatement(getTableName(rdbSideTableInfo), Arrays.asList(sideSelectFields.split(",")), sideTableInfo.getPredicateInfoes());
+        System.out.println("--------side sql query-------\n" + sqlCondition);
+    }
+
+    public String getSelectFromStatement(String tableName, List<String> selectFields,  List<PredicateInfo> predicateInfoes) {
+        String fromClause = selectFields.stream().map(this::quoteIdentifier).collect(Collectors.joining(", "));
+        String predicateClause = predicateInfoes.stream().map(this::buildFilterCondition).collect(Collectors.joining(" AND "));
+        String sql = "SELECT " + fromClause + " FROM " + tableName + (predicateInfoes.size() > 0 ? " WHERE " + predicateClause : "");
+        return sql;
     }
 
     @Override
@@ -107,4 +117,29 @@ public void parseSelectFields(JoinInfo joinInfo) {
 
         sideSelectFields = String.join(",", fields);
     }
+
+    public String buildFilterCondition(PredicateInfo info) {
+        switch (info.getOperatorKind()) {
+            case "IN":
+            case "NOT_IN":
+                return quoteIdentifier(info.getFieldName()) + " " + info.getOperatorName() + " ( " + info.getCondition() + " )";
+            case "NOT_EQUALS":
+                return quoteIdentifier(info.getFieldName()) + " != " + info.getCondition();
+            case "BETWEEN":
+                return quoteIdentifier(info.getFieldName()) + " BETWEEN  " + info.getCondition();
+            case "IS_NOT_NULL":
+            case "IS_NULL":
+                return quoteIdentifier(info.getFieldName()) + " " + info.getOperatorName();
+            default:
+                return quoteIdentifier(info.getFieldName()) + " " + info.getOperatorName() + " " + info.getCondition();
+        }
+    }
+
+    public String getTableName(RdbSideTableInfo rdbSideTableInfo) {
+        return rdbSideTableInfo.getTableName();
+    }
+
+    public String quoteIdentifier(String identifier) {
+        return " " + identifier + " ";
+    }
 }