Merge remote-tracking branch 'origin/hotfix_1.8_3.9.x_23442' into hotfix_1.8_3.10.x_merged23442

todd5167 · todd5167 · commit 658a6401f4f4 · 2020-03-23T19:49:33.000+08:00
# Conflicts:
#	core/src/main/java/com/dtstack/flink/sql/table/AbsTableParser.java
#	oracle/oracle-sink/src/main/java/com/dtstack/flink/sql/sink/oracle/OracleSink.java
#	rdb/rdb-sink/src/main/java/com/dtstack/flink/sql/sink/rdb/RdbSink.java
#	rdb/rdb-sink/src/main/java/com/dtstack/flink/sql/sink/rdb/table/RdbTableInfo.java
diff --git a/core/src/main/java/com/dtstack/flink/sql/table/AbsTableParser.java b/core/src/main/java/com/dtstack/flink/sql/table/AbsTableParser.java
@@ -43,10 +43,12 @@ 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 physicalFieldFunPattern = Pattern.compile("\\w+\\((\\w+)\\)$");
+    private static Pattern charTypePattern = Pattern.compile("(?i)CHAR\\((\\d*)\\)$");
 
     private Map<String, Pattern> patternMap = Maps.newHashMap();
 
@@ -107,13 +109,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();
diff --git a/core/src/main/java/com/dtstack/flink/sql/table/TableInfo.java b/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 +
+                    '}';
+        }
     }
 }
diff --git a/oracle/oracle-side/oracle-async-side/src/main/java/com/dtstack/flink/sql/side/oracle/OracleAsyncSideInfo.java b/oracle/oracle-side/oracle-async-side/src/main/java/com/dtstack/flink/sql/side/oracle/OracleAsyncSideInfo.java
@@ -23,13 +23,10 @@
 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 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;
 
 
@@ -49,4 +46,21 @@ 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);
+
+        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(rpadFormat, charLength);
+            }
+        }
+        return sqlDefaultPlaceholder;
+    }
 }
diff --git a/oracle/oracle-sink/src/main/java/com/dtstack/flink/sql/sink/oracle/OracleSink.java b/oracle/oracle-sink/src/main/java/com/dtstack/flink/sql/sink/oracle/OracleSink.java
@@ -31,6 +31,10 @@
  */
 public class OracleSink extends RdbSink implements IStreamSinkGener<RdbSink> {
 
+    private final String SQL_DEFAULT_PLACEHOLDER = " ? ";
+    private final String DEAL_CHAR_KEY = "char";
+    private String RPAD_FORMAT = " rpad(?, %d, ' ') ";
+
     public OracleSink() {
         super(new OracleDialect());
     }
diff --git a/rdb/rdb-side/src/main/java/com/dtstack/flink/sql/side/rdb/async/RdbAsyncSideInfo.java b/rdb/rdb-side/src/main/java/com/dtstack/flink/sql/side/rdb/async/RdbAsyncSideInfo.java
@@ -137,14 +137,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) {
diff --git a/rdb/rdb-side/src/main/java/com/dtstack/flink/sql/side/rdb/table/RdbSideTableInfo.java b/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;
     }
 
diff --git a/rdb/rdb-sink/src/main/java/com/dtstack/flink/sql/sink/rdb/RdbSink.java b/rdb/rdb-sink/src/main/java/com/dtstack/flink/sql/sink/rdb/RdbSink.java
@@ -21,7 +21,6 @@
 import com.dtstack.flink.sql.sink.rdb.format.JDBCUpsertOutputFormat;
 import com.dtstack.flink.sql.sink.rdb.table.RdbTableInfo;
 import com.dtstack.flink.sql.table.TargetTableInfo;
-import org.apache.commons.lang3.StringUtils;
 import org.apache.flink.api.common.typeinfo.TypeInformation;
 import org.apache.flink.api.java.tuple.Tuple2;
 import org.apache.flink.api.java.typeutils.RowTypeInfo;
@@ -34,10 +33,6 @@
 import com.dtstack.flink.sql.sink.rdb.dialect.JDBCDialect;
 
 import java.io.Serializable;
-import java.math.BigDecimal;
-import java.sql.Date;
-import java.sql.Timestamp;
-import java.sql.Types;
 import java.util.Arrays;
 import java.util.List;
 
diff --git a/rdb/rdb-sink/src/main/java/com/dtstack/flink/sql/sink/rdb/table/RdbTableInfo.java b/rdb/rdb-sink/src/main/java/com/dtstack/flink/sql/sink/rdb/table/RdbTableInfo.java
@@ -183,6 +183,8 @@ public boolean check() {
         }
 
 
+        Preconditions.checkArgument(getFieldList().size() == getFieldExtraInfoList().size(),
+                "fields and fieldExtraInfoList attributes must be the same length");
         return true;
     }
 

Original file line number	Diff line number	Diff line change
`@@ -47,6 +47,8 @@ public boolean check() {`
`47`	`47`	`Preconditions.checkNotNull(tableName, "rdb of tableName is required");`
`48`	`48`	`Preconditions.checkNotNull(userName, "rdb of userName is required");`
`49`	`49`	`Preconditions.checkNotNull(password, "rdb of password is required");`
	`50`	`+ Preconditions.checkArgument(getFieldList().size() == getFieldExtraInfoList().size(),`
	`51`	`+ "fields and fieldExtraInfoList attributes must be the same length");`
`50`	`52`	`return true;`
`51`	`53`	`}`
`52`	`54`
Original file line number	Diff line number	Diff line change
`@@ -183,6 +183,8 @@ public boolean check() {`
`183`	`183`	`}`
`184`	`184`
`185`	`185`
	`186`	`+ Preconditions.checkArgument(getFieldList().size() == getFieldExtraInfoList().size(),`
	`187`	`+ "fields and fieldExtraInfoList attributes must be the same length");`
`186`	`188`	`return true;`
`187`	`189`	`}`
`188`	`190`