[hotfix-37535][core&elasticsearch]Fixed dimension table join not out of the data.

Author: HiLany

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

@@ -34,6 +34,7 @@
 import org.apache.flink.table.runtime.typeutils.BaseRowTypeInfo;
 
 import java.io.Serializable;
+import java.util.Arrays;
 import java.util.List;
 import java.util.Map;
 
@@ -92,17 +93,20 @@ public void parseSelectFields(JoinInfo joinInfo){
         String sideTableName = joinInfo.getSideTableName();
         String nonSideTableName = joinInfo.getNonSideTable();
         List<String> fields = Lists.newArrayList();
-        int sideTableFieldIndex = 0;
+        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);
-                sideTableFieldIndex++;
             }else if(fieldInfo.getTable().equalsIgnoreCase(nonSideTableName)){
                 int nonSideIndex = rowTypeInfo.getFieldIndex(fieldInfo.getFieldName());
                 inFieldIndex.put(i, nonSideIndex);

elasticsearch7/elasticsearch7-side/elasticsearch7-all-side/src/main/java/com/dtstack/flink/sql/side/elasticsearch7/Elasticsearch7AllReqRow.java

@@ -208,7 +208,6 @@ private SearchSourceBuilder initConfiguration(BoolQueryBuilder boolQueryBuilder)
         }
 
         searchSourceBuilder.size(getFetchSize());
-        searchSourceBuilder.sort("_id", SortOrder.DESC);
 
         // fields included in the source data
         String[] sideFieldNames = StringUtils.split(sideInfo.getSideSelectFields().trim(), ",");

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

@@ -218,7 +218,6 @@ public void close() throws Exception {
     private SearchSourceBuilder initConfiguration() {
         SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
         searchSourceBuilder.size(getFetchSize());
-        searchSourceBuilder.sort("_id", SortOrder.DESC);
         String[] sideFieldNames = StringUtils.split(sideInfo.getSideSelectFields().trim(), ",");
         searchSourceBuilder.fetchSource(sideFieldNames, null);