[异步维表支持非等值连接]

Author: todd5167

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

@@ -109,8 +109,8 @@ public void parseSelectFields(JoinInfo joinInfo){
     }
 
     public void dealOneEqualCon(SqlNode sqlNode, String sideTableName){
-        if(sqlNode.getKind() != SqlKind.EQUALS){
-            throw new RuntimeException("not equal operator.");
+        if(!SqlKind.COMPARISON.contains(sqlNode.getKind())){
+            throw new RuntimeException("not compare operator.");
         }
 
         SqlIdentifier left = (SqlIdentifier)((SqlBasicCall)sqlNode).getOperands()[0];

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

@@ -585,8 +585,8 @@ public List<String> getConditionFields(SqlNode conditionNode, String specifyTabl
         ParseUtils.parseAnd(conditionNode, sqlNodeList);
         List<String> conditionFields = Lists.newArrayList();
         for(SqlNode sqlNode : sqlNodeList){
-            if(sqlNode.getKind() != SqlKind.EQUALS){
-                throw new RuntimeException("not equal operator.");
+            if (!SqlKind.COMPARISON.contains(sqlNode.getKind())) {
+                throw new RuntimeException("not compare operator.");
             }
 
             SqlIdentifier left = (SqlIdentifier)((SqlBasicCall)sqlNode).getOperands()[0];

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

@@ -3,8 +3,11 @@
 import org.apache.calcite.sql.SqlBasicCall;
 import org.apache.calcite.sql.SqlKind;
 import org.apache.calcite.sql.SqlNode;
+import org.apache.commons.lang3.StringUtils;
 
+import java.util.HashSet;
 import java.util.List;
+import java.util.Set;
 
 /**
  * @Auther: jiangjunjie
@@ -20,4 +23,25 @@ public static void parseAnd(SqlNode conditionNode, List<SqlNode> sqlNodeList){
             sqlNodeList.add(conditionNode);
         }
     }
+
+    public static void parseJoinCompareOperate(SqlNode condition, List<String> sqlJoinCompareOperate) {
+        SqlBasicCall joinCondition = (SqlBasicCall) condition;
+        if (joinCondition.getKind() == SqlKind.AND) {
+            List<SqlNode> operandList = joinCondition.getOperandList();
+            for (SqlNode sqlNode : operandList) {
+                parseJoinCompareOperate(sqlNode, sqlJoinCompareOperate);
+            }
+        } else {
+            String operator = parseOperator(joinCondition.getKind());
+            sqlJoinCompareOperate.add(operator);
+        }
+    }
+
+    public static String parseOperator(SqlKind sqlKind) {
+        if (StringUtils.equalsIgnoreCase(sqlKind.toString(), "NOT_EQUALS")){
+            return "!=";
+        }
+        return sqlKind.sql;
+    }
+
 }

oracle/oracle-side/oracle-async-side/src/main/java/com/dtstack/flink/sql/side/oracle/OracleAsyncSideInfo.java

@@ -24,6 +24,7 @@
 import com.dtstack.flink.sql.side.rdb.async.RdbAsyncSideInfo;
 import com.dtstack.flink.sql.side.rdb.table.RdbSideTableInfo;
 import com.dtstack.flink.sql.util.ParseUtils;
+import org.apache.calcite.sql.SqlKind;
 import org.apache.calcite.sql.SqlNode;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.flink.api.java.typeutils.RowTypeInfo;
@@ -48,7 +49,11 @@ public void buildEqualInfo(JoinInfo joinInfo, SideTableInfo sideTableInfo) {
         SqlNode conditionNode = joinInfo.getCondition();
 
         List<SqlNode> sqlNodeList = Lists.newArrayList();
+        List<String> sqlJoinCompareOperate= Lists.newArrayList();
+
         ParseUtils.parseAnd(conditionNode, sqlNodeList);
+        ParseUtils.parseJoinCompareOperate(conditionNode, sqlJoinCompareOperate);
+
 
         for (SqlNode sqlNode : sqlNodeList) {
             dealOneEqualCon(sqlNode, sideTableName);
@@ -57,8 +62,7 @@ public void buildEqualInfo(JoinInfo joinInfo, SideTableInfo sideTableInfo) {
         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));
-
-            sqlCondition += dealLowerFiled(equalField) + "=? ";
+            sqlCondition += dealLowerFiled(equalField) + " " + sqlJoinCompareOperate.get(i) + " " + " ?";
             if (i != equalFieldList.size() - 1) {
                 sqlCondition += " and ";
             }

oracle/oracle-sink/src/main/java/com/dtstack/flink/sql/sink/oracle/OracleSink.java

@@ -71,6 +71,8 @@ private void buildInsertSql(String tableName, List<String> fields) {
         placeholder = placeholder.replaceFirst(",", "");
         sqlTmp = sqlTmp.replace("${fields}", fieldsStr).replace("${placeholder}", placeholder);
         this.sql = sqlTmp;
+        System.out.println("--------insert sqlTmp--------");
+        System.out.println(sql);
     }
 
     @Override

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

@@ -58,7 +58,11 @@ public void buildEqualInfo(JoinInfo joinInfo, SideTableInfo sideTableInfo) {
         SqlNode conditionNode = joinInfo.getCondition();
 
         List<SqlNode> sqlNodeList = Lists.newArrayList();
+
+        List<String> sqlJoinCompareOperate= Lists.newArrayList();
+
         ParseUtils.parseAnd(conditionNode, sqlNodeList);
+        ParseUtils.parseJoinCompareOperate(conditionNode, sqlJoinCompareOperate);
 
         for (SqlNode sqlNode : sqlNodeList) {
             dealOneEqualCon(sqlNode, sideTableName);
@@ -68,7 +72,7 @@ public void buildEqualInfo(JoinInfo joinInfo, SideTableInfo sideTableInfo) {
         for (int i = 0; i < equalFieldList.size(); i++) {
             String equalField = sideTableInfo.getPhysicalFields().getOrDefault(equalFieldList.get(i), equalFieldList.get(i));
 
-            sqlCondition += equalField + "=? ";
+            sqlCondition += equalField + "\t" + sqlJoinCompareOperate.get(i) + " ？";
             if (i != equalFieldList.size() - 1) {
                 sqlCondition += " and ";
             }
@@ -81,8 +85,8 @@ public void buildEqualInfo(JoinInfo joinInfo, SideTableInfo sideTableInfo) {
 
     @Override
     public void dealOneEqualCon(SqlNode sqlNode, String sideTableName) {
-        if (sqlNode.getKind() != SqlKind.EQUALS) {
-            throw new RuntimeException("not equal operator.");
+        if (!SqlKind.COMPARISON.contains(sqlNode.getKind())) {
+            throw new RuntimeException("not compare operator.");
         }
 
         SqlIdentifier left = (SqlIdentifier) ((SqlBasicCall) sqlNode).getOperands()[0];