oracle schema

Author: todd5167

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

@@ -253,19 +253,8 @@ 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) {
+    public static String addQuoteForStr(String column) {
         return getStartQuote() + column + getEndQuote();
     }
 

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

@@ -57,11 +57,11 @@ public void buildSql(String tableName, List<String> fields) {
 
     private void buildInsertSql(String tableName, List<String> fields) {
 
-        tableName = DtStringUtil.addQuoteForTableName(tableName);
+        tableName = DtStringUtil.addQuoteForStr(tableName);
         String sqlTmp = "insert into " + tableName + " (${fields}) values (${placeholder})";
 
         List<String> adaptFields = Lists.newArrayList();
-        fields.forEach(field -> adaptFields.add(DtStringUtil.addQuoteForColumn(field)));
+        fields.forEach(field -> adaptFields.add(DtStringUtil.addQuoteForStr(field)));
 
         String fieldsStr = StringUtils.join(adaptFields, ",");
         String placeholder = "";
@@ -84,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 = DtStringUtil.addQuoteForTableName(tableName);
+        tableName = DtStringUtil.addQuoteForStr(tableName);
         StringBuilder sb = new StringBuilder();
 
         sb.append("MERGE INTO " + tableName + " T1 USING "
@@ -112,10 +112,10 @@ public String quoteColumns(List<String> column) {
     }
 
     public String quoteColumns(List<String> column, String table) {
-        String prefix = StringUtils.isBlank(table) ? "" : DtStringUtil.addQuoteForTableName(table) + ".";
+        String prefix = StringUtils.isBlank(table) ? "" : DtStringUtil.addQuoteForStr(table) + ".";
         List<String> list = new ArrayList<>();
         for (String col : column) {
-            list.add(prefix + DtStringUtil.addQuoteForColumn(col));
+            list.add(prefix + DtStringUtil.addQuoteForStr(col));
         }
         return StringUtils.join(list, ",");
     }
@@ -148,18 +148,18 @@ 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) ? "" : DtStringUtil.addQuoteForTableName(leftTable) + ".";
-        String prefixRight = StringUtils.isBlank(rightTable) ? "" : DtStringUtil.addQuoteForTableName(rightTable) + ".";
+        String prefixLeft = StringUtils.isBlank(leftTable) ? "" : DtStringUtil.addQuoteForStr(leftTable) + ".";
+        String prefixRight = StringUtils.isBlank(rightTable) ? "" : DtStringUtil.addQuoteForStr(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 + DtStringUtil.addQuoteForColumn(col) + "=" + prefixRight + DtStringUtil.addQuoteForColumn(col));
+                list.add(prefixLeft + DtStringUtil.addQuoteForStr(col) + "=" + prefixRight + DtStringUtil.addQuoteForStr(col));
             } else {
-                list.add(prefixLeft + DtStringUtil.addQuoteForColumn(col) + "=null");
+                list.add(prefixLeft + DtStringUtil.addQuoteForStr(col) + "=null");
             }
         }
         return StringUtils.join(list, ",");
@@ -176,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." + DtStringUtil.addQuoteForColumn(col) + "=T2." + DtStringUtil.addQuoteForColumn(col));
+                colList.add("T1." + DtStringUtil.addQuoteForStr(col) + "=T2." + DtStringUtil.addQuoteForStr(col));
             }
             exprList.add(StringUtils.join(colList, " AND "));
         }
@@ -195,7 +195,7 @@ public String makeValues(List<String> column) {
             if (i != 0) {
                 sb.append(",");
             }
-            sb.append("? " + DtStringUtil.addQuoteForColumn(column.get(i)));
+            sb.append("? " + DtStringUtil.addQuoteForStr(column.get(i)));
         }
         sb.append(" FROM DUAL");
         return sb.toString();

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

@@ -63,7 +63,8 @@ public boolean isReplaceInsertQuery() throws SQLException {
      */
     public void fillRealIndexes() throws SQLException {
         Map<String, List<String>> map = Maps.newHashMap();
-        ResultSet rs = getDbConn().getMetaData().getIndexInfo(null, null, DtStringUtil.addQuoteForTableName(getTableName()), true, false);
+
+        ResultSet rs = getDbConn().getMetaData().getIndexInfo(null, getSchema().toUpperCase(), DtStringUtil.addQuoteForStr(getTableName()), true, false);
 
         while (rs.next()) {
             String indexName = rs.getString("INDEX_NAME");
@@ -94,7 +95,7 @@ public void fillRealIndexes() throws SQLException {
      */
     public void fillFullColumns() throws SQLException {
         // table name not quote
-        ResultSet rs = getDbConn().getMetaData().getColumns(null, null, getTableName(), null);
+        ResultSet rs = getDbConn().getMetaData().getColumns(null, getSchema().toUpperCase(), getTableName(), null);
         while (rs.next()) {
             String columnName = rs.getString("COLUMN_NAME");
             if (StringUtils.isNotBlank(columnName)) {
@@ -111,4 +112,8 @@ public boolean containsIgnoreCase(List<String> l, String s) {
         }
         return false;
     }
+
+    public boolean existTabname() throws SQLException {
+        return getDbConn().getMetaData().getTables(null, getSchema().toUpperCase(), getTableName(), null).next();
+    }
 }

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

@@ -106,7 +106,7 @@ public void open(int taskNumber, int numTasks) throws IOException {
             dbConn = establishConnection();
             initMetric();
 
-            if (dbConn.getMetaData().getTables(null, null, tableName, null).next()) {
+            if (existTabname()) {
                 if (isReplaceInsertQuery()) {
                     insertQuery = dbSink.buildUpdateSql(tableName, Arrays.asList(dbSink.getFieldNames()), realIndexes, fullField);
                 }
@@ -400,6 +400,18 @@ public void setUsername(String username) {
         this.username = username;
     }
 
+    /**
+     *  username as default schema
+     * @return
+     */
+    public String getSchema() {
+        return username;
+    }
+
+    public boolean existTabname() throws SQLException {
+        return dbConn.getMetaData().getTables(null, null, tableName, null).next();
+    }
+
     public void setPassword(String password) {
         this.password = password;
     }