Merge branch 'v1.5.0_oracle_field_case' into 'v1.5.0_dev'

zoudaokoulife · zoudaokoulife · commit c098bd930945 · 2019-07-12T14:14:26.000+08:00
oracle 小写字段，表名转义



See merge request !30
diff --git a/mysql/mysql-side/mysql-all-side/src/main/java/com/dtstack/flink/sql/side/mysql/MysqlAllReqRow.java b/mysql/mysql-side/mysql-all-side/src/main/java/com/dtstack/flink/sql/side/mysql/MysqlAllReqRow.java
@@ -50,7 +50,7 @@ public class MysqlAllReqRow extends RdbAllReqRow {
     private static final String MYSQL_DRIVER = "com.mysql.jdbc.Driver";
 
     public MysqlAllReqRow(RowTypeInfo rowTypeInfo, JoinInfo joinInfo, List<FieldInfo> outFieldInfoList, SideTableInfo sideTableInfo) {
-        super(rowTypeInfo, joinInfo, outFieldInfoList, sideTableInfo);
+        super(new MysqlAllSideInfo(rowTypeInfo, joinInfo, outFieldInfoList, sideTableInfo));
     }
 
     @Override
diff --git a/mysql/mysql-side/mysql-async-side/src/main/java/com/dtstack/flink/sql/side/mysql/MysqlAsyncReqRow.java b/mysql/mysql-side/mysql-async-side/src/main/java/com/dtstack/flink/sql/side/mysql/MysqlAsyncReqRow.java
@@ -50,7 +50,7 @@ public class MysqlAsyncReqRow extends RdbAsyncReqRow {
     private final static String MYSQL_DRIVER = "com.mysql.jdbc.Driver";
 
     public MysqlAsyncReqRow(RowTypeInfo rowTypeInfo, JoinInfo joinInfo, List<FieldInfo> outFieldInfoList, SideTableInfo sideTableInfo) {
-        super(rowTypeInfo, joinInfo, outFieldInfoList, sideTableInfo);
+        super(new MysqlAsyncSideInfo(rowTypeInfo, joinInfo, outFieldInfoList, sideTableInfo));
     }
 
 
diff --git a/oracle/oracle-side/oracle-all-side/src/main/java/com/dtstack/flink/sql/side/oracle/OracleAllReqRow.java b/oracle/oracle-side/oracle-all-side/src/main/java/com/dtstack/flink/sql/side/oracle/OracleAllReqRow.java
@@ -43,7 +43,7 @@ public class OracleAllReqRow extends RdbAllReqRow {
     private static final String ORACLE_DRIVER = "oracle.jdbc.driver.OracleDriver";
 
     public OracleAllReqRow(RowTypeInfo rowTypeInfo, JoinInfo joinInfo, List<FieldInfo> outFieldInfoList, SideTableInfo sideTableInfo) {
-        super(rowTypeInfo, joinInfo, outFieldInfoList, sideTableInfo);
+        super(new OracleAllSideInfo(rowTypeInfo, joinInfo, outFieldInfoList, sideTableInfo));
     }
 
     @Override
diff --git a/oracle/oracle-side/oracle-all-side/src/main/java/com/dtstack/flink/sql/side/oracle/OracleAllSideInfo.java b/oracle/oracle-side/oracle-all-side/src/main/java/com/dtstack/flink/sql/side/oracle/OracleAllSideInfo.java
@@ -21,6 +21,8 @@
 import com.dtstack.flink.sql.side.JoinInfo;
 import com.dtstack.flink.sql.side.SideTableInfo;
 import com.dtstack.flink.sql.side.rdb.all.RdbAllSideInfo;
+import com.dtstack.flink.sql.side.rdb.table.RdbSideTableInfo;
+import org.apache.commons.lang3.StringUtils;
 import org.apache.flink.api.java.typeutils.RowTypeInfo;
 
 import java.util.List;
@@ -30,4 +32,32 @@ public class OracleAllSideInfo extends RdbAllSideInfo {
     public OracleAllSideInfo(RowTypeInfo rowTypeInfo, JoinInfo joinInfo, List<FieldInfo> outFieldInfoList, SideTableInfo sideTableInfo) {
         super(rowTypeInfo, joinInfo, outFieldInfoList, sideTableInfo);
     }
+
+    @Override
+    public void buildEqualInfo(JoinInfo joinInfo, SideTableInfo sideTableInfo) {
+        RdbSideTableInfo rdbSideTableInfo = (RdbSideTableInfo) sideTableInfo;
+
+        sqlCondition = "select ${selectField} from ${tableName} ";
+
+
+        sqlCondition = sqlCondition.replace("${tableName}", dealFiled(rdbSideTableInfo.getTableName())).replace("${selectField}", dealLowerSelectFiled(sideSelectFields));
+        System.out.println("---------side_exe_sql-----\n" + sqlCondition);
+    }
+
+
+    private String dealFiled(String field) {
+        return   "\"" + field + "\"";
+    }
+
+    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();
+    }
 }
diff --git a/oracle/oracle-side/oracle-async-side/src/main/java/com/dtstack/flink/sql/side/oracle/OracleAsyncReqRow.java b/oracle/oracle-side/oracle-async-side/src/main/java/com/dtstack/flink/sql/side/oracle/OracleAsyncReqRow.java
@@ -43,15 +43,15 @@ public class OracleAsyncReqRow extends RdbAsyncReqRow {
     private static final String ORACLE_DRIVER = "oracle.jdbc.driver.OracleDriver";
 
     public OracleAsyncReqRow(RowTypeInfo rowTypeInfo, JoinInfo joinInfo, List<FieldInfo> outFieldInfoList, SideTableInfo sideTableInfo) {
-        super(rowTypeInfo, joinInfo, outFieldInfoList, sideTableInfo);
+        super(new OracleAsyncSideInfo(rowTypeInfo, joinInfo, outFieldInfoList, sideTableInfo));
     }
 
     @Override
     public void open(Configuration parameters) throws Exception {
         super.open(parameters);
-        JsonObject sqlserverClientConfig = new JsonObject();
+        JsonObject oracleClientConfig = new JsonObject();
         RdbSideTableInfo rdbSideTableInfo = (RdbSideTableInfo) sideInfo.getSideTableInfo();
-        sqlserverClientConfig.put("url", rdbSideTableInfo.getUrl())
+        oracleClientConfig.put("url", rdbSideTableInfo.getUrl())
                 .put("driver_class", ORACLE_DRIVER)
                 .put("max_pool_size", DEFAULT_MAX_DB_CONN_POOL_SIZE)
                 .put("user", rdbSideTableInfo.getUserName())
@@ -61,6 +61,6 @@ public void open(Configuration parameters) throws Exception {
         vo.setEventLoopPoolSize(DEFAULT_VERTX_EVENT_LOOP_POOL_SIZE);
         vo.setWorkerPoolSize(DEFAULT_VERTX_WORKER_POOL_SIZE);
         Vertx vertx = Vertx.vertx(vo);
-        setRdbSQLClient(JDBCClient.createNonShared(vertx, sqlserverClientConfig));
+        setRdbSQLClient(JDBCClient.createNonShared(vertx, oracleClientConfig));
     }
 }
diff --git a/oracle/oracle-side/oracle-async-side/src/main/java/com/dtstack/flink/sql/side/oracle/OracleAsyncSideInfo.java b/oracle/oracle-side/oracle-async-side/src/main/java/com/dtstack/flink/sql/side/oracle/OracleAsyncSideInfo.java
@@ -22,8 +22,14 @@
 import com.dtstack.flink.sql.side.JoinInfo;
 import com.dtstack.flink.sql.side.SideTableInfo;
 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.SqlNode;
+import org.apache.commons.lang3.StringUtils;
 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;
 
 
@@ -32,4 +38,50 @@ public class OracleAsyncSideInfo extends RdbAsyncSideInfo {
     public OracleAsyncSideInfo(RowTypeInfo rowTypeInfo, JoinInfo joinInfo, List<FieldInfo> outFieldInfoList, SideTableInfo sideTableInfo) {
         super(rowTypeInfo, joinInfo, outFieldInfoList, sideTableInfo);
     }
+
+    @Override
+    public void buildEqualInfo(JoinInfo joinInfo, SideTableInfo sideTableInfo) {
+        RdbSideTableInfo rdbSideTableInfo = (RdbSideTableInfo) sideTableInfo;
+
+        String sideTableName = joinInfo.getSideTableName();
+
+        SqlNode conditionNode = joinInfo.getCondition();
+
+        List<SqlNode> sqlNodeList = Lists.newArrayList();
+        ParseUtils.parseAnd(conditionNode, sqlNodeList);
+
+        for (SqlNode sqlNode : sqlNodeList) {
+            dealOneEqualCon(sqlNode, sideTableName);
+        }
+
+        sqlCondition = "select ${selectField} from ${tableName} where ";
+        for (int i = 0; i < equalFieldList.size(); i++) {
+            String equalField = equalFieldList.get(i);
+
+            sqlCondition += dealLowerFiled(equalField) + "=? ";
+            if (i != equalFieldList.size() - 1) {
+                sqlCondition += " and ";
+            }
+        }
+
+        sqlCondition = sqlCondition.replace("${tableName}", dealLowerFiled(rdbSideTableInfo.getTableName())).replace("${selectField}", dealLowerSelectFiled(sideSelectFields));
+        System.out.println("---------side_exe_sql-----\n" + sqlCondition);
+    }
+
+    private String dealLowerFiled(String field) {
+        return   "\"" + field + "\"";
+    }
+
+    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();
+    }
+
 }
diff --git a/rdb/rdb-side/src/main/java/com/dtstack/flink/sql/side/rdb/all/RdbAllReqRow.java b/rdb/rdb-side/src/main/java/com/dtstack/flink/sql/side/rdb/all/RdbAllReqRow.java
@@ -19,16 +19,11 @@
 package com.dtstack.flink.sql.side.rdb.all;
 
 import com.dtstack.flink.sql.side.AllReqRow;
-import com.dtstack.flink.sql.side.FieldInfo;
-import com.dtstack.flink.sql.side.JoinInfo;
-import com.dtstack.flink.sql.side.SideTableInfo;
+import com.dtstack.flink.sql.side.SideInfo;
 import com.dtstack.flink.sql.side.rdb.table.RdbSideTableInfo;
-import com.dtstack.flink.sql.side.rdb.util.MathUtil;
 import com.dtstack.flink.sql.side.rdb.util.SwitchUtil;
 import org.apache.calcite.sql.JoinType;
 import org.apache.commons.collections.CollectionUtils;
-import org.apache.flink.api.common.typeinfo.TypeInformation;
-import org.apache.flink.api.java.typeutils.RowTypeInfo;
 import org.apache.flink.calcite.shaded.com.google.common.collect.Lists;
 import org.apache.flink.calcite.shaded.com.google.common.collect.Maps;
 import org.apache.flink.table.typeutils.TimeIndicatorTypeInfo;
@@ -37,8 +32,11 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import java.math.BigDecimal;
-import java.sql.*;
+import java.sql.Connection;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.sql.Timestamp;
 import java.util.Calendar;
 import java.util.List;
 import java.util.Map;
@@ -62,9 +60,8 @@ public abstract class RdbAllReqRow extends AllReqRow {
 
     private AtomicReference<Map<String, List<Map<String, Object>>>> cacheRef = new AtomicReference<>();
 
-
-    public RdbAllReqRow(RowTypeInfo rowTypeInfo, JoinInfo joinInfo, List<FieldInfo> outFieldInfoList, SideTableInfo sideTableInfo) {
-        super(new RdbAllSideInfo(rowTypeInfo, joinInfo, outFieldInfoList, sideTableInfo));
+    public RdbAllReqRow(SideInfo sideInfo) {
+        super(sideInfo);
     }
 
     @Override
diff --git a/rdb/rdb-side/src/main/java/com/dtstack/flink/sql/side/rdb/async/RdbAsyncReqRow.java b/rdb/rdb-side/src/main/java/com/dtstack/flink/sql/side/rdb/async/RdbAsyncReqRow.java
@@ -59,8 +59,8 @@ public class RdbAsyncReqRow extends AsyncReqRow {
 
     private transient SQLClient rdbSQLClient;
 
-    public RdbAsyncReqRow(RowTypeInfo rowTypeInfo, JoinInfo joinInfo, List<FieldInfo> outFieldInfoList, SideTableInfo sideTableInfo) {
-        super(new RdbAsyncSideInfo(rowTypeInfo, joinInfo, outFieldInfoList, sideTableInfo));
+    public RdbAsyncReqRow(SideInfo sideInfo) {
+        super(sideInfo);
     }
 
     @Override
diff --git a/sqlserver/sqlserver-side/sqlserver-all-side/src/main/java/com/dtstack/flink/sql/side/sqlserver/SqlserverAllReqRow.java b/sqlserver/sqlserver-side/sqlserver-all-side/src/main/java/com/dtstack/flink/sql/side/sqlserver/SqlserverAllReqRow.java
@@ -43,7 +43,7 @@ public class SqlserverAllReqRow extends RdbAllReqRow {
     private static final String SQLSERVER_DRIVER = "net.sourceforge.jtds.jdbc.Driver";
 
     public SqlserverAllReqRow(RowTypeInfo rowTypeInfo, JoinInfo joinInfo, List<FieldInfo> outFieldInfoList, SideTableInfo sideTableInfo) {
-        super(rowTypeInfo, joinInfo, outFieldInfoList, sideTableInfo);
+        super(new SqlserverAllSideInfo(rowTypeInfo, joinInfo, outFieldInfoList, sideTableInfo));
     }
 
     @Override
diff --git a/sqlserver/sqlserver-side/sqlserver-async-side/src/main/java/com/dtstack/flink/sql/side/sqlserver/SqlserverAsyncReqRow.java b/sqlserver/sqlserver-side/sqlserver-async-side/src/main/java/com/dtstack/flink/sql/side/sqlserver/SqlserverAsyncReqRow.java
@@ -41,7 +41,7 @@ public class SqlserverAsyncReqRow extends RdbAsyncReqRow {
     private final static String SQLSERVER_DRIVER = "net.sourceforge.jtds.jdbc.Driver";
 
     public SqlserverAsyncReqRow(RowTypeInfo rowTypeInfo, JoinInfo joinInfo, List<FieldInfo> outFieldInfoList, SideTableInfo sideTableInfo) {
-        super(rowTypeInfo, joinInfo, outFieldInfoList, sideTableInfo);
+        super(new SqlserverAsyncSideInfo(rowTypeInfo, joinInfo, outFieldInfoList, sideTableInfo));
     }
 
     @Override

Original file line number	Diff line number	Diff line change
`@@ -50,7 +50,7 @@ public class MysqlAllReqRow extends RdbAllReqRow {`
`50`	`50`	`private static final String MYSQL_DRIVER = "com.mysql.jdbc.Driver";`
`51`	`51`
`52`	`52`	`public MysqlAllReqRow(RowTypeInfo rowTypeInfo, JoinInfo joinInfo, List<FieldInfo> outFieldInfoList, SideTableInfo sideTableInfo) {`
`53`		`- super(rowTypeInfo, joinInfo, outFieldInfoList, sideTableInfo);`
	`53`	`+ super(new MysqlAllSideInfo(rowTypeInfo, joinInfo, outFieldInfoList, sideTableInfo));`
`54`	`54`	`}`
`55`	`55`
`56`	`56`	`@Override`
Original file line number	Diff line number	Diff line change
`@@ -50,7 +50,7 @@ public class MysqlAsyncReqRow extends RdbAsyncReqRow {`
`50`	`50`	`private final static String MYSQL_DRIVER = "com.mysql.jdbc.Driver";`
`51`	`51`
`52`	`52`	`public MysqlAsyncReqRow(RowTypeInfo rowTypeInfo, JoinInfo joinInfo, List<FieldInfo> outFieldInfoList, SideTableInfo sideTableInfo) {`
`53`		`- super(rowTypeInfo, joinInfo, outFieldInfoList, sideTableInfo);`
	`53`	`+ super(new MysqlAsyncSideInfo(rowTypeInfo, joinInfo, outFieldInfoList, sideTableInfo));`
`54`	`54`	`}`
`55`	`55`
`56`	`56`
Original file line number	Diff line number	Diff line change
`@@ -43,7 +43,7 @@ public class OracleAllReqRow extends RdbAllReqRow {`
`43`	`43`	`private static final String ORACLE_DRIVER = "oracle.jdbc.driver.OracleDriver";`
`44`	`44`
`45`	`45`	`public OracleAllReqRow(RowTypeInfo rowTypeInfo, JoinInfo joinInfo, List<FieldInfo> outFieldInfoList, SideTableInfo sideTableInfo) {`
`46`		`- super(rowTypeInfo, joinInfo, outFieldInfoList, sideTableInfo);`
	`46`	`+ super(new OracleAllSideInfo(rowTypeInfo, joinInfo, outFieldInfoList, sideTableInfo));`
`47`	`47`	`}`
`48`	`48`
`49`	`49`	`@Override`
Original file line number	Diff line number	Diff line change
`@@ -59,8 +59,8 @@ public class RdbAsyncReqRow extends AsyncReqRow {`
`59`	`59`
`60`	`60`	`private transient SQLClient rdbSQLClient;`
`61`	`61`
`62`		`- public RdbAsyncReqRow(RowTypeInfo rowTypeInfo, JoinInfo joinInfo, List<FieldInfo> outFieldInfoList, SideTableInfo sideTableInfo) {`
`63`		`- super(new RdbAsyncSideInfo(rowTypeInfo, joinInfo, outFieldInfoList, sideTableInfo));`
	`62`	`+ public RdbAsyncReqRow(SideInfo sideInfo) {`
	`63`	`+ super(sideInfo);`
`64`	`64`	`}`
`65`	`65`
`66`	`66`	`@Override`
Original file line number	Diff line number	Diff line change
`@@ -43,7 +43,7 @@ public class SqlserverAllReqRow extends RdbAllReqRow {`
`43`	`43`	`private static final String SQLSERVER_DRIVER = "net.sourceforge.jtds.jdbc.Driver";`
`44`	`44`
`45`	`45`	`public SqlserverAllReqRow(RowTypeInfo rowTypeInfo, JoinInfo joinInfo, List<FieldInfo> outFieldInfoList, SideTableInfo sideTableInfo) {`
`46`		`- super(rowTypeInfo, joinInfo, outFieldInfoList, sideTableInfo);`
	`46`	`+ super(new SqlserverAllSideInfo(rowTypeInfo, joinInfo, outFieldInfoList, sideTableInfo));`
`47`	`47`	`}`
`48`	`48`
`49`	`49`	`@Override`
Original file line number	Diff line number	Diff line change
`@@ -41,7 +41,7 @@ public class SqlserverAsyncReqRow extends RdbAsyncReqRow {`
`41`	`41`	`private final static String SQLSERVER_DRIVER = "net.sourceforge.jtds.jdbc.Driver";`
`42`	`42`
`43`	`43`	`public SqlserverAsyncReqRow(RowTypeInfo rowTypeInfo, JoinInfo joinInfo, List<FieldInfo> outFieldInfoList, SideTableInfo sideTableInfo) {`
`44`		`- super(rowTypeInfo, joinInfo, outFieldInfoList, sideTableInfo);`
	`44`	`+ super(new SqlserverAsyncSideInfo(rowTypeInfo, joinInfo, outFieldInfoList, sideTableInfo));`
`45`	`45`	`}`
`46`	`46`
`47`	`47`	`@Override`