Merge branch '1.8_release_3.10.x' into 1.8_release_4.0.x

Author: gituser

core/src/main/java/com/dtstack/flink/sql/format/dtnest/DtNestRowDeserializationSchema.java

@@ -106,11 +106,7 @@ private void parseTree(JsonNode jsonNode, String prefix) {
             String nodeKey = getNodeKey(prefix, next);
 
             nodeAndJsonNodeMapping.put(nodeKey, child);
-            if (child.isArray()) {
-                parseTree(child, nodeKey);
-            } else {
-                parseTree(child, nodeKey);
-            }
+            parseTree(child, nodeKey);
         }
     }
 

core/src/main/java/com/dtstack/flink/sql/table/AbstractTableInfo.java

@@ -16,8 +16,6 @@
  * limitations under the License.
  */
 
- 
-
 package com.dtstack.flink.sql.table;
 
 import com.google.common.collect.Lists;
@@ -178,9 +176,9 @@ public void setPhysicalFields(Map<String, String> physicalFields) {
     }
 
     public void finish(){
-        this.fields = fieldList.toArray(new String[fieldList.size()]);
-        this.fieldClasses = fieldClassList.toArray(new Class[fieldClassList.size()]);
-        this.fieldTypes = fieldTypeList.toArray(new String[fieldTypeList.size()]);
+        this.fields = fieldList.toArray(new String[0]);
+        this.fieldClasses = fieldClassList.toArray(new Class[0]);
+        this.fieldTypes = fieldTypeList.toArray(new String[0]);
     }
 
     /**

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

@@ -108,9 +108,22 @@ public static Date localDateTimetoDate(LocalDateTime localDateTime){
     }
 
     public static LocalDateTime dateToLocalDateTime(Date date){
+        date = transformSqlDateToUtilDate(date);
         return date.toInstant().atZone(ZoneId.systemDefault()).toLocalDateTime();
     }
 
+    /**
+     * 将java.sql.Date 转化为 java.util.Date
+     * @param date 不知道是java.sql.Date 还是 java.util.Date
+     * @return 最后返回 java.util.Date
+     */
+    public  static Date transformSqlDateToUtilDate(Date date) {
+        if (date instanceof java.sql.Date) {
+            date = new Date(date.getTime());
+        }
+        return date;
+    }
+
     /**
      *
      * 

elasticsearch6/elasticsearch6-sink/src/main/java/com/dtstack/flink/sql/sink/elasticsearch/CustomerSinkFunc.java

@@ -18,6 +18,7 @@
 
 package com.dtstack.flink.sql.sink.elasticsearch;
 
+import com.dtstack.flink.sql.util.DateUtil;
 import org.apache.flink.api.common.functions.RuntimeContext;
 import org.apache.flink.api.java.tuple.Tuple2;
 import org.apache.flink.metrics.Counter;
@@ -31,6 +32,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.sql.Date;
 import java.util.List;
 import java.util.Map;
 import java.util.stream.Collectors;
@@ -42,7 +44,9 @@
 public class CustomerSinkFunc implements ElasticsearchSinkFunction<Tuple2> {
 
     private final Logger logger = LoggerFactory.getLogger(CustomerSinkFunc.class);
-    /** 用作ID的属性值连接符号 */
+    /**
+     * 用作ID的属性值连接符号
+     */
     private static final String ID_VALUE_SPLIT = "_";
 
     private String index;
@@ -107,6 +111,10 @@ private IndexRequest createIndexRequest(Row element) {
         Map<String, Object> dataMap = Es6Util.rowToJsonMap(element, fieldNames, fieldTypes);
         int length = Math.min(element.getArity(), fieldNames.size());
         for (int i = 0; i < length; i++) {
+            if (element.getField(i) instanceof Date) {
+                dataMap.put(fieldNames.get(i), DateUtil.transformSqlDateToUtilDate((Date) element.getField(i)));
+                continue;
+            }
             dataMap.put(fieldNames.get(i), element.getField(i));
         }