Supporting analytics over large amounts of pre‑partitioned data
- Dominant language
- Rust
- Stars
- 9.3k
- Forks
- 2.4k
- Avg merge
- 3d 7h
- Merged PRs (30d)
- 344
Description
# Overview
After discussing with @alamb in the comments, I updated the subject to **supporting analytics over large amounts of pre‑partitioned data**. One concrete ask for that use case (from the original subject) is:
- Support streaming aggregates when partitions are unsorted but non‑overlapping streaming aggregates
@xavlee: If this becomes a larger epic, you may want to add well‑defined subtasks
# Problem statement
Our data is range‑partitioned on two dimensions, `time` and `key`, and each file is sorted by `(key, time)`.
This layout allows us to execute fully streaming the query below very efficiently, as shown in the plan below. Each of our partitions can be considered as one file-group and mapped directly to a DataFusion partition.
```SQL
SELECT key, date_bin(...), sum(...)
FROM my_table
GROUP BY key, date_bin(...)
```
The challenge arises when a query needs to scan many more data partitions than the number of CPU cores, which is also the default for `target_partitions`. Since we all know it’s not recommended to set `target_partitions` far above the CPU count, we’re forced to merge many of our data partitions into a single DataFusion partition. Once we do that, we lose the `(key, time)` sort order, which means `AggregateExec` can no longer stream the data.
# Describe the solution you'd like
Looking at the query plan below with the partitioning described above, we can see that even though each DataFusion partition (stream) is not sorted, the execution is still fully streaming. This works because the data across partitions does not overlap on the grouping keys (key, date_bin(..)).
If we introduce a new property that tells AggregateExec the input is non‑overlapping on the group‑by keys, then it can safely execute in a fully streaming fashion even without a global sort order.
We’ll handle the merging of many data partitions into a single DataFusion partition on our side and ensure the merged data remains non‑overlapping. All we need upstream is a property that can be propagated to AggregateExec to indicate this.
**Update:**
1. For the first solution, we may only need to propagate this property from the datasource through ProjectionExec (if necessary) and into the first AggregateExec. Whether it should propagate beyond that point requires more thought, but that’s outside the scope of this feature request. We can simply stop propagation there.
2. If it passes through ProjectionExec, we may initially support only monotonic functions such as date_bin, since that won’t affect correctness or behavior.
# Describe alternatives you've considered
_No response_
# Additional context
_No response_
Contributor guide
Assessment
This issue has not been assessed yet.