apache / apache/gluten

[FLINK] Support window aggregate for nexmark `q12`

Open
#10,246 0 comments 0 reactions 0 assignees View on GitHub
enhancement
Dominant language
Scala
Stars
1.6k
Forks
657
Avg merge
2d 21h
Merged PRs (30d)
85

Description

### Description
We need to support different kind of window aggregate: processing-time window agg, event-time window agg, and we can use nexmark q12 to verfiy.
- processing-time window aggregate
The original `q12` sql is a process-time window aggregate, its execution pipeline as below:
Image.

1. Offload `Kafka->calc`, and `WindowAggregate -> nexmarkq12` to velox, map `WindowAggregate` to velox's `StreamingAggregate` with `SINGLE` aggregate step;
2. The data shuffle between opeator chains still use flink original mechanism;
3. the state of window aggregate would use the heap state and rocksdb state that have been implemented in velox.
all these features implemented by pr: https://github.com/apache/gluten/pull/10898

- event-time window aggregate
we change the q12 sql, change the view definition as
```
CREATE VIEW B AS SELECT *, `dateTime` as p_time FROM bid;
```

then it becomes a event-time window aggregate, its execution pipeline

Image.
1. Offload `kafka->calc->localAgg`, `global->agg` to velox, `localAgg` map to velox `streamingAgg` with `Partial` step, `globalAgg` map to velox with `immediate` and `final` step; then we map flink's `local->merge->global` aggregate into velox `part->imediate->final` agg;
2. still use shuffle of original flink, velox calculate the key of streamRecord, and flink determine by this key which next operator its will shuffle to;
3. still use heap/ rocksdb state backend to store the state data of agg;

### Gluten version

None

Contributor guide

Open the contributing guide

Research direction

Start with Nexmark q12 and review pull request 10898, then trace the processing-time and event-time window-aggregate pipelines described in the issue. Done means q12 supports both modes with the specified Velox aggregate mappings, Flink shuffle behavior, and heap or RocksDB state.

Written by the indexing model from the issue text.

Assessment

Tech stack
kafka, sql
Domain
data-engineering, stream-processing
Issue type
Feature
Difficulty
5/5
Estimated time
Over a week
Activity status
Stale
Clarity
Mostly clear
Newbie friendliness
30/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.