From ffd5f1b43bc68ffa61b85e5f36166963fa47c070 Mon Sep 17 00:00:00 2001
From: Xincheng Huang <60057611+ying-hua@users.noreply.github.com>
Date: Wed, 28 Aug 2024 14:12:21 +0800
Subject: [PATCH] [INLONG-10872][SDK] Add a More Complex Json Test Case
 (#10885)

---
 .../process/TestTransformProcessor.java       | 74 +++++++++++++++----
 1 file changed, 61 insertions(+), 13 deletions(-)

diff --git a/inlong-sdk/transform-sdk/src/test/java/org/apache/inlong/sdk/transform/process/TestTransformProcessor.java b/inlong-sdk/transform-sdk/src/test/java/org/apache/inlong/sdk/transform/process/TestTransformProcessor.java
index 52838ab06c7..20af097de60 100644
--- a/inlong-sdk/transform-sdk/src/test/java/org/apache/inlong/sdk/transform/process/TestTransformProcessor.java
+++ b/inlong-sdk/transform-sdk/src/test/java/org/apache/inlong/sdk/transform/process/TestTransformProcessor.java
@@ -148,16 +148,16 @@ public void testKv2CsvNoField() throws Exception {
 
     @Test
     public void testJson2Csv() throws Exception {
-        List<FieldInfo> fields = this.getTestFieldList();
-        JsonSourceInfo jsonSource = new JsonSourceInfo("UTF-8", "msgs");
-        CsvSinkInfo csvSink = new CsvSinkInfo("UTF-8", '|', '\\', fields);
-        String transformSql = "select $root.sid,$root.packageID,$child.msgTime,$child.msg from source";
-        TransformConfig config = new TransformConfig(transformSql);
+        List<FieldInfo> fields1 = this.getTestFieldList();
+        JsonSourceInfo jsonSource1 = new JsonSourceInfo("UTF-8", "msgs");
+        CsvSinkInfo csvSink1 = new CsvSinkInfo("UTF-8", '|', '\\', fields1);
+        String transformSql1 = "select $root.sid,$root.packageID,$child.msgTime,$child.msg from source";
+        TransformConfig config1 = new TransformConfig(transformSql1);
         // case1
-        TransformProcessor<String, String> processor = TransformProcessor
-                .create(config, SourceDecoderFactory.createJsonDecoder(jsonSource),
-                        SinkEncoderFactory.createCsvEncoder(csvSink));
-        String srcString = "{\n"
+        TransformProcessor<String, String> processor1 = TransformProcessor
+                .create(config1, SourceDecoderFactory.createJsonDecoder(jsonSource1),
+                        SinkEncoderFactory.createCsvEncoder(csvSink1));
+        String srcString1 = "{\n"
                 + "  \"sid\":\"value1\",\n"
                 + "  \"packageID\":\"value2\",\n"
                 + "  \"msgs\":[\n"
@@ -165,10 +165,42 @@ public void testJson2Csv() throws Exception {
                 + "  {\"msg\":\"v4\",\"msgTime\":1713243918000}\n"
                 + "  ]\n"
                 + "}";
-        List<String> output = processor.transform(srcString, new HashMap<>());
-        Assert.assertEquals(2, output.size());
-        Assert.assertEquals(output.get(0), "value1|value2|1713243918000|value4");
-        Assert.assertEquals(output.get(1), "value1|value2|1713243918000|v4");
+        List<String> output1 = processor1.transform(srcString1, new HashMap<>());
+        Assert.assertEquals(2, output1.size());
+        Assert.assertEquals(output1.get(0), "value1|value2|1713243918000|value4");
+        Assert.assertEquals(output1.get(1), "value1|value2|1713243918000|v4");
+        // case2
+        List<FieldInfo> fields2 = this.getTestFieldList2();
+        JsonSourceInfo jsonSource2 = new JsonSourceInfo("UTF-8", "items");
+        CsvSinkInfo csvSink2 = new CsvSinkInfo("UTF-8", '|', '\\', fields2);
+        String transformSql2 =
+                "select $root.id,$child.itemId,$child.subItems(0).subItemId,$child.subItems(1).msg from source";
+        TransformConfig config2 = new TransformConfig(transformSql2);
+        TransformProcessor<String, String> processor2 = TransformProcessor
+                .create(config2, SourceDecoderFactory.createJsonDecoder(jsonSource2),
+                        SinkEncoderFactory.createCsvEncoder(csvSink2));
+        String srcString2 = "{\n"
+                + "  \"id\":\"value1\",\n"
+                + "  \"name\":\"value2\",\n"
+                + "  \"items\":[\n"
+                + "    {\"itemId\":\"item1\",\n"
+                + "     \"subItems\":[\n"
+                + "       {\"subItemId\":\"1001\", \"msg\":\"1001msg\"},\n"
+                + "       {\"subItemId\":\"1002\", \"msg\":\"1002msg\"}\n"
+                + "     ]\n"
+                + "    },\n"
+                + "    {\"itemId\":\"item2\",\n"
+                + "     \"subItems\":[\n"
+                + "       {\"subItemId\":\"2001\", \"msg\":\"2001msg\"},\n"
+                + "       {\"subItemId\":\"2002\", \"msg\":\"2002msg\"}\n"
+                + "     ]\n"
+                + "    }\n"
+                + "  ]\n"
+                + "}";
+        List<String> output2 = processor2.transform(srcString2, new HashMap<>());
+        Assert.assertEquals(2, output2.size());
+        Assert.assertEquals(output2.get(0), "value1|item1|1001|1002msg");
+        Assert.assertEquals(output2.get(1), "value1|item2|2001|2002msg");
     }
 
     @Test
@@ -230,6 +262,22 @@ private List<FieldInfo> getTestFieldList() {
         fields.add(msg);
         return fields;
     }
+    private List<FieldInfo> getTestFieldList2() {
+        List<FieldInfo> fields = new ArrayList<>();
+        FieldInfo id = new FieldInfo();
+        id.setName("id");
+        fields.add(id);
+        FieldInfo itemId = new FieldInfo();
+        itemId.setName("itemId");
+        fields.add(itemId);
+        FieldInfo subItemId = new FieldInfo();
+        subItemId.setName("subItemId");
+        fields.add(subItemId);
+        FieldInfo msg = new FieldInfo();
+        msg.setName("msg");
+        fields.add(msg);
+        return fields;
+    }
 
     private byte[] getPbTestData() {
         String srcString =