[I] [FLINK] Implement streaming group aggregation natively [gluten]

["ggjh-159 (via GitHub)" <gi...@apache.org>]

ggjh-159 opened a new issue, #13417:
URL: https://github.com/apache/gluten/issues/13417

   ### Description
   
   ### Summary
   
   gluten-flink offloads streaming operators to Velox native execution, but unbounded group aggregation is not among them. This issue tracks implementing streaming group aggregation natively, with per-key accumulators living in the native keyed state backend.
   
   ### Current behavior
   
   - `StatefulPlanner::transformGroupAggregationOperator` throws `VELOX_NYI("GroupAggregation operator is not supported yet")`, so any Flink unbounded `GROUP BY` query fails at plan translation.
   - The alternative path constructs a `GroupAggregator` whose every method is an empty stub (`accumulate`, `retract`, `getValue` all no-op or return `nullptr`), flagged `TODO: not complete yet`. It cannot produce correct results.
   - Velox's baseline `avg` returns DOUBLE for every numeric input, which disagrees with Flink's `AvgAggFunction` and makes Flink `avg` queries fail the operator's result-type validation.
   
   ### Proposal
   
   Implement streaming group aggregation as a native operator over the keyed state backend, matching Flink's `GroupAggFunction` semantics:
   
   - **Per-key state.** Each distinct grouping key owns its accumulated state in the native keyed state backend, so results survive across input batches.
   - **Changelog emission.** `+I` for first-seen keys, an adjacent `-U`/`+U` pair for changed keys (a single `+U` when update-before is not requested), and nothing for unchanged keys; null grouping keys aggregate as a group of their own.
   - **FILTER clause.** `AGG(x) FILTER (WHERE cond)` accumulates each aggregate over its qualifying rows only; a key whose every row is filtered out still emits on first sight with a NULL aggregate value, matching Flink.
   - **`avg` with Flink semantics.** The baseline `avg` cannot serve Flink queries, so a separate `avg` family is provided for the FLINK preset, where Flink semantics take precedence. The supported input set follows Flink's `AvgAggFunction`: integral inputs keep the input result type with truncating integer division; REAL/DOUBLE inputs keep the input type; DECIMAL(p, s) produces DECIMAL(38, max(6, s)) with HALF_UP rounding. Any other input fails loudly at signature resolution, and accumulators survive snapshot/restore.
   - **Fail-loud boundaries.** Inputs and aggregate shapes outside the supported set are rejected with explicit errors rather than producing wrong values (see limitations).
   
   ### Scope and limitations (initial landing)
   
   - Append-only input only; retraction is rejected at construction.
   - Accumulate-mode aggregates only; DISTINCT aggregates and sorted aggregates are rejected.
   - `avg` supports the limited input set above (integral, REAL/DOUBLE, DECIMAL); other aggregate functions run when their Velox signature already matches the Flink result type, and any mismatch fails loudly at initialization instead of producing wrong values.
   
   ### Gluten version
   
   None


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org