Merge branch 'hotfix_1.10_4.0.x_31672' into 1.10_test_4.0.x

Author: gituser

elasticsearch6/elasticsearch6-side/elasticsearch6-all-side/src/main/java/com/dtstack/flink/sql/side/elasticsearch6/Elasticsearch6AllReqRow.java

@@ -24,7 +24,6 @@
 import com.dtstack.flink.sql.side.JoinInfo;
 import com.dtstack.flink.sql.side.elasticsearch6.table.Elasticsearch6SideTableInfo;
 import com.dtstack.flink.sql.side.elasticsearch6.util.Es6Util;
-import com.dtstack.flink.sql.side.elasticsearch6.util.SwitchUtil;
 import com.google.common.collect.Lists;
 import com.google.common.collect.Maps;
 import org.apache.commons.collections.CollectionUtils;
@@ -235,7 +234,7 @@ private void loadToCache(SearchHit[] searchHits, Map<String, List<Map<String, Ob
             for (String fieldName : sideFieldNames) {
                 Object object = searchHit.getSourceAsMap().get(fieldName.trim());
                 int fieldIndex = sideInfo.getSideTableInfo().getFieldList().indexOf(fieldName.trim());
-                object = SwitchUtil.getTarget(object, sideFieldTypes[fieldIndex]);
+                object = Es6Util.getTarget(object, sideFieldTypes[fieldIndex]);
                 oneRow.put(fieldName.trim(), object);
             }
 

elasticsearch6/elasticsearch6-side/elasticsearch6-async-side/src/main/java/com/dtstack/flink/sql/side/elasticsearch6/Elasticsearch6AsyncReqRow.java

@@ -18,6 +18,7 @@
 
 package com.dtstack.flink.sql.side.elasticsearch6;
 
+import com.dtstack.flink.sql.side.elasticsearch6.util.Es6Util;
 import com.dtstack.flink.sql.util.RowDataComplete;
 import org.apache.flink.api.java.typeutils.RowTypeInfo;
 import org.apache.flink.configuration.Configuration;
@@ -30,8 +31,6 @@
 import com.dtstack.flink.sql.side.*;
 import com.dtstack.flink.sql.side.cache.CacheObj;
 import com.dtstack.flink.sql.side.elasticsearch6.table.Elasticsearch6SideTableInfo;
-import com.dtstack.flink.sql.side.elasticsearch6.util.Es6Util;
-import com.dtstack.flink.sql.side.elasticsearch6.util.SwitchUtil;
 import com.dtstack.flink.sql.util.ParseUtils;
 import com.google.common.collect.Lists;
 import org.apache.calcite.sql.SqlNode;
@@ -110,7 +109,7 @@ public void onResponse(SearchResponse searchResponse) {
                             if (searchHits.length < getFetchSize()) {
                                 break;
                             }
-                            if (tableInfo == null && tmpRhlClient == null) {
+                            if (tableInfo == null) {
                                 // create new connection to fetch data
                                 tableInfo = (Elasticsearch6SideTableInfo) sideInfo.getSideTableInfo();
                                 tmpRhlClient = Es6Util.getClient(tableInfo.getAddress(), tableInfo.isAuthMesh(), tableInfo.getUserName(), tableInfo.getPassword());
@@ -201,7 +200,7 @@ public Row fillData(Row input, Object line) {
             if (cacheInfo == null) {
                 row.setField(entry.getKey(), null);
             } else {
-                Object object = SwitchUtil.getTarget(cacheInfo.get(sideInfo.getSideFieldNameIndex().get(entry.getKey())), fields[entry.getValue()]);
+                Object object = Es6Util.getTarget(cacheInfo.get(sideInfo.getSideFieldNameIndex().get(entry.getKey())), fields[entry.getValue()]);
                 row.setField(entry.getKey(), object);
             }
         }

elasticsearch6/elasticsearch6-side/elasticsearch6-side-core/src/main/java/com/dtstack/flink/sql/side/elasticsearch6/table/Elasticsearch6SideParser.java

@@ -18,9 +18,9 @@
 
 package com.dtstack.flink.sql.side.elasticsearch6.table;
 
-import com.dtstack.flink.sql.side.elasticsearch6.util.ClassUtil;
 import com.dtstack.flink.sql.table.AbstractSideTableParser;
 import com.dtstack.flink.sql.table.AbstractTableInfo;
+import com.dtstack.flink.sql.util.ClassUtil;
 import com.dtstack.flink.sql.util.MathUtil;
 import org.apache.commons.lang3.StringUtils;
 

elasticsearch6/elasticsearch6-side/elasticsearch6-side-core/src/main/java/com/dtstack/flink/sql/side/elasticsearch6/util/ClassUtil.java

@@ -21,6 +21,7 @@
 import com.dtstack.flink.sql.side.BaseSideInfo;
 import com.dtstack.flink.sql.side.PredicateInfo;
 import com.dtstack.flink.sql.side.elasticsearch6.table.Elasticsearch6SideTableInfo;
+import com.dtstack.flink.sql.util.MathUtil;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.http.HttpHost;
 import org.apache.http.auth.AuthScope;
@@ -40,6 +41,7 @@
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Locale;
 
 /**
  * @author yinxi
@@ -49,7 +51,7 @@ public class Es6Util {
 
     private static final Logger LOG = LoggerFactory.getLogger(Es6Util.class);
     private static final String KEY_WORD_TYPE = ".keyword";
-    private static final String APOSTROPHE = "\'";
+    private static final String APOSTROPHE = "'";
 
     // connect to the elasticsearch
     public static RestHighLevelClient getClient(String esAddress, Boolean isAuthMesh, String userName, String password) {
@@ -60,12 +62,12 @@ public static RestHighLevelClient getClient(String esAddress, Boolean isAuthMesh
             int port = 9200;
             String host = infoArray[0].trim();
             if (infoArray.length > 1) {
-                port = Integer.valueOf(infoArray[1].trim());
+                port = Integer.parseInt(infoArray[1].trim());
             }
             httpHostList.add(new HttpHost(host, port, "http"));
         }
 
-        RestClientBuilder restClientBuilder = RestClient.builder(httpHostList.toArray(new HttpHost[httpHostList.size()]));
+        RestClientBuilder restClientBuilder = RestClient.builder(httpHostList.toArray(new HttpHost[0]));
 
         if (isAuthMesh) {
             // 进行用户和密码认证
@@ -171,6 +173,9 @@ public static BoolQueryBuilder buildFilterCondition(BoolQueryBuilder boolQueryBu
                 return boolQueryBuilder.must(QueryBuilders.existsQuery(info.getFieldName()));
             case "=":
             case "EQUALS":
+                if (StringUtils.isBlank(info.getCondition())) {
+                    return boolQueryBuilder;
+                }
                 return boolQueryBuilder.must(QueryBuilders.termQuery(textConvertToKeyword(info.getFieldName(), sideInfo), removeSpaceAndApostrophe(info.getCondition())[0]));
             case "<>":
             case "NOT_EQUALS":
@@ -199,7 +204,7 @@ public static String textConvertToKeyword(String fieldName, BaseSideInfo sideInf
         String[] sideFieldTypes = sideInfo.getSideTableInfo().getFieldTypes();
         int fieldIndex = sideInfo.getSideTableInfo().getFieldList().indexOf(fieldName.trim());
         String fieldType = sideFieldTypes[fieldIndex];
-        switch (fieldType.toLowerCase()) {
+        switch (fieldType.toLowerCase(Locale.ENGLISH)) {
             case "varchar":
             case "char":
             case "text":
@@ -208,4 +213,59 @@ public static String textConvertToKeyword(String fieldName, BaseSideInfo sideInf
                 return fieldName;
         }
     }
+
+    public static Object getTarget(Object obj, String targetType) {
+        switch (targetType.toLowerCase(Locale.ENGLISH)) {
+
+            case "smallint":
+            case "smallintunsigned":
+            case "tinyint":
+            case "tinyintunsigned":
+            case "mediumint":
+            case "mediumintunsigned":
+            case "integer":
+            case "int":
+                return MathUtil.getIntegerVal(obj);
+
+            case "bigint":
+            case "bigintunsigned":
+            case "intunsigned":
+            case "integerunsigned":
+                return MathUtil.getLongVal(obj);
+
+            case "boolean":
+                return MathUtil.getBoolean(obj);
+
+            case "blob":
+                return MathUtil.getByte(obj);
+
+            case "varchar":
+            case "char":
+            case "text":
+                return MathUtil.getString(obj);
+
+            case "real":
+            case "float":
+            case "realunsigned":
+            case "floatunsigned":
+                return MathUtil.getFloatVal(obj);
+
+            case "double":
+            case "doubleunsigned":
+                return MathUtil.getDoubleVal(obj);
+
+            case "decimal":
+            case "decimalunsigned":
+                return MathUtil.getBigDecimal(obj);
+
+            case "date":
+                return MathUtil.getDate(obj);
+
+            case "timestamp":
+            case "datetime":
+                return MathUtil.getTimestamp(obj);
+            default:
+        }
+        return obj;
+    }
 }