use rpad deal char

Author: todd5167

oracle/oracle-side/oracle-async-side/src/main/java/com/dtstack/flink/sql/side/oracle/OracleAsyncSideInfo.java

@@ -38,11 +38,9 @@
 
 public class OracleAsyncSideInfo extends RdbAsyncSideInfo {
 
-    private final static String SQL_DEFAULT_PLACEHOLDER = " ? ";
-    private final static String DEAL_CHAR_KEY = "char";
-
-    private static String rpadFormat = "rpad(?, %d, ' ')";
-
+    private final String SQL_DEFAULT_PLACEHOLDER = " ? ";
+    private final String DEAL_CHAR_KEY = "char";
+    private String RPAD_FORMAT = "rpad(?, %d, ' ')";
 
 
     public OracleAsyncSideInfo(RowTypeInfo rowTypeInfo, JoinInfo joinInfo, List<FieldInfo> outFieldInfoList, SideTableInfo sideTableInfo) {
@@ -66,9 +64,9 @@ public String wrapperPlaceholder(String fieldName) {
 
         if (StringUtils.contains(type.toLowerCase(), DEAL_CHAR_KEY)) {
             TableInfo.FieldExtraInfo fieldExtraInfo = sideTableInfo.getFieldExtraInfoList().get(pos);
-            int charLength = fieldExtraInfo.getLength();
-            if (fieldExtraInfo.getLength() > 0) {
-                return String.format(rpadFormat, charLength);
+            int charLength = fieldExtraInfo == null ? 0 : fieldExtraInfo.getLength();
+            if (charLength > 0) {
+                return String.format(RPAD_FORMAT, charLength);
             }
         }
         return SQL_DEFAULT_PLACEHOLDER;

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

@@ -21,14 +21,17 @@
 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.table.TableInfo;
 import com.dtstack.flink.sql.util.DtStringUtil;
 import org.apache.commons.lang3.StringUtils;
 import com.google.common.collect.Lists;
 
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
+import java.util.stream.Collectors;
 
 /**
  * Reason:
@@ -40,6 +43,10 @@
 public class OracleSink extends RdbSink implements IStreamSinkGener<RdbSink> {
     private static final String ORACLE_DRIVER = "oracle.jdbc.driver.OracleDriver";
 
+    private final String SQL_DEFAULT_PLACEHOLDER = " ? ";
+    private final String DEAL_CHAR_KEY = "char";
+    private String RPAD_FORMAT = " rpad(?, %d, ' ') ";
+
     @Override
     public String getDriverName() {
         return ORACLE_DRIVER;
@@ -193,16 +200,35 @@ public String updateKeySql(Map<String, List<String>> updateKey) {
      */
     public String makeValues(List<String> column) {
         StringBuilder sb = new StringBuilder("SELECT ");
-        for (int i = 0; i < column.size(); ++i) {
-            if (i != 0) {
-                sb.append(",");
+        String collect = column.stream()
+                .map(col -> wrapperPlaceholder(col) + DtStringUtil.addQuoteForStr(col))
+                .collect(Collectors.joining(", "));
+
+        sb.append(collect).append(" FROM DUAL");
+        return sb.toString();
+    }
+
+    /**
+     *  char type is wrapped with rpad
+     * @param fieldName
+     * @return
+     */
+    public String wrapperPlaceholder(String fieldName) {
+        int pos = rdbTableInfo.getFieldList().indexOf(fieldName);
+        String type = rdbTableInfo.getFieldTypeList().get(pos);
+
+        if (StringUtils.contains(type.toLowerCase(), DEAL_CHAR_KEY)) {
+            TableInfo.FieldExtraInfo fieldExtraInfo = rdbTableInfo.getFieldExtraInfoList().get(pos);
+            int charLength = fieldExtraInfo == null ? 0 : fieldExtraInfo.getLength();
+            if (charLength > 0) {
+                return String.format(RPAD_FORMAT, charLength);
             }
-            sb.append("? " + DtStringUtil.addQuoteForStr(column.get(i)));
         }
-        sb.append(" FROM DUAL");
-        return sb.toString();
+        return SQL_DEFAULT_PLACEHOLDER;
     }
 
+
+
     public boolean containsIgnoreCase(List<String> l, String s) {
         Iterator<String> it = l.iterator();
         while (it.hasNext()) {

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

@@ -83,6 +83,8 @@ public abstract class RdbSink implements RetractStreamTableSink<Row>, Serializab
 
     private String schema;
 
+    protected RdbTableInfo rdbTableInfo;
+
     public RichSinkFunction createJdbcSinkFunc() {
         if (driverName == null || dbURL == null || userName == null
                 || password == null || sqlTypes == null || tableName == null) {
@@ -111,8 +113,7 @@ public RichSinkFunction createJdbcSinkFunc() {
 
     @Override
     public RdbSink genStreamSink(TargetTableInfo targetTableInfo) {
-        RdbTableInfo rdbTableInfo = (RdbTableInfo) targetTableInfo;
-
+        this.rdbTableInfo = (RdbTableInfo) targetTableInfo;
         String tmpDbURL = rdbTableInfo.getUrl();
         String tmpUserName = rdbTableInfo.getUserName();
         String tmpPassword = rdbTableInfo.getPassword();
@@ -263,7 +264,6 @@ public void setDbType(String dbType) {
     /**
      * sqlserver and oracle maybe implement
      *
-     * @param tableName
      * @param fieldNames
      * @param realIndexes
      * @return