DT_NEST deserialization support ObjeactArray and Row Type

Author: a49a

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

@@ -24,6 +24,7 @@
 import org.apache.flink.api.common.serialization.AbstractDeserializationSchema;
 import org.apache.flink.api.common.typeinfo.TypeInformation;
 import org.apache.flink.api.common.typeinfo.Types;
+import org.apache.flink.api.java.typeutils.ObjectArrayTypeInfo;
 import org.apache.flink.api.java.typeutils.RowTypeInfo;
 import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonProcessingException;
 import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.JsonNode;
@@ -34,6 +35,7 @@
 import org.apache.flink.types.Row;
 
 import java.io.IOException;
+import java.lang.reflect.Array;
 import java.sql.Date;
 import java.sql.Time;
 import java.sql.Timestamp;
@@ -71,30 +73,7 @@ public DtNestRowDeserializationSchema(TypeInformation<Row> typeInfo, Map<String,
     public Row deserialize(byte[] message) throws IOException {
         JsonNode root = objectMapper.readTree(message);
         this.parseTree(root, null);
-        Row row = new Row(fieldNames.length);
-
-        try {
-            for (int i = 0; i < fieldNames.length; i++) {
-                JsonNode node = getIgnoreCase(fieldNames[i]);
-                AbstractTableInfo.FieldExtraInfo fieldExtraInfo = fieldExtraInfos.get(i);
-
-                if (node == null) {
-                    if (fieldExtraInfo != null && fieldExtraInfo.getNotNull()) {
-                        throw new IllegalStateException("Failed to find field with name '"
-                                + fieldNames[i] + "'.");
-                    } else {
-                        row.setField(i, null);
-                    }
-                } else {
-                    // Read the value as specified type
-                    Object value = convert(node, fieldTypes[i]);
-                    row.setField(i, value);
-                }
-            }
-            return row;
-        } finally {
-            nodeAndJsonNodeMapping.clear();
-        }
+        return convertTopRow();
     }
 
     private void parseTree(JsonNode jsonNode, String prefix){
@@ -168,7 +147,11 @@ private Object convert(JsonNode node, TypeInformation<?> info) {
         } else if (info.getTypeClass().equals(Types.SQL_TIMESTAMP.getTypeClass())) {
             // local zone
             return Timestamp.valueOf(node.asText());
-        }  else {
+        }  else if (info instanceof RowTypeInfo) {
+            return convertRow(node, (RowTypeInfo) info);
+        } else if (info instanceof ObjectArrayTypeInfo) {
+            return convertObjectArray(node, ((ObjectArrayTypeInfo) info).getComponentInfo());
+        } else {
             // for types that were specified without JSON schema
             // e.g. POJOs
             try {
@@ -179,5 +162,55 @@ private Object convert(JsonNode node, TypeInformation<?> info) {
         }
     }
 
+    private Row convertTopRow() {
+        Row row = new Row(fieldNames.length);
+        try {
+            for (int i = 0; i < fieldNames.length; i++) {
+                JsonNode node = getIgnoreCase(fieldNames[i]);
+                AbstractTableInfo.FieldExtraInfo fieldExtraInfo = fieldExtraInfos.get(i);
 
+                if (node == null) {
+                    if (fieldExtraInfo != null && fieldExtraInfo.getNotNull()) {
+                        throw new IllegalStateException("Failed to find field with name '"
+                            + fieldNames[i] + "'.");
+                    } else {
+                        row.setField(i, null);
+                    }
+                } else {
+                    // Read the value as specified type
+                    Object value = convert(node, fieldTypes[i]);
+                    row.setField(i, value);
+                }
+            }
+            return row;
+        } finally {
+            nodeAndJsonNodeMapping.clear();
+        }
+    }
+
+    private Row convertRow(JsonNode node, RowTypeInfo info) {
+        final String[] names = info.getFieldNames();
+        final TypeInformation<?>[] types = info.getFieldTypes();
+
+        final Row row = new Row(names.length);
+        for (int i = 0; i < names.length; i++) {
+            final String name = names[i];
+            final JsonNode subNode = node.get(name);
+            if (subNode == null) {
+                row.setField(i, null);
+            } else {
+                row.setField(i, convert(subNode, types[i]));
+            }
+        }
+
+        return row;
+    }
+
+    private Object convertObjectArray(JsonNode node, TypeInformation<?> elementType) {
+        final Object[] array = (Object[]) Array.newInstance(elementType.getTypeClass(), node.size());
+        for (int i = 0; i < node.size(); i++) {
+            array[i] = convert(node.get(i), elementType);
+        }
+        return array;
+    }
 }