From e358c63494f444b691eb4316961e80b12895c7f7 Mon Sep 17 00:00:00 2001
From: redscarf <hongweijin1993@gmail.com>
Date: Fri, 2 Feb 2024 18:20:43 +0800
Subject: [PATCH] [Bugfix]: write nest json schema

Signed-off-by: redscarf <hongweijin1993@gmail.com>
---
 .../row/sink/StarRocksTableRowTransformer.java | 18 +++++++++++++++---
 1 file changed, 15 insertions(+), 3 deletions(-)

diff --git a/src/main/java/com/starrocks/connector/flink/row/sink/StarRocksTableRowTransformer.java b/src/main/java/com/starrocks/connector/flink/row/sink/StarRocksTableRowTransformer.java
index aef0c02bd..7d73a5a5e 100644
--- a/src/main/java/com/starrocks/connector/flink/row/sink/StarRocksTableRowTransformer.java
+++ b/src/main/java/com/starrocks/connector/flink/row/sink/StarRocksTableRowTransformer.java
@@ -186,7 +186,19 @@ private Object typeConvertion(LogicalType type, RowData record, int pos) {
 
     private List<Object> convertNestedArray(ArrayData arrData, LogicalType type) {
         if (arrData instanceof GenericArrayData) {
-            return Lists.newArrayList(((GenericArrayData)arrData).toObjectArray());
+            RowType rType = (RowType) ((ArrayType) type).getElementType();
+            List<Object> list = Lists.newLinkedList();
+            for (Object data : ((GenericArrayData) arrData).toObjectArray()) {
+                if (data instanceof RowData) {
+                    Map<String, Object> map = Maps.newHashMap();
+                    rType.getFields().forEach(field -> map.put(field.getName(), typeConvertion(field.getType(), (RowData) data, rType.getFieldIndex(field.getName()))));
+                    list.add(map);
+                } else {
+                    list.add(data);
+                }
+            }
+
+            return list;
         }
         if (arrData instanceof BinaryArrayData) {
             LogicalType lt = ((ArrayType)type).getElementType();
@@ -197,7 +209,7 @@ private List<Object> convertNestedArray(ArrayData arrData, LogicalType type) {
                 return data.parallelStream().map(row -> {
                     Map<String, Object> m = Maps.newHashMap();
                     rType.getFields().parallelStream().forEach(f -> m.put(f.getName(), typeConvertion(f.getType(), (RowData)row, rType.getFieldIndex(f.getName()))));
-                    return jsonWrapper.toJSONString(m);
+                    return m;
                 }).collect(Collectors.toList());
             }
             if (LogicalTypeRoot.MAP.equals(lt.getTypeRoot())) {
@@ -248,5 +260,5 @@ private Map<Object, Object> convertNestedMap(MapData mapData, LogicalType type)
         }
         throw new UnsupportedOperationException(String.format("Unsupported map data: %s", mapData.getClass()));
     }
-    
+
 }