Merge branch 'v1.5.0_dev' of github.com:DTStack/flinkStreamSQL into v1.5.0_dev

yanxi0227 · yanxi0227 · commit 26bbe359cc5a · 2018-12-13T15:17:12.000+08:00
diff --git a/core/src/main/java/com/dtstack/flink/sql/Main.java b/core/src/main/java/com/dtstack/flink/sql/Main.java
@@ -16,7 +16,7 @@
  * limitations under the License.
  */
 
- 
+
 
 package com.dtstack.flink.sql;
 
@@ -34,6 +34,7 @@
 import com.dtstack.flink.sql.watermarker.WaterMarkerAssigner;
 import com.dtstack.flink.sql.util.FlinkUtil;
 import com.dtstack.flink.sql.util.PluginUtil;
+import org.apache.calcite.config.Lex;
 import org.apache.calcite.sql.SqlInsert;
 import org.apache.calcite.sql.SqlNode;
 import org.apache.commons.cli.CommandLine;
@@ -185,7 +186,12 @@ public static void main(String[] args) throws Exception {
                 if (sqlTree.getTmpTableMap().containsKey(tableName)) {
                     CreateTmpTableParser.SqlParserResult tmp = sqlTree.getTmpTableMap().get(tableName);
                     String realSql = DtStringUtil.replaceIgnoreQuota(result.getExecSql(), "`", "");
-                    SqlNode sqlNode = org.apache.calcite.sql.parser.SqlParser.create(realSql).parseStmt();
+
+                    org.apache.calcite.sql.parser.SqlParser.Config config = org.apache.calcite.sql.parser.SqlParser
+                            .configBuilder()
+                            .setLex(Lex.MYSQL)
+                            .build();
+                    SqlNode sqlNode = org.apache.calcite.sql.parser.SqlParser.create(realSql,config).parseStmt();
                     String tmpSql = ((SqlInsert) sqlNode).getSource().toString();
                     tmp.setExecSql(tmpSql);
                     sideSqlExec.registerTmpTable(tmp, sideTableMap, tableEnv, registerTableCache);
@@ -246,7 +252,7 @@ private static void registerUDF(SqlTree sqlTree, List<URL> jarURList, URLClassLo
                 classLoader = FlinkUtil.loadExtraJar(jarURList, parentClassloader);
             }
             classLoader.loadClass(funcInfo.getClassName());
-            FlinkUtil.registerUDF(funcInfo.getType(), funcInfo.getClassName(), funcInfo.getName().toUpperCase(),
+            FlinkUtil.registerUDF(funcInfo.getType(), funcInfo.getClassName(), funcInfo.getName(),
                     tableEnv, classLoader);
         }
     }
diff --git a/core/src/main/java/com/dtstack/flink/sql/parser/CreateTableParser.java b/core/src/main/java/com/dtstack/flink/sql/parser/CreateTableParser.java
@@ -54,7 +54,7 @@ public boolean verify(String sql) {
     public void parseSql(String sql, SqlTree sqlTree) {
         Matcher matcher = PATTERN.matcher(sql);
         if(matcher.find()){
-            String tableName = matcher.group(1).toUpperCase();
+            String tableName = matcher.group(1);
             String fieldsInfoStr = matcher.group(2);
             String propsStr = matcher.group(3);
             Map<String, Object> props = parseProp(propsStr);
diff --git a/core/src/main/java/com/dtstack/flink/sql/parser/CreateTmpTableParser.java b/core/src/main/java/com/dtstack/flink/sql/parser/CreateTmpTableParser.java
@@ -21,6 +21,7 @@
 package com.dtstack.flink.sql.parser;
 
 import com.dtstack.flink.sql.util.DtStringUtil;
+import org.apache.calcite.config.Lex;
 import org.apache.calcite.sql.*;
 import org.apache.calcite.sql.parser.SqlParseException;
 import org.apache.calcite.sql.parser.SqlParser;
@@ -62,11 +63,14 @@ public void parseSql(String sql, SqlTree sqlTree) {
             String tableName = null;
             String selectSql = null;
             if(matcher.find()) {
-                tableName = matcher.group(1).toUpperCase();
+                tableName = matcher.group(1);
                 selectSql = "select " + matcher.group(2);
             }
-
-            SqlParser sqlParser = SqlParser.create(selectSql);
+            SqlParser.Config config = SqlParser
+                    .configBuilder()
+                    .setLex(Lex.MYSQL)
+                    .build();
+            SqlParser sqlParser = SqlParser.create(selectSql,config);
             SqlNode sqlNode = null;
             try {
                 sqlNode = sqlParser.parseStmt();
@@ -89,7 +93,7 @@ public void parseSql(String sql, SqlTree sqlTree) {
                 String tableName = null;
                 String fieldsInfoStr = null;
                 if (matcher.find()){
-                    tableName = matcher.group(1).toUpperCase();
+                    tableName = matcher.group(1);
                     fieldsInfoStr = matcher.group(2);
                 }
                 CreateTmpTableParser.SqlParserResult sqlParseResult = new CreateTmpTableParser.SqlParserResult();
diff --git a/core/src/main/java/com/dtstack/flink/sql/parser/InsertSqlParser.java b/core/src/main/java/com/dtstack/flink/sql/parser/InsertSqlParser.java
@@ -20,6 +20,7 @@
 
 package com.dtstack.flink.sql.parser;
 
+import org.apache.calcite.config.Lex;
 import org.apache.calcite.sql.SqlBasicCall;
 import org.apache.calcite.sql.SqlInsert;
 import org.apache.calcite.sql.SqlJoin;
@@ -57,7 +58,11 @@ public static InsertSqlParser newInstance(){
 
     @Override
     public void parseSql(String sql, SqlTree sqlTree) {
-        SqlParser sqlParser = SqlParser.create(sql);
+        SqlParser.Config config = SqlParser
+                .configBuilder()
+                .setLex(Lex.MYSQL)
+                .build();
+        SqlParser sqlParser = SqlParser.create(sql,config);
         SqlNode sqlNode = null;
         try {
             sqlNode = sqlParser.parseStmt();
diff --git a/core/src/main/java/com/dtstack/flink/sql/side/SideSQLParser.java b/core/src/main/java/com/dtstack/flink/sql/side/SideSQLParser.java
@@ -21,6 +21,7 @@
 package com.dtstack.flink.sql.side;
 
 import com.dtstack.flink.sql.util.DtStringUtil;
+import org.apache.calcite.config.Lex;
 import org.apache.calcite.sql.JoinType;
 import org.apache.calcite.sql.SqlAsOperator;
 import org.apache.calcite.sql.SqlBasicCall;
@@ -52,11 +53,14 @@
 public class SideSQLParser {
 
     public Queue<Object> getExeQueue(String exeSql, Set<String> sideTableSet) throws SqlParseException {
-        exeSql = DtStringUtil.replaceIgnoreQuota(exeSql, "`", "");
         System.out.println("---exeSql---");
         System.out.println(exeSql);
         Queue<Object> queueInfo = Queues.newLinkedBlockingQueue();
-        SqlParser sqlParser = SqlParser.create(exeSql);
+        SqlParser.Config config = SqlParser
+                .configBuilder()
+                .setLex(Lex.MYSQL)
+                .build();
+        SqlParser sqlParser = SqlParser.create(exeSql,config);
         SqlNode sqlNode = sqlParser.parseStmt();
         parseSql(sqlNode, sideTableSet, queueInfo);
         queueInfo.offer(sqlNode);
diff --git a/core/src/main/java/com/dtstack/flink/sql/side/SideSqlExec.java b/core/src/main/java/com/dtstack/flink/sql/side/SideSqlExec.java
@@ -645,7 +645,7 @@ private boolean checkFieldsInfo(CreateTmpTableParser.SqlParserResult result, Tab
                 String[] filedNameArr = new String[filed.length - 1];
                 System.arraycopy(filed, 0, filedNameArr, 0, filed.length - 1);
                 String fieldName = String.join(" ", filedNameArr);
-                fieldNames.add(fieldName.toUpperCase());
+                fieldNames.add(fieldName);
                 String fieldType = filed[filed.length - 1 ].trim();
                 Class fieldClass = ClassUtil.stringConvertClass(fieldType);
                 Class tableField = table.getSchema().getType(i).get().getTypeClass();
diff --git a/core/src/main/java/com/dtstack/flink/sql/table/AbsTableParser.java b/core/src/main/java/com/dtstack/flink/sql/table/AbsTableParser.java
@@ -82,10 +82,6 @@ public void parseFieldsInfo(String fieldsInfo, TableInfo tableInfo){
         String[] fieldRows = DtStringUtil.splitIgnoreQuotaBrackets(fieldsInfo, ",");
         for(String fieldRow : fieldRows){
             fieldRow = fieldRow.trim();
-            if(fieldNameNeedsUpperCase()) {
-                fieldRow = fieldRow.toUpperCase();
-            }
-
             boolean isMatcherKey = dealKeyPattern(fieldRow, tableInfo);
 
             if(isMatcherKey){
diff --git a/oracle/oracle-sink/src/main/java/com/dtstack/flink/sql/sink/oracle/OracleSink.java b/oracle/oracle-sink/src/main/java/com/dtstack/flink/sql/sink/oracle/OracleSink.java
@@ -24,6 +24,7 @@
 import org.apache.commons.lang3.StringUtils;
 
 import java.util.ArrayList;
+import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
 
@@ -94,7 +95,7 @@ protected List<String> keyColList(Map<String, List<String>> updateKey) {
         for (Map.Entry<String, List<String>> entry : updateKey.entrySet()) {
             List<String> list = entry.getValue();
             for (String col : list) {
-                if (!keyCols.contains(col)) {
+                if (!containsIgnoreCase(keyCols,col)) {
                     keyCols.add(col);
                 }
             }
@@ -107,10 +108,10 @@ public String getUpdateSql(List<String> column, List<String> fullColumn, String
         String prefixRight = StringUtils.isBlank(rightTable) ? "" : quoteTable(rightTable) + ".";
         List<String> list = new ArrayList<>();
         for (String col : fullColumn) {
-            if (keyCols == null || keyCols.size() == 0 || keyCols.contains(col)) {
+            if (keyCols == null || keyCols.size() == 0 || containsIgnoreCase(keyCols,col)) {
                 continue;
             }
-            if (fullColumn == null || column.contains(col)) {
+            if (fullColumn == null ||containsIgnoreCase(column,col)) {
                 list.add(prefixLeft + col + "=" + prefixRight + col);
             } else {
                 list.add(prefixLeft + col + "=null");
@@ -157,6 +158,15 @@ public String makeValues(List<String> column) {
         return sb.toString();
     }
 
+    public boolean containsIgnoreCase(List<String> l, String s) {
+        Iterator<String> it = l.iterator();
+        while (it.hasNext()) {
+            if (it.next().equalsIgnoreCase(s))
+                return true;
+        }
+        return false;
+    }
+
     public String quoteColumn(String column) {
         return getStartQuote() + column + getEndQuote();
     }
diff --git a/rdb/rdb-sink/src/main/java/com/dtstack/flink/sql/sink/rdb/format/ExtendOutputFormat.java b/rdb/rdb-sink/src/main/java/com/dtstack/flink/sql/sink/rdb/format/ExtendOutputFormat.java
@@ -24,6 +24,7 @@
 import java.sql.ResultSet;
 import java.sql.SQLException;
 import java.util.ArrayList;
+import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
 
@@ -45,7 +46,7 @@ public boolean isReplaceInsertQuery() throws SQLException {
         if (!getRealIndexes().isEmpty()) {
             for (List<String> value : getRealIndexes().values()) {
                 for (String fieldName : getDbSink().getFieldNames()) {
-                    if (value.contains(fieldName)) {
+                    if (containsIgnoreCase(value, fieldName)) {
                         return true;
                     }
                 }
@@ -99,5 +100,12 @@ public void fillFullColumns() throws SQLException {
         }
     }
 
-
+    public boolean containsIgnoreCase(List<String> l, String s) {
+        Iterator<String> it = l.iterator();
+        while (it.hasNext()) {
+            if (it.next().equalsIgnoreCase(s))
+                return true;
+        }
+        return false;
+    }
 }
diff --git a/sqlserver/sqlserver-sink/src/main/java/com/dtstack/flink/sql/sink/sqlserver/SqlserverSink.java b/sqlserver/sqlserver-sink/src/main/java/com/dtstack/flink/sql/sink/sqlserver/SqlserverSink.java
@@ -92,7 +92,7 @@ protected List<String> keyColList(Map<String, List<String>> updateKey) {
         for (Map.Entry<String, List<String>> entry : updateKey.entrySet()) {
             List<String> list = entry.getValue();
             for (String col : list) {
-                if (!keyCols.contains(col)) {
+                if (!containsIgnoreCase(keyCols,col)) {
                     keyCols.add(col);
                 }
             }
@@ -108,7 +108,7 @@ public String getUpdateSql(List<String> column, List<String> fullColumn, String
             if (keyCols == null || keyCols.size() == 0) {
                 continue;
             }
-            if (fullColumn == null || column.contains(col)) {
+            if (fullColumn == null || containsIgnoreCase(column,col)) {
                 list.add(prefixLeft + col + "=" + prefixRight + col);
             } else {
                 list.add(prefixLeft + col + "=null");
@@ -154,6 +154,14 @@ public String makeValues(List<String> column) {
         return sb.toString();
     }
 
+    public boolean containsIgnoreCase(List<String> l, String s) {
+        Iterator<String> it = l.iterator();
+        while (it.hasNext()) {
+            if (it.next().equalsIgnoreCase(s))
+                return true;
+        }
+        return false;
+    }
     public String quoteColumn(String column) {
         return getStartQuote() + column + getEndQuote();
     }

Original file line number	Diff line number	Diff line change
`@@ -24,6 +24,7 @@`
`24`	`24`	`import java.sql.ResultSet;`
`25`	`25`	`import java.sql.SQLException;`
`26`	`26`	`import java.util.ArrayList;`
	`27`	`+import java.util.Iterator;`
`27`	`28`	`import java.util.List;`
`28`	`29`	`import java.util.Map;`
`29`	`30`
`@@ -45,7 +46,7 @@ public boolean isReplaceInsertQuery() throws SQLException {`
`45`	`46`	`if (!getRealIndexes().isEmpty()) {`
`46`	`47`	`for (List<String> value : getRealIndexes().values()) {`
`47`	`48`	`for (String fieldName : getDbSink().getFieldNames()) {`
`48`		`- if (value.contains(fieldName)) {`
	`49`	`+ if (containsIgnoreCase(value, fieldName)) {`
`49`	`50`	`return true;`
`50`	`51`	`}`
`51`	`52`	`}`
`@@ -99,5 +100,12 @@ public void fillFullColumns() throws SQLException {`
`99`	`100`	`}`
`100`	`101`	`}`
`101`	`102`
`102`		`-`
	`103`	`+ public boolean containsIgnoreCase(List<String> l, String s) {`
	`104`	`+ Iterator<String> it = l.iterator();`
	`105`	`+ while (it.hasNext()) {`
	`106`	`+ if (it.next().equalsIgnoreCase(s))`
	`107`	`+ return true;`
	`108`	`+ }`
	`109`	`+ return false;`
	`110`	`+ }`
`103`	`111`	`}`