deal char length

Author: todd5167

core/src/main/java/com/dtstack/flink/sql/table/AbsTableParser.java

@@ -42,9 +42,11 @@ public abstract class AbsTableParser {
 
     private static final String PRIMARY_KEY = "primaryKey";
     private static final String NEST_JSON_FIELD_KEY = "nestFieldKey";
+    private static final String CHAR_TYPE_NO_LENGTH = "CHAR";
 
     private static Pattern primaryKeyPattern = Pattern.compile("(?i)PRIMARY\\s+KEY\\s*\\((.*)\\)");
     private static Pattern nestJsonFieldKeyPattern = Pattern.compile("(?i)((@*\\S+\\.)*\\S+)\\s+(\\w+)\\s+AS\\s+(\\w+)(\\s+NOT\\s+NULL)?$");
+    private static Pattern charTypePattern = Pattern.compile("(?i)CHAR\\((\\d*)\\)$");
 
     private Map<String, Pattern> patternMap = Maps.newHashMap();
 
@@ -105,13 +107,25 @@ public void parseFieldsInfo(String fieldsInfo, TableInfo tableInfo){
             System.arraycopy(filedInfoArr, 0, filedNameArr, 0, filedInfoArr.length - 1);
             String fieldName = String.join(" ", filedNameArr);
             String fieldType = filedInfoArr[filedInfoArr.length - 1 ].trim();
-            Class fieldClass = dbTypeConvertToJavaType(fieldType);
+
+
+            Class fieldClass = null;
+            TableInfo.FieldExtraInfo fieldExtraInfo = null;
+
+            Matcher matcher = charTypePattern.matcher(fieldType);
+            if (matcher.find()) {
+                fieldClass = dbTypeConvertToJavaType(CHAR_TYPE_NO_LENGTH);
+                fieldExtraInfo = new TableInfo.FieldExtraInfo();
+                fieldExtraInfo.setLength(Integer.valueOf(matcher.group(1)));
+            } else {
+                fieldClass = dbTypeConvertToJavaType(fieldType);
+            }
 
             tableInfo.addPhysicalMappings(filedInfoArr[0],filedInfoArr[0]);
             tableInfo.addField(fieldName);
             tableInfo.addFieldClass(fieldClass);
             tableInfo.addFieldType(fieldType);
-            tableInfo.addFieldExtraInfo(null);
+            tableInfo.addFieldExtraInfo(fieldExtraInfo);
         }
 
         tableInfo.finish();

core/src/main/java/com/dtstack/flink/sql/table/TableInfo.java

@@ -194,6 +194,18 @@ public static class FieldExtraInfo implements Serializable {
          * default false：allow field is null
          */
         boolean notNull = false;
+        /**
+         *  field length,eg.char(4)
+         */
+        int length;
+
+        public int getLength() {
+            return length;
+        }
+
+        public void setLength(int length) {
+            this.length = length;
+        }
 
         public boolean getNotNull() {
             return notNull;
@@ -202,5 +214,13 @@ public boolean getNotNull() {
         public void setNotNull(boolean notNull) {
             this.notNull = notNull;
         }
+
+        @Override
+        public String toString() {
+            return "FieldExtraInfo{" +
+                    "notNull=" + notNull +
+                    ", length=" + length +
+                    '}';
+        }
     }
 }

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

@@ -23,9 +23,12 @@
 import com.dtstack.flink.sql.side.SideTableInfo;
 import com.dtstack.flink.sql.side.rdb.async.RdbAsyncSideInfo;
 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;
 
@@ -35,6 +38,13 @@
 
 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, ' ')";
+
+
+
     public OracleAsyncSideInfo(RowTypeInfo rowTypeInfo, JoinInfo joinInfo, List<FieldInfo> outFieldInfoList, SideTableInfo sideTableInfo) {
         super(rowTypeInfo, joinInfo, outFieldInfoList, sideTableInfo);
     }
@@ -49,4 +59,18 @@ public  String quoteIdentifier(String identifier) {
         return "\"" + identifier + "\"";
     }
 
+    @Override
+    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)) {
+            TableInfo.FieldExtraInfo fieldExtraInfo = sideTableInfo.getFieldExtraInfoList().get(pos);
+            int charLength = fieldExtraInfo.getLength();
+            if (fieldExtraInfo.getLength() > 0) {
+                return String.format(rpadFormat, charLength);
+            }
+        }
+        return SQL_DEFAULT_PLACEHOLDER;
+    }
 }

rdb/rdb-side/src/main/java/com/dtstack/flink/sql/side/rdb/async/RdbAsyncSideInfo.java

@@ -136,14 +136,26 @@ public String getAdditionalWhereClause() {
 
     public String getSelectFromStatement(String tableName, List<String> selectFields, List<String> conditionFields, List<String> sqlJoinCompareOperate,
                                          List<PredicateInfo> predicateInfoes) {
-        String fromClause = selectFields.stream().map(this::quoteIdentifier).collect(Collectors.joining(", "));
-        String whereClause = conditionFields.stream().map(f -> quoteIdentifier(f) + sqlJoinCompareOperate.get(conditionFields.indexOf(f)) + " ? ")
+        String fromClause = selectFields.stream()
+                .map(this::quoteIdentifier)
+                .collect(Collectors.joining(", "));
+
+        String whereClause = conditionFields.stream()
+                .map(f -> quoteIdentifier(f) + sqlJoinCompareOperate.get(conditionFields.indexOf(f)) + wrapperPlaceholder(f))
+                .collect(Collectors.joining(" AND "));
+
+        String predicateClause = predicateInfoes.stream()
+                .map(this::buildFilterCondition)
                 .collect(Collectors.joining(" AND "));
-        String predicateClause = predicateInfoes.stream().map(this::buildFilterCondition).collect(Collectors.joining(" AND "));
 
-        String sql = "SELECT " + fromClause + " FROM " + tableName + (conditionFields.size() > 0 ? " WHERE " + whereClause : "")
+        String dimQuerySql = "SELECT " + fromClause + " FROM " + tableName + (conditionFields.size() > 0 ? " WHERE " + whereClause : "")
                 + (predicateInfoes.size() > 0 ? " AND " + predicateClause : "") + getAdditionalWhereClause();
-        return sql;
+
+        return dimQuerySql;
+    }
+
+    public String wrapperPlaceholder(String fieldName) {
+        return " ? ";
     }
 
     public String buildFilterCondition(PredicateInfo info) {