Merge branch 'hotfix_1.8_zy_24996' into hotfix_1.8_3.10.x_25114

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];

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

@@ -101,6 +101,8 @@ private void parseSql(SqlNode sqlNode, Map<String, SideTableInfo> sideTableMap,
                 parseSql(unionLeft, sideTableMap, tabMapping);
                 parseSql(unionRight, sideTableMap, tabMapping);
                 break;
+            default:
+                break;
         }
     }
 
@@ -128,10 +130,10 @@ private void extractPredicateInfo(SqlNode whereNode, List<PredicateInfo> predica
 
             // 跳过函数
             if ((((SqlBasicCall) whereNode).getOperands()[0] instanceof SqlIdentifier)
-                    && (((SqlBasicCall) whereNode).getOperands()[1].getKind() != SqlKind.OTHER_FUNCTION)) {
+                    && (((SqlBasicCall) whereNode).getOperands()[1].getKind() == SqlKind.LITERAL)) {
                 fillPredicateInfoToList((SqlBasicCall) whereNode, predicatesInfoList, operatorName, operatorKind, 0, 1);
             } else if ((((SqlBasicCall) whereNode).getOperands()[1] instanceof SqlIdentifier)
-                    && (((SqlBasicCall) whereNode).getOperands()[0].getKind() != SqlKind.OTHER_FUNCTION)) {
+                    && (((SqlBasicCall) whereNode).getOperands()[0].getKind() == LITERAL)) {
                 fillPredicateInfoToList((SqlBasicCall) whereNode, predicatesInfoList, operatorName, operatorKind, 1, 0);
             }
         }

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/RdbAsyncReqRow.java

@@ -88,7 +88,7 @@ protected void init(SideInfo sideInfo) {
 
     @Override
     public void asyncInvoke(CRow input, ResultFuture<CRow> resultFuture) throws Exception {
-        CRow copyCrow = new CRow(input.row(), input.change());
+        CRow copyCrow = new CRow(Row.copy(input.row()), input.change());
         JsonArray inputParams = new JsonArray();
         for (Integer conValIndex : sideInfo.getEqualValIndex()) {
             Object equalObj = copyCrow.row().getField(conValIndex);

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()