Merge branch '1.10_release_4.0.x' into 1.10_release_4.1.x

Author: gituser

core/src/main/java/com/dtstack/flink/sql/parser/InsertSqlParser.java

@@ -162,23 +162,32 @@ private static void rebuildSelectNode(SqlNodeList selectList, SqlSelect sqlSelec
         SqlNodeList sqlNodes = new SqlNodeList(selectList.getParserPosition());
 
         for (int index = 0; index < selectList.size(); index++) {
-            if (selectList.get(index).getKind().equals(SqlKind.AS)
-                    || (selectList.get(index).getClass().equals(SqlIdentifier.class)
-                    && ((SqlIdentifier) selectList.get(index)).names.size() == 1)) {
-                sqlNodes.add(selectList.get(index));
+            SqlNode sqlNode = selectList.get(index);
+            // 判断sqlNode的类型是否属于 't1.f1 as f2'
+            boolean isAsNode = sqlNode.getKind().equals(SqlKind.AS);
+
+            // 判断sqlNode的结构是否属于'f1' 或者 't.*'
+            boolean isIdentifierOrStar = sqlNode.getClass().equals(SqlIdentifier.class)
+                    // sqlNode like 'f1'
+                    && (((SqlIdentifier) sqlNode).names.size() == 1
+                    // sqlNode like 't.*'
+                    || StringUtils.isBlank(((SqlIdentifier) sqlNode).names.get(1)));
+
+            if (isAsNode || isIdentifierOrStar) {
+                sqlNodes.add(sqlNode);
                 continue;
             }
 
-            if (!selectList.get(index).getClass().equals(SqlIdentifier.class)) {
-                if (selectList.get(index).getKind().equals(SqlKind.LITERAL)) {
+            if (!sqlNode.getClass().equals(SqlIdentifier.class)) {
+                if (sqlNode.getKind().equals(SqlKind.LITERAL)) {
                     throw new IllegalArgumentException(String.format("Constants %s in the SELECT statement must be aliased!",
-                            selectList.get(index).toString()));
+                            sqlNode.toString()));
                 }
                 throw new RuntimeException(String.format("Illegal statement! Please check the statement: %s",
-                        selectList.get(index).toString()));
+                        sqlNode.toString()));
             }
 
-            sqlNodes.add(transformToAsNode(selectList.get(index)));
+            sqlNodes.add(transformToAsNode(sqlNode));
         }
         sqlSelect.setSelectList(sqlNodes);
     }

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

@@ -44,6 +44,33 @@ public HbaseAllSideInfo(RowTypeInfo rowTypeInfo, JoinInfo joinInfo, List<FieldIn
         super(rowTypeInfo, joinInfo, outFieldInfoList, sideTableInfo);
     }
 
+    @Override
+    public void parseSelectFields(JoinInfo joinInfo) {
+        String sideTableName = joinInfo.getSideTableName();
+        String nonSideTableName = joinInfo.getNonSideTable();
+        List<String> fields = Lists.newArrayList();
+        int sideTableFieldIndex = 0;
+
+        for( int i=0; i<outFieldInfoList.size(); i++){
+            FieldInfo fieldInfo = outFieldInfoList.get(i);
+            if(fieldInfo.getTable().equalsIgnoreCase(sideTableName)){
+                String sideFieldName = sideTableInfo.getPhysicalFields().getOrDefault(fieldInfo.getFieldName(), fieldInfo.getFieldName());
+                fields.add(sideFieldName);
+                sideSelectFieldsType.put(sideTableFieldIndex, getTargetFieldType(fieldInfo.getFieldName()));
+                sideFieldIndex.put(i, sideTableFieldIndex);
+                sideFieldNameIndex.put(i, sideFieldName);
+                sideTableFieldIndex++;
+            }else if(fieldInfo.getTable().equalsIgnoreCase(nonSideTableName)){
+                int nonSideIndex = rowTypeInfo.getFieldIndex(fieldInfo.getFieldName());
+                inFieldIndex.put(i, nonSideIndex);
+            }else{
+                throw new RuntimeException("unknown table " + fieldInfo.getTable());
+            }
+        }
+
+        sideSelectFields = String.join(",", fields);
+    }
+
     @Override
     public void buildEqualInfo(JoinInfo joinInfo, AbstractSideTableInfo sideTableInfo) {
         rowKeyBuilder = new RowKeyBuilder();

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

@@ -33,6 +33,33 @@ public HbaseAsyncSideInfo(RowTypeInfo rowTypeInfo, JoinInfo joinInfo, List<Field
         super(rowTypeInfo, joinInfo, outFieldInfoList, sideTableInfo);
     }
 
+    @Override
+    public void parseSelectFields(JoinInfo joinInfo) {
+        String sideTableName = joinInfo.getSideTableName();
+        String nonSideTableName = joinInfo.getNonSideTable();
+        List<String> fields = Lists.newArrayList();
+        int sideTableFieldIndex = 0;
+
+        for( int i=0; i<outFieldInfoList.size(); i++){
+            FieldInfo fieldInfo = outFieldInfoList.get(i);
+            if(fieldInfo.getTable().equalsIgnoreCase(sideTableName)){
+                String sideFieldName = sideTableInfo.getPhysicalFields().getOrDefault(fieldInfo.getFieldName(), fieldInfo.getFieldName());
+                fields.add(sideFieldName);
+                sideSelectFieldsType.put(sideTableFieldIndex, getTargetFieldType(fieldInfo.getFieldName()));
+                sideFieldIndex.put(i, sideTableFieldIndex);
+                sideFieldNameIndex.put(i, sideFieldName);
+                sideTableFieldIndex++;
+            }else if(fieldInfo.getTable().equalsIgnoreCase(nonSideTableName)){
+                int nonSideIndex = rowTypeInfo.getFieldIndex(fieldInfo.getFieldName());
+                inFieldIndex.put(i, nonSideIndex);
+            }else{
+                throw new RuntimeException("unknown table " + fieldInfo.getTable());
+            }
+        }
+
+        sideSelectFields = String.join(",", fields);
+    }
+
     @Override
     public void buildEqualInfo(JoinInfo joinInfo, AbstractSideTableInfo sideTableInfo) {
         rowKeyBuilder = new RowKeyBuilder();