乱序，别名和hase

Author: dapeng

core/src/main/java/com/dtstack/flink/sql/side/SideInfo.java

@@ -132,13 +132,12 @@ public void dealOneEqualCon(SqlNode sqlNode, String sideTableName){
 
         String leftTableName = left.getComponent(0).getSimple();
         String leftField = left.getComponent(1).getSimple();
-        Map<String, String> physicalFields = sideTableInfo.getPhysicalFields();
 
         String rightTableName = right.getComponent(0).getSimple();
         String rightField = right.getComponent(1).getSimple();
 
         if(leftTableName.equalsIgnoreCase(sideTableName)){
-            equalFieldList.add(physicalFields.get(leftField));
+            equalFieldList.add(leftField);
             int equalFieldIndex = -1;
             for(int i=0; i<rowTypeInfo.getFieldNames().length; i++){
                 String fieldName = rowTypeInfo.getFieldNames()[i];
@@ -154,7 +153,7 @@ public void dealOneEqualCon(SqlNode sqlNode, String sideTableName){
 
         }else if(rightTableName.equalsIgnoreCase(sideTableName)){
 
-            equalFieldList.add(physicalFields.get(rightField));
+            equalFieldList.add(rightField);
             int equalFieldIndex = -1;
             for(int i=0; i<rowTypeInfo.getFieldNames().length; i++){
                 String fieldName = rowTypeInfo.getFieldNames()[i];

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

@@ -124,7 +124,7 @@ public void open(Configuration parameters) throws Exception {
 
     @Override
     public void asyncInvoke(CRow input, ResultFuture<CRow> resultFuture) throws Exception {
-        CRow inputCopy = new CRow(input.row(), input.change());
+        CRow inputCopy = new CRow(Row.copy(input.row()), input.change());
         Map<String, Object> refData = Maps.newHashMap();
         for (int i = 0; i < sideInfo.getEqualValIndex().size(); i++) {
             Integer conValIndex = sideInfo.getEqualValIndex().get(i);

hbase/hbase-side/hbase-side-core/src/main/java/com/dtstack/flink/sql/side/hbase/table/HbaseSideParser.java

@@ -100,6 +100,7 @@ private void dealField(Matcher matcher, TableInfo tableInfo){
         sideTableInfo.addFieldClass(fieldClass);
         sideTableInfo.addFieldType(fieldType);
         sideTableInfo.putAliasNameRef(aliasStr, fieldName);
+        sideTableInfo.addPhysicalMappings(aliasStr, fieldName);
     }
 
 

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

@@ -87,7 +87,6 @@ public void dealOneEqualCon(SqlNode sqlNode, String sideTableName) {
 
         SqlIdentifier left = (SqlIdentifier) ((SqlBasicCall) sqlNode).getOperands()[0];
         SqlIdentifier right = (SqlIdentifier) ((SqlBasicCall) sqlNode).getOperands()[1];
-        Map<String, String> physicalFields = sideTableInfo.getPhysicalFields();
 
         String leftTableName = left.getComponent(0).getSimple();
         String leftField = left.getComponent(1).getSimple();
@@ -96,7 +95,7 @@ public void dealOneEqualCon(SqlNode sqlNode, String sideTableName) {
         String rightField = right.getComponent(1).getSimple();
 
         if (leftTableName.equalsIgnoreCase(sideTableName)) {
-            equalFieldList.add(physicalFields.get(leftField));
+            equalFieldList.add(leftField);
             int equalFieldIndex = -1;
             for (int i = 0; i < rowTypeInfo.getFieldNames().length; i++) {
                 String fieldName = rowTypeInfo.getFieldNames()[i];
@@ -112,7 +111,7 @@ public void dealOneEqualCon(SqlNode sqlNode, String sideTableName) {
 
         } else if (rightTableName.equalsIgnoreCase(sideTableName)) {
 
-            equalFieldList.add(physicalFields.get(rightField));
+            equalFieldList.add(rightField);
             int equalFieldIndex = -1;
             for (int i = 0; i < rowTypeInfo.getFieldNames().length; i++) {
                 String fieldName = rowTypeInfo.getFieldNames()[i];
@@ -144,7 +143,7 @@ public String getSelectFromStatement(String tableName, List<String> selectFields
                 .collect(Collectors.joining(", "));
 
         String whereClause = conditionFields.stream()
-                .map(f -> quoteIdentifier(f) + sqlJoinCompareOperate.get(conditionFields.indexOf(f)) + wrapperPlaceholder(f))
+                .map(f -> quoteIdentifier(sideTableInfo.getPhysicalFields().getOrDefault(f, f)) + sqlJoinCompareOperate.get(conditionFields.indexOf(f)) + wrapperPlaceholder(f))
                 .collect(Collectors.joining(" AND "));
 
         String predicateClause = predicateInfoes.stream()