Merge branch '1.8_dev_oracle_tablename' into 'v1.8.0_dev'

oracle表名大小写转换



See merge request !77

Author: zoudaokoulife

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

@@ -248,7 +248,32 @@ public static Object parse(String str,Class clazz){
         return object;
     }
 
+
     public static String firstUpperCase(String str) {
         return str.substring(0, 1).toUpperCase() + str.substring(1);
     }
+
+    public static String addQuoteForTableName(String table) {
+        String[] parts = table.split("\\.");
+        StringBuilder sb = new StringBuilder();
+        for (int i = 0; i < parts.length; ++i) {
+            if (i != 0) {
+                sb.append(".");
+            }
+            sb.append(DtStringUtil.addQuoteForColumn(parts[i]));
+        }
+        return sb.toString();
+    }
+
+    public static String addQuoteForColumn(String column) {
+        return getStartQuote() + column + getEndQuote();
+    }
+
+    public static String getStartQuote() {
+        return "\"";
+    }
+
+    public static String getEndQuote() {
+        return "\"";
+    }
 }

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

@@ -21,6 +21,7 @@
 import com.dtstack.flink.sql.sink.rdb.RdbSink;
 import com.dtstack.flink.sql.sink.rdb.format.ExtendOutputFormat;
 import com.dtstack.flink.sql.sink.rdb.format.RetractJDBCOutputFormat;
+import com.dtstack.flink.sql.util.DtStringUtil;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.flink.shaded.guava18.com.google.common.collect.Lists;
 
@@ -56,11 +57,11 @@ public void buildSql(String tableName, List<String> fields) {
 
     private void buildInsertSql(String tableName, List<String> fields) {
 
-        tableName = quoteTable(tableName);
+        tableName = DtStringUtil.addQuoteForTableName(tableName);
         String sqlTmp = "insert into " + tableName + " (${fields}) values (${placeholder})";
 
         List<String> adaptFields = Lists.newArrayList();
-        fields.forEach(field -> adaptFields.add(quoteColumn(field)));
+        fields.forEach(field -> adaptFields.add(DtStringUtil.addQuoteForColumn(field)));
 
         String fieldsStr = StringUtils.join(adaptFields, ",");
         String placeholder = "";
@@ -83,7 +84,7 @@ private void buildInsertSql(String tableName, List<String> fields) {
      */
     @Override
     public String buildUpdateSql(String tableName, List<String> fieldNames, Map<String, List<String>> realIndexes, List<String> fullField) {
-        tableName = quoteTable(tableName);
+        tableName = DtStringUtil.addQuoteForTableName(tableName);
         StringBuilder sb = new StringBuilder();
 
         sb.append("MERGE INTO " + tableName + " T1 USING "
@@ -111,10 +112,10 @@ public String quoteColumns(List<String> column) {
     }
 
     public String quoteColumns(List<String> column, String table) {
-        String prefix = StringUtils.isBlank(table) ? "" : quoteTable(table) + ".";
+        String prefix = StringUtils.isBlank(table) ? "" : DtStringUtil.addQuoteForTableName(table) + ".";
         List<String> list = new ArrayList<>();
         for (String col : column) {
-            list.add(prefix + quoteColumn(col));
+            list.add(prefix + DtStringUtil.addQuoteForColumn(col));
         }
         return StringUtils.join(list, ",");
     }
@@ -147,34 +148,23 @@ protected List<String> keyColList(Map<String, List<String>> realIndexes) {
      * @return
      */
     public String getUpdateSql(List<String> updateColumn, List<String> fullColumn, String leftTable, String rightTable, List<String> indexCols) {
-        String prefixLeft = StringUtils.isBlank(leftTable) ? "" : quoteTable(leftTable) + ".";
-        String prefixRight = StringUtils.isBlank(rightTable) ? "" : quoteTable(rightTable) + ".";
+        String prefixLeft = StringUtils.isBlank(leftTable) ? "" : DtStringUtil.addQuoteForTableName(leftTable) + ".";
+        String prefixRight = StringUtils.isBlank(rightTable) ? "" : DtStringUtil.addQuoteForTableName(rightTable) + ".";
         List<String> list = new ArrayList<>();
         for (String col : fullColumn) {
             // filter index column
             if (indexCols == null || indexCols.size() == 0 || containsIgnoreCase(indexCols,col)) {
                 continue;
             }
             if (containsIgnoreCase(updateColumn,col)) {
-                list.add(prefixLeft + col + "=" + prefixRight + col);
+                list.add(prefixLeft + DtStringUtil.addQuoteForColumn(col) + "=" + prefixRight + DtStringUtil.addQuoteForColumn(col));
             } else {
-                list.add(prefixLeft + col + "=null");
+                list.add(prefixLeft + DtStringUtil.addQuoteForColumn(col) + "=null");
             }
         }
         return StringUtils.join(list, ",");
     }
 
-    public String quoteTable(String table) {
-        String[] parts = table.split("\\.");
-        StringBuilder sb = new StringBuilder();
-        for (int i = 0; i < parts.length; ++i) {
-            if (i != 0) {
-                sb.append(".");
-            }
-            sb.append(getStartQuote() + parts[i] + getEndQuote());
-        }
-        return sb.toString();
-    }
 
     /**
      *  build connect sql by index column, such as    T1."A"=T2."A"
@@ -186,7 +176,7 @@ public String updateKeySql(Map<String, List<String>> updateKey) {
         for (Map.Entry<String, List<String>> entry : updateKey.entrySet()) {
             List<String> colList = new ArrayList<>();
             for (String col : entry.getValue()) {
-                colList.add("T1." + quoteColumn(col) + "=T2." + quoteColumn(col));
+                colList.add("T1." + DtStringUtil.addQuoteForColumn(col) + "=T2." + DtStringUtil.addQuoteForColumn(col));
             }
             exprList.add(StringUtils.join(colList, " AND "));
         }
@@ -205,7 +195,7 @@ public String makeValues(List<String> column) {
             if (i != 0) {
                 sb.append(",");
             }
-            sb.append("? " + quoteColumn(column.get(i)));
+            sb.append("? " + DtStringUtil.addQuoteForColumn(column.get(i)));
         }
         sb.append(" FROM DUAL");
         return sb.toString();
@@ -220,17 +210,6 @@ public boolean containsIgnoreCase(List<String> l, String s) {
         return false;
     }
 
-    public String quoteColumn(String column) {
-        return getStartQuote() + column + getEndQuote();
-    }
-
-    public String getStartQuote() {
-        return "\"";
-    }
-
-    public String getEndQuote() {
-        return "\"";
-    }
 
 
 }

rdb/rdb-sink/src/main/java/com/dtstack/flink/sql/sink/rdb/format/ExtendOutputFormat.java

@@ -18,6 +18,7 @@
 
 package com.dtstack.flink.sql.sink.rdb.format;
 
+import com.dtstack.flink.sql.util.DtStringUtil;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.flink.shaded.guava18.com.google.common.collect.Maps;
 
@@ -62,7 +63,7 @@ public boolean isReplaceInsertQuery() throws SQLException {
      */
     public void fillRealIndexes() throws SQLException {
         Map<String, List<String>> map = Maps.newHashMap();
-        ResultSet rs = getDbConn().getMetaData().getIndexInfo(null, null, getTableName(), true, false);
+        ResultSet rs = getDbConn().getMetaData().getIndexInfo(null, null, DtStringUtil.addQuoteForTableName(getTableName()), true, false);
 
         while (rs.next()) {
             String indexName = rs.getString("INDEX_NAME");
@@ -92,6 +93,7 @@ public void fillRealIndexes() throws SQLException {
      * @throws SQLException
      */
     public void fillFullColumns() throws SQLException {
+        // table name not quote
         ResultSet rs = getDbConn().getMetaData().getColumns(null, null, getTableName(), null);
         while (rs.next()) {
             String columnName = rs.getString("COLUMN_NAME");