Merge remote-tracking branch 'origin/v1.5.0_dev' into v1.5.0_dev

zoudaokoulife · zoudaokoulife · commit ea8cee439cee · 2019-01-11T17:26:54.000+08:00
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
@@ -109,7 +109,7 @@ public void parseFieldsInfo(String fieldsInfo, TableInfo tableInfo){
     }
 
     public static void dealPrimaryKey(Matcher matcher, TableInfo tableInfo){
-        String primaryFields = matcher.group(1);
+        String primaryFields = matcher.group(1).trim();
         String[] splitArry = primaryFields.split(",");
         List<String> primaryKes = Lists.newArrayList(splitArry);
         tableInfo.setPrimaryKeys(primaryKes);
diff --git a/hbase/hbase-side/hbase-all-side/src/main/java/com/dtstack/flink/sql/side/hbase/HbaseAllReqRow.java b/hbase/hbase-side/hbase-all-side/src/main/java/com/dtstack/flink/sql/side/hbase/HbaseAllReqRow.java
@@ -170,7 +170,7 @@ private void loadData(Map<String, Map<String, Object>> tmpCache) throws SQLExcep
                     StringBuilder key = new StringBuilder();
                     key.append(family).append(":").append(qualifier);
 
-                    kv.put(aliasNameInversion.get(key.toString().toUpperCase()), value);
+                    kv.put(aliasNameInversion.get(key.toString()), value);
                 }
                 tmpCache.put(new String(r.getRow()), kv);
             }
diff --git a/hbase/hbase-side/hbase-async-side/src/main/java/com/dtstack/flink/sql/side/hbase/rowkeydealer/PreRowKeyModeDealerDealer.java b/hbase/hbase-side/hbase-async-side/src/main/java/com/dtstack/flink/sql/side/hbase/rowkeydealer/PreRowKeyModeDealerDealer.java
@@ -98,9 +98,9 @@ private String dealOneRow(ArrayList<ArrayList<KeyValue>> args, String rowKeyStr,
                     String mapKey = cf + ":" + col;
 
                     //The table format defined using different data type conversion byte
-                    String colType = colRefType.get(mapKey.toUpperCase());
+                    String colType = colRefType.get(mapKey);
                     Object val = HbaseUtils.convertByte(keyValue.value(), colType);
-                    sideMap.put(mapKey.toUpperCase(), val);
+                    sideMap.put(mapKey, val);
                 }
 
                 if (oneRow.size() > 0) {
diff --git a/hbase/hbase-side/hbase-async-side/src/main/java/com/dtstack/flink/sql/side/hbase/rowkeydealer/RowKeyEqualModeDealer.java b/hbase/hbase-side/hbase-async-side/src/main/java/com/dtstack/flink/sql/side/hbase/rowkeydealer/RowKeyEqualModeDealer.java
@@ -73,9 +73,9 @@ public void asyncGetData(String tableName, String rowKeyStr, Row input, ResultFu
                     String col = new String(keyValue.qualifier());
                     String mapKey = cf + ":" + col;
                     //The table format defined using different data type conversion byte
-                    String colType = colRefType.get(mapKey.toUpperCase());
+                    String colType = colRefType.get(mapKey);
                     Object val = HbaseUtils.convertByte(keyValue.value(), colType);
-                    sideMap.put(mapKey.toUpperCase(), val);
+                    sideMap.put(mapKey, val);
                 }
 
                 if(arg.size() > 0){
diff --git a/mongo/mongo-side/mongo-async-side/src/main/java/com/dtstack/flink/sql/side/mongo/MongoAsyncReqRow.java b/mongo/mongo-side/mongo-async-side/src/main/java/com/dtstack/flink/sql/side/mongo/MongoAsyncReqRow.java
@@ -130,6 +130,7 @@ public void asyncInvoke(Row input, ResultFuture<Row> resultFuture) throws Except
             Object equalObj = input.getField(conValIndex);
             if (equalObj == null) {
                 resultFuture.complete(null);
+                return;
             }
             basicDBObject.put(sideInfo.getEqualFieldList().get(i), equalObj);
         }
@@ -142,12 +143,12 @@ public void asyncInvoke(Row input, ResultFuture<Row> resultFuture) throws Except
                     dealMissKey(input, resultFuture);
                     return;
                 } else if (ECacheContentType.MultiLine == val.getType()) {
-
+                    List<Row> rowList = Lists.newArrayList();
                     for (Object jsonArray : (List) val.getContent()) {
                         Row row = fillData(input, jsonArray);
-                        resultFuture.complete(Collections.singleton(row));
+                        rowList.add(row);
                     }
-
+                    resultFuture.complete(rowList);
                 } else {
                     throw new RuntimeException("not support cache obj type " + val.getType());
                 }
diff --git a/rdb/rdb-side/src/main/java/com/dtstack/flink/sql/side/rdb/async/RdbAsyncReqRow.java b/rdb/rdb-side/src/main/java/com/dtstack/flink/sql/side/rdb/async/RdbAsyncReqRow.java
@@ -75,8 +75,8 @@ public void asyncInvoke(Row input, ResultFuture<Row> resultFuture) throws Except
             Object equalObj = input.getField(conValIndex);
             if (equalObj == null) {
                 resultFuture.complete(null);
+                return;
             }
-
             inputParams.add(equalObj);
         }
 
@@ -89,12 +89,12 @@ public void asyncInvoke(Row input, ResultFuture<Row> resultFuture) throws Except
                     dealMissKey(input, resultFuture);
                     return;
                 } else if (ECacheContentType.MultiLine == val.getType()) {
-
+                    List<Row> rowList = Lists.newArrayList();
                     for (Object jsonArray : (List) val.getContent()) {
                         Row row = fillData(input, jsonArray);
-                        resultFuture.complete(Collections.singleton(row));
+                        rowList.add(row);
                     }
-
+                    resultFuture.complete(rowList);
                 } else {
                     throw new RuntimeException("not support cache obj type " + val.getType());
                 }
diff --git a/redis5/redis5-side/redis-async-side/src/main/java/com/dtstack/flink/sql/side/redis/RedisAsyncReqRow.java b/redis5/redis5-side/redis-async-side/src/main/java/com/dtstack/flink/sql/side/redis/RedisAsyncReqRow.java
@@ -141,8 +141,12 @@ public void asyncInvoke(Row input, ResultFuture<Row> resultFuture) throws Except
                     dealMissKey(input, resultFuture);
                     return;
                 }else if(ECacheContentType.MultiLine == val.getType()){
-                    Row row = fillData(input, val.getContent());
-                    resultFuture.complete(Collections.singleton(row));
+                    List<Row> rowList = Lists.newArrayList();
+                    for (Object jsonArray : (List) val.getContent()) {
+                        Row row = fillData(input, val.getContent());
+                        rowList.add(row);
+                    }
+                    resultFuture.complete(rowList);
                 }else{
                     throw new RuntimeException("not support cache obj type " + val.getType());
                 }

Original file line number	Diff line number	Diff line change
`@@ -109,7 +109,7 @@ public void parseFieldsInfo(String fieldsInfo, TableInfo tableInfo){`
`109`	`109`	`}`
`110`	`110`
`111`	`111`	`public static void dealPrimaryKey(Matcher matcher, TableInfo tableInfo){`
`112`		`- String primaryFields = matcher.group(1);`
	`112`	`+ String primaryFields = matcher.group(1).trim();`
`113`	`113`	`String[] splitArry = primaryFields.split(",");`
`114`	`114`	`List<String> primaryKes = Lists.newArrayList(splitArry);`
`115`	`115`	`tableInfo.setPrimaryKeys(primaryKes);`
Original file line number	Diff line number	Diff line change
`@@ -170,7 +170,7 @@ private void loadData(Map<String, Map<String, Object>> tmpCache) throws SQLExcep`
`170`	`170`	`StringBuilder key = new StringBuilder();`
`171`	`171`	`key.append(family).append(":").append(qualifier);`
`172`	`172`
`173`		`- kv.put(aliasNameInversion.get(key.toString().toUpperCase()), value);`
	`173`	`+ kv.put(aliasNameInversion.get(key.toString()), value);`
`174`	`174`	`}`
`175`	`175`	`tmpCache.put(new String(r.getRow()), kv);`
`176`	`176`	`}`
Original file line number	Diff line number	Diff line change
`@@ -130,6 +130,7 @@ public void asyncInvoke(Row input, ResultFuture<Row> resultFuture) throws Except`
`130`	`130`	`Object equalObj = input.getField(conValIndex);`
`131`	`131`	`if (equalObj == null) {`
`132`	`132`	`resultFuture.complete(null);`
	`133`	`+ return;`
`133`	`134`	`}`
`134`	`135`	`basicDBObject.put(sideInfo.getEqualFieldList().get(i), equalObj);`
`135`	`136`	`}`
`@@ -142,12 +143,12 @@ public void asyncInvoke(Row input, ResultFuture<Row> resultFuture) throws Except`
`142`	`143`	`dealMissKey(input, resultFuture);`
`143`	`144`	`return;`
`144`	`145`	`} else if (ECacheContentType.MultiLine == val.getType()) {`
`145`		`-`
	`146`	`+ List<Row> rowList = Lists.newArrayList();`
`146`	`147`	`for (Object jsonArray : (List) val.getContent()) {`
`147`	`148`	`Row row = fillData(input, jsonArray);`
`148`		`- resultFuture.complete(Collections.singleton(row));`
	`149`	`+ rowList.add(row);`
`149`	`150`	`}`
`150`		`-`
	`151`	`+ resultFuture.complete(rowList);`
`151`	`152`	`} else {`
`152`	`153`	`throw new RuntimeException("not support cache obj type " + val.getType());`
`153`	`154`	`}`
Original file line number	Diff line number	Diff line change
`@@ -75,8 +75,8 @@ public void asyncInvoke(Row input, ResultFuture<Row> resultFuture) throws Except`
`75`	`75`	`Object equalObj = input.getField(conValIndex);`
`76`	`76`	`if (equalObj == null) {`
`77`	`77`	`resultFuture.complete(null);`
	`78`	`+ return;`
`78`	`79`	`}`
`79`		`-`
`80`	`80`	`inputParams.add(equalObj);`
`81`	`81`	`}`
`82`	`82`
`@@ -89,12 +89,12 @@ public void asyncInvoke(Row input, ResultFuture<Row> resultFuture) throws Except`
`89`	`89`	`dealMissKey(input, resultFuture);`
`90`	`90`	`return;`
`91`	`91`	`} else if (ECacheContentType.MultiLine == val.getType()) {`
`92`		`-`
	`92`	`+ List<Row> rowList = Lists.newArrayList();`
`93`	`93`	`for (Object jsonArray : (List) val.getContent()) {`
`94`	`94`	`Row row = fillData(input, jsonArray);`
`95`		`- resultFuture.complete(Collections.singleton(row));`
	`95`	`+ rowList.add(row);`
`96`	`96`	`}`
`97`		`-`
	`97`	`+ resultFuture.complete(rowList);`
`98`	`98`	`} else {`
`99`	`99`	`throw new RuntimeException("not support cache obj type " + val.getType());`
`100`	`100`	`}`