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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
