ClickHouse / ClickHouse/ClickHouse
Support shipping window (partitioned) sorts with plan-based parallel replicas
- Dominant language
- C++
- Stars
- 49.9k
- Forks
- 9k
- Avg merge
- 21h 32m
- Merged PRs (30d)
- 515
Description
Plan-based parallel replicas (`parallel_replicas_plan_based`) lifts the split marker through `SortingStep`: the sort is cloned into the fragment shipped to the replicas and the initiator keeps a `MergingSorted` step above the union, so each replica sorts its own rows and the initiator only merges.
A window function's pre-sort is also a `SortingStep`, but a special one: it is built with the partitioning constructor (`Planner::addWindowSteps`), so `partition_by_description` is set. Its contract is *one stream per `PARTITION BY` group*, not one sorted stream - `SortingStep::fullSort` scatters by the partition keys and deliberately skips the final merge, and `WindowStep` then runs one `WindowTransform` per stream.
That contract cannot be expressed by the `MergingSorted` step the pass puts on the initiator, so such a sort is currently rejected in `sortingCanBeShipped` (`src/Processors/QueryPlan/Optimizations/applyParallelReplicas.cpp`) and window queries keep the split below the sort, the same as with classic parallel replicas.
Shipping it collapsed both the sort and the window to a single stream. With `max_threads = 4` and a three-replica cluster:
```sql
SELECT p, row_number() OVER (PARTITION BY p ORDER BY a) FROM t_win;
```
`parallel_replicas_plan_based = 0` (sort stays on the initiator):
```
(Window)
Resize 4 → 4
WindowTransform × 4
(Sorting)
MergeSortingTransform × 4
LimitsCheckingTransform × 4
PartialSortingTransform × 4
Resize × 4 3 → 1
ScatterByPartitionTransform × 3 1 → 4
```
`parallel_replicas_plan_based = 1` with the sort shipped:
```
(Window)
Resize 1 → 4
WindowTransform
(Sorting)
MergingSortedTransform 3 → 1
```
Results stayed correct, because `full_sort_description` is `PARTITION BY` followed by `ORDER BY`, so a single globally merged stream still holds each partition contiguously and in order. What is lost is the per-partition parallelism of both the sort and the window.
Two ways to support shipping it:
1. Clone the sort into the fragment *without* `partition_by_description`. Each replica then returns one stream sorted by `full_sort_description` and the initiator's merge produces a valid input for a single `WindowTransform`. The sorting work is distributed; the window itself still runs on one stream (`query_plan_enable_multithreading_after_window_functions` restores parallelism after it).
2. Preserve the per-partition scope on the initiator - re-scatter above the merge, or rebuild the parent as a partitioned sort. Closer to the current pipeline, but the replica-side scatter produces mutually unordered streams, so the merge contract has to be reworked.
Note that read-in-order gives no reason to prefer shipping today: `optimizeReadInOrder` early-returns on a partitioned sort unless `query_plan_reuse_storage_ordering_for_window_functions` is enabled, and that setting defaults to `false`.
Related: https://github.com/ClickHouse/ClickHouse/pull/114315
Contributor guide
Assessment
This issue has not been assessed yet.