apache / apache/datafusion-comet

Support native batch session_window grouping (MergingSessionsExec / UpdatingSessionsExec)

Open
#4,785 0 comments 0 reactions 0 assignees View on GitHub
area:aggregation enhancement priority:low
Dominant language
Scala
Stars
1.3k
Forks
373
Avg merge
2d 4h
Merged PRs (30d)
198

Description

## Is your feature request related to a problem or challenge?

Comet now supports native batch time-window grouping via `window()` / `window_time()` (see #4553 and PR #4784, which added `PreciseTimestampConversion` and `KnownNullable`). Batch `session_window()` grouping is the remaining time-window construct and still falls back to Spark.

To be clear, this is about **regular batch** `session_window()` in DataFrame/SQL, not Structured Streaming. The plan for a batch query confirms this:

```
IS_STREAMING=false
Sort
+- HashAggregate(keys=[session_window], functions=[count(1)]) -- final
+- MergingSessions ..., false, [session_window], ... -- `false` = isStreaming=false
+- Sort [session_window ASC]
+- CometColumnarExchange SinglePartition
+- HashAggregate(keys=[session_window], functions=[partial_count(1)]) -- partial
+- Project [named_struct(start, precisetimestampconversion(...),
end, knownnullable(precisetimestampconversion(...)))]
+- CometFilter isnotnull(ts)
+- CometNativeScan parquet ...
```

There is no state store, watermark, or streaming operator involved (those only appear on the streaming path in `AggUtils`, and streaming plans never reach Comet). The scan, filter, and the session-struct projection already run natively; the gap is the session-merging operator.

## Describe the potential solution

Implement native support for Spark's batch session-window operators so the aggregation stack around `session_window()` runs natively instead of falling back:

- `MergingSessionsExec` (with `isStreaming = false`) — appears in the sort-based batch path shown above.
- `UpdatingSessionsExec` (with `isStreaming = false`) — appears in other batch shapes (see `AggUtils.mayAppendUpdatingSessionExec`).

Both are sort-based operators that assign/merge overlapping session windows for adjacent rows within a group and feed the surrounding partial/final aggregates. Neither has a DataFusion equivalent, so this is a from-scratch native operator (Rust operator + serde + `operator.proto` message + `planner.rs` wiring + child distribution/ordering requirements), registered in `CometExecRule`'s `nativeExecs` map. This is substantially larger than the expression PRs under #4553, which is why it is split out here.

Streaming session windows (`SessionWindowStateStoreSaveExec` / `RestoreExec`) remain out of scope — Comet targets batch execution.

## Additional context

- Umbrella: #4553 (time-window grouping)
- Depends on: #4784 (adds `PreciseTimestampConversion` + `KnownNullable`, which the session-struct projection needs to run natively)
- Spark references: `org.apache.spark.sql.execution.aggregate.{UpdatingSessionsExec, MergingSessionsExec}`, `AggUtils.planAggregateWithoutDistinct` / `mayAppendUpdatingSessionExec`, and `ResolveTimeWindows` (`SessionWindowing`).

Contributor guide

Open the contributing guide

Research direction

Start with Spark's MergingSessionsExec and UpdatingSessionsExec references in AggUtils, then trace the native execution wiring in planner.rs and CometExecRule's nativeExecs map. Read operator.proto and the child distribution/ordering requirements before designing the Rust operator and serde path. Done means regular batch session_window() plans execute natively for both operators, while streaming session windows remain unsupported.

Written by the indexing model from the issue text.

Assessment

Tech stack
rust, scala
Domain
backend, data-engineering
Issue type
Feature
Difficulty
5/5
Estimated time
Over a week
Activity status
Quiet
Clarity
Mostly clear
Newbie friendliness
35/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.