Solving the problem of data disorganization

ocean-zhc · ocean-zhc · commit bc6b0a3f6d2f · 2025-04-12T23:25:56.000+08:00
diff --git a/seatunnel-connectors-v2/connector-http/connector-http-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/http/source/HttpSourceReader.java b/seatunnel-connectors-v2/connector-http/connector-http-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/http/source/HttpSourceReader.java
@@ -257,31 +257,75 @@ private List<Map<String, String>> parseToMap(List<List<String>> datas, JsonField
 
     private List<List<String>> decodeJSON(String data) {
         ReadContext jsonReadContext = JsonPath.using(jsonConfiguration).parse(data);
-        List<List<String>> results = new ArrayList<>(jsonPaths.length);
-        for (JsonPath path : jsonPaths) {
-            List<String> result = jsonReadContext.read(path);
-            results.add(result);
-        }
+
         if (httpParameter.isJsonFiledMissedReturnNull()) {
-            int maxLength = 0;
-            for (List<?> result : results) {
-                maxLength = Math.max(maxLength, result.size());
+            // Extract the common parent path from all configured paths
+            String commonParentPath = extractCommonParentPath(jsonPaths);
+            if (commonParentPath == null) {
+                throw new HttpConnectorException(
+                        HttpConnectorErrorCode.FIELD_DATA_IS_INCONSISTENT,
+                        "Could not find common parent path in JsonPaths. All paths must share a common array parent.");
+            }
+
+            // Get all objects under the common parent path
+            List<Map<String, Object>> objects;
+            try {
+                objects = jsonReadContext.read(commonParentPath);
+            } catch (Exception e) {
+                throw new HttpConnectorException(
+                        HttpConnectorErrorCode.FIELD_DATA_IS_INCONSISTENT,
+                        "Failed to read data from JSON using path "
+                                + commonParentPath
+                                + ": "
+                                + e.getMessage());
             }
-            for (int i = 0; i < results.size(); i++) {
-                List<String> result = results.get(i);
-                if (result.size() < maxLength) {
-                    log.warn(
-                            "Field [{}] with size ({}) is less than max size ({}), will be padded with null values. "
-                                    + "This may happen when JSON paths return different numbers of elements.",
-                            jsonPaths[i].getPath(),
-                            result.size(),
-                            maxLength);
-                    for (int j = result.size(); j < maxLength; j++) {
-                        result.add(null);
+
+            // Create results list for each field
+            List<List<String>> results = new ArrayList<>(jsonPaths.length);
+            for (int i = 0; i < jsonPaths.length; i++) {
+                results.add(new ArrayList<>(objects.size()));
+            }
+
+            // For each object in the array
+            for (Map<String, Object> obj : objects) {
+                // For each configured field
+                for (int i = 0; i < jsonPaths.length; i++) {
+                    String fieldPath = jsonPaths[i].getPath();
+                    // Get the relative path from the common parent to the field
+                    String relativePath = getRelativePath(commonParentPath, fieldPath);
+
+                    try {
+                        // Create a new context for this object
+                        ReadContext objContext = JsonPath.using(jsonConfiguration).parse(obj);
+                        Object value = objContext.read(relativePath);
+                        String stringValue = null;
+                        if (value != null) {
+                            if (value instanceof List) {
+                                // If it's a list with one element, take that element
+                                List<?> list = (List<?>) value;
+                                if (!list.isEmpty()) {
+                                    stringValue = list.get(0).toString();
+                                }
+                            } else {
+                                stringValue = value.toString();
+                            }
+                        }
+                        results.get(i).add(stringValue);
+                    } catch (Exception e) {
+                        // If field doesn't exist in this object, add null
+                        results.get(i).add(null);
                     }
                 }
             }
+
+            return dataFlip(results);
         } else {
+            // Original implementation for when isJsonFiledMissedReturnNull is false
+            List<List<String>> results = new ArrayList<>(jsonPaths.length);
+            for (JsonPath path : jsonPaths) {
+                List<String> result = jsonReadContext.read(path);
+                results.add(result);
+            }
             for (int i = 1; i < results.size(); i++) {
                 List<?> result0 = results.get(0);
                 List<?> result = results.get(i);
@@ -296,8 +340,56 @@ private List<List<String>> decodeJSON(String data) {
                                     result.size()));
                 }
             }
+
+            return dataFlip(results);
+        }
+    }
+
+    private String extractCommonParentPath(JsonPath[] paths) {
+        if (paths == null || paths.length == 0) {
+            return null;
+        }
+
+        // Get all paths as strings
+        String[] pathStrings = new String[paths.length];
+        for (int i = 0; i < paths.length; i++) {
+            pathStrings[i] = paths[i].getPath();
+        }
+
+        // Find the array notation position in the first path
+        String firstPath = pathStrings[0];
+        int arrayPos = firstPath.indexOf("[*]");
+        if (arrayPos == -1) {
+            throw new HttpConnectorException(
+                    HttpConnectorErrorCode.FIELD_DATA_IS_INCONSISTENT,
+                    "No array notation [*] found in path: " + firstPath);
+        }
+
+        // Get the parent path up to and including [*]
+        String parentPath = firstPath.substring(0, arrayPos + 3);
+
+        // Verify all other paths have the same parent
+        for (int i = 1; i < pathStrings.length; i++) {
+            if (!pathStrings[i].startsWith(parentPath)) {
+                throw new HttpConnectorException(
+                        HttpConnectorErrorCode.FIELD_DATA_IS_INCONSISTENT,
+                        String.format(
+                                "Paths have different array parents. Expected '%s' but found path starting with '%s'",
+                                parentPath, pathStrings[i]));
+            }
+        }
+
+        return parentPath;
+    }
+
+    private String getRelativePath(String parentPath, String fullPath) {
+        // Remove the parent path (including [*]) and keep everything after
+        String relativePart = fullPath.substring(parentPath.length());
+        // If the relative part starts with a dot, remove it
+        if (relativePart.startsWith(".")) {
+            relativePart = relativePart.substring(1);
         }
-        return dataFlip(results);
+        return "$." + relativePart;
     }
 
     private String getPartOfJson(String data) {
diff --git a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-http-e2e/src/test/resources/http_to_console.conf b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-http-e2e/src/test/resources/http_to_console.conf
@@ -25,17 +25,18 @@ env {
 source{
   Http {
     url = "http://mockserver:1080/example/httpFixParsingJson"
+    #json_filed_missed_return_null = true
     method = "POST"
     format = "json"
     json_field = {
-      barcode = "$.result.rows[*].item.barcode"
-      amount = "$.result.rows[*].delivery_commission.amount"
+          key1 = "$[*].key1"
+          key2 = "$[*].key2"
     }
 
     schema = {
       fields {
-        barcode = string
-        amount= string
+        key1 = string
+        key2= string
       }
     }
   }
diff --git a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-http-e2e/src/test/resources/mockserver-config.json b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-http-e2e/src/test/resources/mockserver-config.json
@@ -4780,62 +4780,17 @@
       "method": "POST"
     },
     "httpResponse": {
-      "body": [
+      "body": [{
+        "key1":"value11",
+        "key2":"value22"
+      },
         {
-          "result": {
-            "rows": [
-              {
-                "rowNumber": 1,
-                "item": {
-                  "name": "Джинсы Feimailis",
-                  "offer_id": "AB926-52E-7-30",
-                  "barcode": "OZN201543969",
-                  "sku": 201543969
-                },
-                "seller_price_per_instance": 1028,
-                "delivery_commission": {
-                  "price_per_instance": 782,
-                  "quantity": 1,
-                  "amount": 782,
-                  "compensation": 0,
-                  "commission": 0,
-                  "bonus": 244.04,
-                  "standard_fee": 128.5,
-                  "total": 899.5,
-                  "stars": 0,
-                  "bank_coinvestment": 1.96,
-                  "pick_up_point_coinvestment": 0
-                },
-                "return_commission": null,
-                "commission_ratio": 0.125
-              },
-              {
-                "rowNumber": 2,
-                "item": {
-                  "name": "Джинсы Feimailis",
-                  "offer_id": "AF979-4-25",
-                  "barcode": "OZN201550625",
-                  "sku": 201550625
-                },
-                "seller_price_per_instance": 737,
-                "delivery_commission": {
-                  "price_per_instance": 521,
-                  "quantity": 1,
-                  "amount": 521,
-                  "compensation": 0,
-                  "commission": 0,
-                  "bonus": 214.7,
-                  "standard_fee": 92.13,
-                  "total": 644.87,
-                  "stars": 0,
-                  "bank_coinvestment": 1.3,
-                  "pick_up_point_coinvestment": 0
-                },
-                "return_commission": null,
-                "commission_ratio": 0.125
-              }
-            ]
-          }
+          "key2":"value33",
+          "key3":"value44"
+        },
+        {
+          "key1":"value55",
+          "key3":"value66"
         }
       ],
       "headers": {

Original file line number	Diff line number	Diff line change
`@@ -25,17 +25,18 @@ env {`
`25`	`25`	`source{`
`26`	`26`	`Http {`
`27`	`27`	`url = "http://mockserver:1080/example/httpFixParsingJson"`
	`28`	`+ #json_filed_missed_return_null = true`
`28`	`29`	`method = "POST"`
`29`	`30`	`format = "json"`
`30`	`31`	`json_field = {`
`31`		`- barcode = "$.result.rows[*].item.barcode"`
`32`		`- amount = "$.result.rows[*].delivery_commission.amount"`
	`32`	`+ key1 = "$[*].key1"`
	`33`	`+ key2 = "$[*].key2"`
`33`	`34`	`}`
`34`	`35`
`35`	`36`	`schema = {`
`36`	`37`	`fields {`
`37`		`- barcode = string`
`38`		`- amount= string`
	`38`	`+ key1 = string`
	`39`	`+ key2= string`
`39`	`40`	`}`
`40`	`41`	`}`
`41`	`42`	`}`