andygrove opened a new issue, #4785:
URL: https://github.com/apache/datafusion-comet/issues/4785
## 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`).
--
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]
