apache / apache/datafusion

Supporting analytics over large amounts of pre‑partitioned data

Open
#24,438 24 comments 0 reactions 1 assignee Assigned to @xavlee View on GitHub
enhancement
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)`.

Image

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(...)
```

Image

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.

Image

**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

Open the contributing guide

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.