[hotfix-37535][elasticsearch7]override parseSelectFields in Elasticsearch7AsyncSideInfo

Author: HiLany

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

@@ -34,7 +34,6 @@
 import org.apache.flink.table.runtime.typeutils.BaseRowTypeInfo;
 
 import java.io.Serializable;
-import java.util.Arrays;
 import java.util.List;
 import java.util.Map;
 
@@ -93,20 +92,17 @@ public void parseSelectFields(JoinInfo joinInfo){
         String sideTableName = joinInfo.getSideTableName();
         String nonSideTableName = joinInfo.getNonSideTable();
         List<String> fields = Lists.newArrayList();
-        int sideTableFieldIndex;
+        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);
-                sideTableFieldIndex = Arrays.asList(sideTableInfo.getFields()).indexOf(sideFieldName);
-                if (sideTableFieldIndex == -1){
-                    throw new RuntimeException(String.format("unknown filed {%s} in sideTable {%s} ", sideFieldName, sideTableName));
-                }
                 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);

elasticsearch7/elasticsearch7-side/elasticsearch7-async-side/src/main/java/com/dtstack/flink/sql/side/elasticsearch7/Elasticsearch7AsyncSideInfo.java

@@ -28,6 +28,7 @@
 import org.apache.calcite.sql.SqlNode;
 import org.apache.flink.api.java.typeutils.RowTypeInfo;
 
+import java.util.Arrays;
 import java.util.List;
 
 /**
@@ -43,6 +44,40 @@ public Elasticsearch7AsyncSideInfo(RowTypeInfo rowTypeInfo, JoinInfo joinInfo, L
         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;
+
+        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);
+                sideTableFieldIndex = Arrays.asList(sideTableInfo.getFields()).indexOf(sideFieldName);
+                if (sideTableFieldIndex == -1){
+                    throw new RuntimeException(String.format("unknown filed {%s} in sideTable {%s} ", sideFieldName, sideTableName));
+                }
+                sideSelectFieldsType.put(sideTableFieldIndex, getTargetFieldType(fieldInfo.getFieldName()));
+                sideFieldIndex.put(i, sideTableFieldIndex);
+                sideFieldNameIndex.put(i, sideFieldName);
+            }else if(fieldInfo.getTable().equalsIgnoreCase(nonSideTableName)){
+                int nonSideIndex = rowTypeInfo.getFieldIndex(fieldInfo.getFieldName());
+                inFieldIndex.put(i, nonSideIndex);
+            }else{
+                throw new RuntimeException("unknown table " + fieldInfo.getTable());
+            }
+        }
+
+        if(fields.size() == 0){
+            throw new RuntimeException("select non field from table " +  sideTableName);
+        }
+
+        sideSelectFields = String.join(",", fields);
+    }
+
     @Override
     public void buildEqualInfo(JoinInfo joinInfo, AbstractSideTableInfo sideTableInfo) {