Native collection expressions: track remaining ARRAY/MAP functions, casts and aggregate results

[liuyongvs]

Verified on PR #220 after rebasing onto merged #222: commit 16ebb1e5a45cc5420abf90473b785d6c0b908a82, released Apache Flink 2.2.1, JDK 17, default compatibility settings. Each probe uses independent stock-Flink and native-enabled environments with bounded runtime DataStream input. Successful comparisons check output values and resolved result types before classifying execution.

Suggested priority: P3. This is a verified native support gap; the successful probes return equal results through explicit Flink fallback.

Keep these complex-type gaps behind the basic scalar work. SQL arms from released-compatible upstream CollectionFunctionsITCase/MapFunctionITCase were executed on runtime inputs. Successful fallback outputs were compared recursively, including arrays inside maps.

Runtime probe

Input type:

ROW<`f0` BOOLEAN, `f1` ARRAY<INT>, `f2` ARRAY<INT>, `f3` MAP<STRING, INT>, `f4` BINARY(2), `f5` BINARY(2)> NOT NULL

Input: [+I[true, [1, null, 3], [4, 5], {a=1, b=2}, [1, 2], [3, 4]]]

SELECT CAST(f1 AS ARRAY<BIGINT>) FROM src;

Flink result: [[[1, null, 3]]]; resolved type: [ARRAY<BIGINT>]. Native substitutions: 0.

Calc: unsupported CAST ARRAY→ARRAY

Runtime probe

Input type:

ROW<`f0` BOOLEAN, `f1` ARRAY<INT>, `f2` ARRAY<INT>, `f3` MAP<STRING, INT>, `f4` BINARY(2), `f5` BINARY(2)> NOT NULL

Input: [+I[true, [1, null, 3], [4, 5], {a=1, b=2}, [1, 2], [3, 4]]]

SELECT MAP_KEYS(f3) FROM src;

Flink result: [[[a, b]]]; resolved type: [ARRAY<STRING>]. Native substitutions: 0.

Calc: unsupported function/operator: MAP_KEYS

Runtime probe

Input type:

ROW<`f0` INT, `f1` BIGINT, `f2` DECIMAL(20, 2), `f3` STRING, `f4` BOOLEAN, `f5` TIME(3), `f6` BYTES, `f7` TIMESTAMP_LTZ(3)> NOT NULL

Input: [+I[1, 2, 3.25, abc, true, 12:34:56, [1, 2], 2024-02-29T12:34:56Z]]

SELECT ARRAY_AGG(f3) FROM src;

Flink result: [[[abc]]]; resolved type: [ARRAY<STRING>]. Native substitutions: 0.

GROUP BY: only SUM/MIN/MAX/COUNT/AVG/FIRST_VALUE/LAST_VALUE/SINGLE_VALUE aggregates

Verified remaining function families: ARRAY_APPEND, ARRAY_CONTAINS, ARRAY_POSITION, ARRAY_PREPEND, ARRAY_REMOVE, ARRAY_REVERSE, ARRAY_UNION, ARRAY_CONCAT, ARRAY_MAX, ARRAY_MIN, ARRAY_JOIN, ARRAY_SLICE, ARRAY_SORT, ARRAY_EXCEPT, ARRAY_INTERSECT, ELEMENT, MAP constructor, MAP_KEYS, MAP_VALUES, MAP_ENTRIES, MAP_FROM_ARRAYS and MAP_UNION.

Shipped and retained performance limits

Integer ARRAY_DISTINCT (TINYINT/SMALLINT/INT/BIGINT) shipped in PR #286, commit 27f1f7ad. Other element types remain pending; this issue does not claim the whole ARRAY_DISTINCT family is implemented.

MAP_KEYS/MAP_VALUES and integer ARRAY_MIN/ARRAY_MAX prototypes preserved parity but regressed whole-job performance. Production retains Flink; the rejected experiments and measurements are recorded in collection coverage. Revisit only with new performance evidence rather than repeating these prototypes.

Remaining casts and aggregates

Also verified: ARRAY to ARRAY CAST and TRY_CAST; dynamic/empty-separator SPLIT; COLLECT and ARRAY_AGG. Nested MAP constructor type gates must remain explicit. MAP_FROM_ENTRIES from the newer source checkout is not available in released Flink 2.2.1 and is not included. Completed dynamic ARRAY/MAP access (#90) and REGEXP_EXTRACT_ALL/STR_TO_MAP work (#219) do not imply support for these distinct functions.

Acceptance additionally requires nullable containers/elements/keys, duplicate handling, ordering, nested outputs, cardinality/error behavior, sliced arrays and schema preservation.

Acceptance

Preserve released Flink values, resolved types, NULL behavior and relevant error/short-circuit semantics. Add runtime-source SQL regressions that assert the specific native operator as well as result parity; do not use constant folding as proof of native support. Keep safe fallback for unverified forms and update the corresponding operator coverage documentation.