Skip to content

Native GROUP BY: support LISTAGG and JSON string aggregates #232

Description

@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: P2. This is a verified native support gap; the successful probes return equal results through explicit Flink fallback.

LISTAGG, JSON_ARRAYAGG and JSON_OBJECTAGG have successful released-Flink baselines but remain outside the aggregate whitelist. Their output is STRING/VARCHAR, so prioritize them ahead of collection-returning aggregation.

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 LISTAGG(f3,',') FROM src;

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

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

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 JSON_OBJECTAGG(f3 VALUE f1) FROM src;

Flink result: [[{"abc":2}]]; resolved type: [VARCHAR(2000) NOT NULL]. Native substitutions: 0.

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

JSON_ARRAYAGG(f3) also returns ["abc"] through the same fallback. Cover LISTAGG separators, NULL/empty inputs, arrival ordering, duplicates, FILTER/retractions, and the released host restrictions on JSON aggregate changelogs. Preserve duplicate JSON-key errors and exact escaping. Do not count host-rejected aggregate/changelog forms as missing native support.

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.

Activity

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Metadata

Metadata

Assignees

No one assigned

    Labels

    No labels
    No labels

    Type

    No type

    Projects

    No projects

      Milestone

      No milestone

      Relationships

      None yet

      Development

      No branches or pull requests

      Issue actions