field length check

Author: todd5167

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

@@ -25,24 +25,13 @@
 import com.dtstack.flink.sql.side.rdb.table.RdbSideTableInfo;
 import com.dtstack.flink.sql.table.TableInfo;
 import com.dtstack.flink.sql.util.DtStringUtil;
-import com.dtstack.flink.sql.util.ParseUtils;
-import com.mchange.lang.CharUtils;
-import org.apache.calcite.sql.SqlNode;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.flink.api.java.typeutils.RowTypeInfo;
-import com.google.common.collect.Lists;
-
-import java.util.Arrays;
 import java.util.List;
 
 
 public class OracleAsyncSideInfo extends RdbAsyncSideInfo {
 
-    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) {
         super(rowTypeInfo, joinInfo, outFieldInfoList, sideTableInfo);
     }
@@ -62,13 +51,16 @@ public String wrapperPlaceholder(String fieldName) {
         int pos = sideTableInfo.getFieldList().indexOf(fieldName);
         String type = sideTableInfo.getFieldTypeList().get(pos);
 
-        if (StringUtils.contains(type.toLowerCase(), DEAL_CHAR_KEY)) {
+        String sqlDefaultPlaceholder = " ? ";
+        String rpadFormat = "rpad(?, %d, ' ')";
+
+        if (StringUtils.contains(type.toLowerCase(), "char")) {
             TableInfo.FieldExtraInfo fieldExtraInfo = sideTableInfo.getFieldExtraInfoList().get(pos);
             int charLength = fieldExtraInfo == null ? 0 : fieldExtraInfo.getLength();
             if (charLength > 0) {
-                return String.format(RPAD_FORMAT, charLength);
+                return String.format(rpadFormat, charLength);
             }
         }
-        return SQL_DEFAULT_PLACEHOLDER;
+        return sqlDefaultPlaceholder;
     }
 }

rdb/rdb-side/src/main/java/com/dtstack/flink/sql/side/rdb/table/RdbSideTableInfo.java

@@ -47,6 +47,8 @@ public boolean check() {
         Preconditions.checkNotNull(tableName, "rdb of tableName is required");
         Preconditions.checkNotNull(userName, "rdb of userName is required");
         Preconditions.checkNotNull(password, "rdb of password is required");
+        Preconditions.checkArgument(getFieldList().size() == getFieldExtraInfoList().size(),
+                "fields and fieldExtraInfoList attributes must be the same length");
         return true;
     }
 

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

@@ -144,6 +144,8 @@ public boolean check() {
         Preconditions.checkNotNull(tableName, "rdb field of tableName is required");
         Preconditions.checkNotNull(userName, "rdb field of userName is required");
         Preconditions.checkNotNull(password, "rdb field of password is required");
+        Preconditions.checkArgument(getFieldList().size() == getFieldExtraInfoList().size(),
+                "fields and fieldExtraInfoList attributes must be the same length");
         return true;
     }