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.
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:
- 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.
- 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
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:
@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,
timeandkey, 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.
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 settarget_partitionsfar 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 meansAggregateExeccan 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:
Describe alternatives you've considered
No response
Additional context
No response