[INLONG-12181][SDK] Add JSON struct/array extraction functions for transform-sdk: json_to_struct, json_to_array, json_extract_struct, json_extract_struct_excluding

[luchunliang]

Fixes #12181

Motivation

Description
Add four JSON data extraction functions to the transform-sdk, enabling users to convert JSON objects and arrays into Flink's GenericRowData / GenericArrayData structures within transform SQL expressions.

New Functions

1. json_extract_struct(path, field1, field2, ...)
   Extracts specified fields from a JSON object or array path and returns structured data.

path: JSON path expression (e.g., $root.person)
field1, field2, ...: field names to include in the result (required)
Returns:
GenericRowData when path resolves to a JSON object — containing the specified fields in order
GenericArrayData when path resolves to a JSON array of objects
NULL if path does not exist or resolves to a non-struct type
Supports nested paths (e.g., address.city) for field extraction
Supports nested json_extract_struct as path or field value
Example: json_extract_struct($root.person, name, age) → GenericRowData[name, age]
2. json_extract_struct_excluding(path, excludeField1, excludeField2, ...)
Extracts all fields from a JSON object/array except the ones explicitly excluded.

path: JSON path expression
excludeField1, excludeField2, ...: field names to exclude
Returns: GenericRowData (object) or GenericArrayData (array) with all fields except excluded ones, in their original JSON order
Supports nested usage
Example: json_extract_struct_excluding($root.person, address, phone) → all person fields except address and phone
3. json_to_array(path)
Converts a JSON array path into a GenericArrayData with full recursive element conversion — no field filtering.

path: JSON path expression; must resolve to a JsonArray, otherwise returns NULL
All elements are recursively converted:
JsonObject → GenericRowData (all fields preserved)
Nested JsonArray → GenericArrayData
String → BinaryStringData, Boolean → Boolean, Number → Number
JsonNull → null
Example: json_to_array($root.items) → GenericArrayData of fully-converted elements
4. json_to_struct(path)
Converts a JSON object path into a GenericRowData with full recursive field conversion — no field filtering.

path: JSON path expression; must resolve to a JsonObject, otherwise returns NULL
All fields included in original JSON order, recursively converted (same rules as json_to_array)
Example: json_to_struct($root.person) → GenericRowData with all person fields
Function Comparison
Function Path Type Returns Field Control Nested Support
json_extract_struct Object / Array of Objects RowData / ArrayData Whitelist (specify fields) Yes (nested path + nested func)
json_extract_struct_excluding Object / Array of Objects RowData / ArrayData Blacklist (exclude fields) Yes
json_to_array Array only ArrayData All fields included Full recursive conversion
json_to_struct Object only RowData All fields included Full recursive conversion

Modifications

Files Changed
New files:

inlong-sdk/transform-sdk/src/main/java/.../function/json/JsonToArrayFunction.java
inlong-sdk/transform-sdk/src/main/java/.../function/json/JsonToStructFunction.java
Modified files:

inlong-sdk/transform-sdk/src/main/java/.../function/json/JsonExtractStructFunction.java — nested struct support, nested path field extraction
inlong-sdk/transform-sdk/src/main/java/.../function/json/JsonExtractStructExcludingFunction.java — nested struct support, element conversion fix
inlong-sdk/transform-sdk/src/main/java/.../utils/FieldToRowDataUtils.java — ARRAY type handler fallback for GenericRowData elements
inlong-sdk/transform-sdk/src/test/java/.../processor/TestJson2RowDataProcessor.java — 28 comprehensive test cases

Verifying this change

(Please pick either of the following options)

• This change is a trivial rework/code cleanup without any test coverage.

• This change is already covered by existing tests, such as:
  (please describe tests)

• This change added tests and can be verified as follows:

  (example:)

  • Added integration tests for end-to-end deployment with large payloads (10MB)
  • Extended integration test for recovery after broker failure

Documentation

• Does this pull request introduce a new feature? (yes / no)
• If yes, how is the feature documented? (not applicable / docs / JavaDocs / not documented)
• If a feature is not applicable for documentation, explain why?
• If a feature is not documented yet in this PR, please create a follow-up issue for adding the documentation