修改table primarykey 类型变更

Author: zoudaokoulife

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

@@ -24,6 +24,7 @@
 import com.dtstack.flink.sql.util.DtStringUtil;
 import com.google.common.collect.Lists;
 import com.google.common.collect.Maps;
+import org.apache.commons.lang3.StringUtils;
 
 import java.util.List;
 import java.util.Map;
@@ -114,7 +115,7 @@ public void parseFieldsInfo(String fieldsInfo, TableInfo tableInfo){
 
     public void dealPrimaryKey(Matcher matcher, TableInfo tableInfo){
         String primaryFields = matcher.group(1).trim();
-        String[] splitArry = primaryFields.split(",");
+        String[] splitArry = StringUtils.split(primaryFields, ",");
         tableInfo.setPrimaryKeys(splitArry);
     }
 

hbase/hbase-side/hbase-all-side/src/main/java/com/dtstack/flink/sql/side/hbase/HbaseAllSideInfo.java

@@ -42,11 +42,11 @@ public HbaseAllSideInfo(RowTypeInfo rowTypeInfo, JoinInfo joinInfo, List<FieldIn
     @Override
     public void buildEqualInfo(JoinInfo joinInfo, SideTableInfo sideTableInfo) {
         rowKeyBuilder = new RowKeyBuilder();
-        if(sideTableInfo.getPrimaryKeys().size() < 1){
+        if(sideTableInfo.getPrimaryKeys() == null || sideTableInfo.getPrimaryKeys().length < 1){
             throw new RuntimeException("Primary key dimension table must be filled");
         }
 
-        rowKeyBuilder.init(sideTableInfo.getPrimaryKeys().get(0));
+        rowKeyBuilder.init(sideTableInfo.getPrimaryKeys()[0]);
 
         String sideTableName = joinInfo.getSideTableName();
         SqlNode conditionNode = joinInfo.getCondition();

hbase/hbase-side/hbase-async-side/src/main/java/com/dtstack/flink/sql/side/hbase/HbaseAsyncSideInfo.java

@@ -36,12 +36,12 @@ public HbaseAsyncSideInfo(RowTypeInfo rowTypeInfo, JoinInfo joinInfo, List<Field
     @Override
     public void buildEqualInfo(JoinInfo joinInfo, SideTableInfo sideTableInfo) {
         rowKeyBuilder = new RowKeyBuilder();
-        if(sideTableInfo.getPrimaryKeys().size() < 1){
+        if(sideTableInfo.getPrimaryKeys() == null || sideTableInfo.getPrimaryKeys().length < 1){
             throw new RuntimeException("Primary key dimension table must be filled");
         }
 
         HbaseSideTableInfo hbaseSideTableInfo = (HbaseSideTableInfo) sideTableInfo;
-        rowKeyBuilder.init(sideTableInfo.getPrimaryKeys().get(0));
+        rowKeyBuilder.init(sideTableInfo.getPrimaryKeys()[0]);
 
         colRefType = Maps.newHashMap();
         for(int i=0; i<hbaseSideTableInfo.getColumnRealNames().length; i++){

redis5/redis5-sink/src/main/java/com/dtstack/flink/sql/sink/redis/RedisOutputFormat.java

@@ -56,7 +56,7 @@ public class RedisOutputFormat extends MetricOutputFormat {
 
     protected TypeInformation<?>[] fieldTypes;
 
-    protected List<String> primaryKeys;
+    protected String[] primaryKeys;
 
     protected int timeout;
 
@@ -243,7 +243,7 @@ public RedisOutputFormatBuilder setFieldTypes(TypeInformation<?>[] fieldTypes){
             return this;
         }
 
-        public RedisOutputFormatBuilder setPrimaryKeys(List<String > primaryKeys){
+        public RedisOutputFormatBuilder setPrimaryKeys(String[] primaryKeys){
             redisOutputFormat.primaryKeys = primaryKeys;
             return this;
         }

redis5/redis5-sink/src/main/java/com/dtstack/flink/sql/sink/redis/RedisSink.java

@@ -48,7 +48,7 @@ public class RedisSink implements RetractStreamTableSink<Row>, IStreamSinkGener<
 
     protected String password;
 
-    protected List<String> primaryKeys;
+    protected String[] primaryKeys;
 
     protected int timeout;