Native JVM functions: track table, aggregate and nested scalar result gaps

[liuyongvs]

Verified at PR #220 commit 16ebb1e5a45cc5420abf90473b785d6c0b908a82 on released Flink 2.2.1/JDK 17 by rerunning the checked-in FlinkPortableSqlAuditTest. These existing portable fixtures provide reproducible input data and function registrations; every case below passed value/changelog parity through explicit fallback with zero native substitutions.

Suggested priority: P2 for table/aggregate functions with basic STRING/INT/BIGINT values; P3 for nested ROW-returning scalar functions. This tracks native support, separately from the completed test-fixture work in #106.

aggregate-function

SELECT k, test_aggregate(v) FROM test_input GROUP BY k;

Observed fallback:

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

Both executions completed with 8 output changelog records. The fixture asserts materialized values and function lifecycle.

cdc-aggregate-function

SELECT k, test_aggregate(v) FROM cdc_input GROUP BY k;

Observed fallback:

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

Both executions completed with 12 output changelog records. The fixture asserts materialized values and function lifecycle.

nested-scalar

SELECT id, test_nested(nested) FROM nested_input;

Observed fallback:

Calc: UDF return type not native: ROW

Both executions completed with 3 output changelog records. The fixture asserts materialized values and function lifecycle.

table-function

SELECT s.id, t.v, t.pos FROM test_input s, LATERAL TABLE(test_split(s.text_value)) AS t(v, pos);

Observed fallback:

correlate: only an INNER/LEFT UNNEST of a single ARRAY (scalar or ROW element), MAP, or MULTISET column, optionally WITH ORDINALITY, is supported — a non-encodable condition or a condition over a LEFT unnest fall back

Both executions completed with 7 output changelog records. The fixture asserts materialized values and function lifecycle.

Acceptance

• Support LATERAL TABLE through a verified columnar operator boundary, including NULL/empty outputs, inner/left semantics, pushed predicates and changelog propagation. Plain UNNEST support does not cover arbitrary table functions.
• Admit supported JVM aggregate functions only with verified accumulate/retract/merge/reset, NULL and state/checkpoint contracts. The CDC fixture must retain its known final BIGINT sums.
• Track nested scalar UDF input/output and lifetime/ownership support separately, after scalar-result work.
• Validate open/close lifecycle and error paths and keep explicit fallback for unsupported function kinds.
• Update docs/operators/unnest-correlate.md, group-by.md, and calc-filter.md in the implementation commits.