Merge branch '1.10_release_4.2.x' into feat_1.10_4.2.x_optFile

Author: FlechazoW

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

@@ -224,9 +224,13 @@ private Map<String, Object> parseInputParam(BaseRow input) {
         for (int i = 0; i < sideInfo.getEqualValIndex().size(); i++) {
             Integer conValIndex = sideInfo.getEqualValIndex().get(i);
             Object equalObj = genericRow.getField(conValIndex);
-            if (equalObj == null) {
-                return inputParams;
-            }
+            // comment by tiezhu
+            // 假设SQL中有三个主键[a, b, c]，同时主键[b]的值为null，那么
+            // inputParams中只会有主键[a]的值，主键[b, c]都不包含，导致
+            // 后面rdb 维表替换where 条件时缺少 value，查询SQL 执行失败
+//            if (equalObj == null) {
+//                return inputParams;
+//            }
             String columnName = sideInfo.getEqualFieldList().get(i);
             inputParams.put(columnName, equalObj);
         }

core/src/main/java/com/dtstack/flink/sql/util/MathUtil.java

@@ -129,6 +129,8 @@ public static Double getDoubleVal(Object obj) {
             return ((BigDecimal) obj).doubleValue();
         } else if (obj instanceof Integer) {
             return ((Integer) obj).doubleValue();
+        } else if (obj instanceof BigInteger) {
+            return ((BigInteger) obj).doubleValue();
         }
 
         throw new RuntimeException("not support type of " + obj.getClass() + " convert to Double.");

elasticsearch5-xh/elasticsearch5-xh-sink/src/main/java/com/dtstack/flink/sql/sink/elasticsearch/table/ElasticsearchTableInfo.java

@@ -166,7 +166,7 @@ public boolean check() {
                         Strings.isNullOrEmpty(keytab) &&
                         Strings.isNullOrEmpty(krb5conf);
 
-        Preconditions.checkState(allNotSet, "xh's elasticsearch type of kerberos file is required");
+        Preconditions.checkState(!allNotSet, "xh's elasticsearch type of kerberos file is required");
 
         return true;
     }

kudu/kudu-side/kudu-all-side/src/main/java/com/dtstack/flink/sql/side/kudu/KuduAllReqRow.java

@@ -150,7 +150,7 @@ private void loadData(Map<String, List<Map<String, Object>>> tmpCache) {
                         String sideFieldName = sideFieldName1.trim();
                         ColumnSchema columnSchema = table.getSchema().getColumn(sideFieldName);
                         if (null != columnSchema) {
-                            KuduUtil.setMapValue(columnSchema.getType(), oneRow, sideFieldName, result);
+                            oneRow.put(sideFieldName, result.getObject(sideFieldName));
                         }
                     }
                     String cacheKey = buildKey(oneRow, sideInfo.getEqualFieldList());

kudu/kudu-side/kudu-async-side/src/main/java/com/dtstack/flink/sql/side/kudu/KuduAsyncReqRow.java

@@ -301,7 +301,7 @@ public Deferred<List<Row>> call(RowResultIterator results) throws Exception {
                     String sideFieldName = sideFieldName1.trim();
                     ColumnSchema columnSchema = table.getSchema().getColumn(sideFieldName);
                     if (null != columnSchema) {
-                        KuduUtil.setMapValue(columnSchema.getType(), oneRow, sideFieldName, result);
+                        oneRow.put(sideFieldName, result.getObject(sideFieldName));
                     }
                 }
                 BaseRow row = fillData(input, oneRow);