oracle sink code opt

Author: todd5167

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

@@ -34,8 +34,8 @@ public class OracleSinkParser extends RdbSinkParser {
 
     @Override
     public TableInfo getTableInfo(String tableName, String fieldsInfo, Map<String, Object> props) {
-        TableInfo sqlserverTableInfo = super.getTableInfo(tableName, fieldsInfo, props);
-        sqlserverTableInfo.setType(CURR_TYPE);
-        return sqlserverTableInfo;
+        TableInfo oracleTableInfo = super.getTableInfo(tableName, fieldsInfo, props);
+        oracleTableInfo.setType(CURR_TYPE);
+        return oracleTableInfo;
     }
 }

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

@@ -66,25 +66,26 @@ public void fillRealIndexes() throws SQLException {
 
         while (rs.next()) {
             String indexName = rs.getString("INDEX_NAME");
-            if (!map.containsKey(indexName)) {
+            if (StringUtils.isNotBlank(indexName) && !map.containsKey(indexName)) {
                 map.put(indexName, new ArrayList<>());
             }
             String column_name = rs.getString("COLUMN_NAME");
             if (StringUtils.isNotBlank(column_name)) {
-                column_name = column_name.toUpperCase();
+                map.get(indexName).add(column_name);
             }
-            map.get(indexName).add(column_name);
         }
 
         for (Map.Entry<String, List<String>> entry : map.entrySet()) {
             String k = entry.getKey();
             List<String> v = entry.getValue();
             if (v != null && v.size() != 0 && v.get(0) != null) {
-                getRealIndexes().put(k, v);
+                realIndexesAdd(k, v);
             }
         }
     }
 
+
+
     /**
      * get db all column name
      *
@@ -95,7 +96,7 @@ public void fillFullColumns() throws SQLException {
         while (rs.next()) {
             String columnName = rs.getString("COLUMN_NAME");
             if (StringUtils.isNotBlank(columnName)) {
-                getFullField().add(columnName.toUpperCase());
+                fullFieldAdd(columnName);
             }
         }
     }

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

@@ -452,7 +452,11 @@ public String getTableName() {
         return tableName;
     }
 
-    public Map<String, List<String>> getRealIndexes() {
+    public void realIndexesAdd(String index, List<String> fieldes) {
+        this.realIndexes.put(index, fieldes);
+    }
+
+    public Map<String,List<String>> getRealIndexes() {
         return realIndexes;
     }
 
@@ -464,4 +468,8 @@ public void setBatchWaitInterval(long batchWaitInterval) {
     public List<String> getFullField() {
         return fullField;
     }
+
+    public void fullFieldAdd(String colName) {
+        this.fullField.add(colName);
+    }
 }