[fix-31383][hbase]hbase dim table just filter stream data, need not select field

Author: simenliuxing

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