Supporting analytics over large amounts of pre‑partitioned data

@alamb

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

take

@alamb

@alamb and @gene-bordegaray : This is the feature request we chatted earlier

@xavlee

@xavlee : feel free to take this ticket and use the branch or create a new branch

take

untake

test: add range-partitioned sorted time-bin aggregation coverage (apa…

Copy some comments from #24497

I see now that the idea here is that somehow the external system knows the data is not sorted but is non overlapping

I think this is going to be really hard to manage / ensure through the plan -- specifically I think the challenge will be to ensure that every operator properly reports if the property is preserved or not

I am not sure this is something we want to complicate datafusion with at this time as it

Is complicated (see above)
I am not sure how many other systems have data organized in this way / could take advantage of it

We are already working on Range Partitioning (which is another form of distribution aware planning) which is also complicated, but I think has a clearly demonstrated wide demand.

Let's see if we can find others who need this before we try and code it

@NGA-TRAN

I can / want to help support this effort (with #22497 as sent to the dev list). I know this is not another need since @NGA-TRAN and I are coming from the same place but would be more than willing to support reviews and dev work for this 🙇

@alamb

Thanks, @alamb for outlining the complication concerns and asking for concrete use cases.

Use case:
We’re entering an AI‑driven era with extremely high‑volume telemetry and analysis workloads. On the same dataset, we now face two extremes:

Interactive queries — require low latency. We partition data finely so queries run in parallel. Both DataFusion and Distributed DataFusion handle this well, and we scale perfectly.
Analytic queries — must finish within acceptable latency on very large datasets. Here, the number of partitions far exceeds the number of cores (even across many workers), so we must merge partitions. We’re fine with sequential execution within a merged partition as long as it completes in reasonable time. But merging drops the sort order, which causes two major issues:
i .Loss of streaming aggregation, hurting downstream efficiency and latency.
ii. For high‑cardinality workloads, hash tables become huge, increasing resource pressure and risking OOMs under concurrency.

We now see the core idea: the external system can indicate that data is unsorted but non‑overlapping. This is similar to other metadata properties (e.g., sort order) that come from the external system.

On complexity:
Yes, it’s complicated — and we’re happy to design and support this (thanks @gene-bordegaray and @xavlee and future Datadog contributors) with input from the DataFusion community. We see this as an opportunity to push DataFusion to the next level for AI workloads, with Datadog acting as a pioneer in providing real use cases.

Let us know your thoughts, and whether you’d like us to bring in other users of range partitioning to contribute.

We now see the core idea: the external system can indicate that data is unsorted but non‑overlapping. This is similar to other metadata properties (e.g., sort order) that come from the external system.

Yes, I think is likely a common pattern (and the use case you describe of "more partitions than cores" certainly happens for other users -- like we have it at Influx for example

Maybe that is the usecase we can highlight -- "supporting analytics over large amounts of pre-partitioned data" 🤔 It sounds pretty neat and not something that I am aware has been written about before

@alamb

@alamb and I just synced in person and agreed to move forward with the new GROUPED approach above.

@xavlee: would you be able to implement this on top of #24697?

FYI: @xudong963 and @gene-bordegaray

Thanks for the discussion. I'll update the design starting from #24697 with the GROUPED approach.

@gene-bordegaray

The stack has been reviewed and approved by @gene-bordegaray.

cc @xudong963 and @alamb for review.

test: cover unsorted contiguous groups in one partition (apache#24737)

refactor: use group clustering terminology in aggregate execution (ap…

For visibility: I opened #26370, extending #25150’s grouping optimization to temporal, decimal, and fixed-size binary keys. It now builds on #24697’s clustering refactor. Since it touches the same aggregation path, I wanted to flag it here and check for overlap with the ongoing work.

feat: add grouping equivalence properties (apache#24698)

feat: backport grouped aggregation support to DF55 staging

feat: stream aggregates over grouped input (apache#24497)

Navigation Menu

Repository navigation

Supporting analytics over large amounts of pre‑partitioned data #24438

Overview

Problem statement

Describe the solution you'd like

Describe alternatives you've considered

Additional context

14 remaining items

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Supporting analytics over large amounts of pre‑partitioned data #24438

Description

Overview

Problem statement

Describe the solution you'd like

Describe alternatives you've considered

Additional context

Activity

NGA-TRAN commented on Aug 17, 2026

NGA-TRAN commented on Aug 17, 2026

NGA-TRAN commented on Aug 17, 2026

xavlee commented on Aug 17, 2026

NGA-TRAN commented on Aug 17, 2026

alamb commented on Aug 25, 2026

gene-bordegaray commented on Aug 25, 2026

NGA-TRAN commented on Aug 25, 2026

alamb commented on Aug 26, 2026

14 remaining items

NGA-TRAN commented on Sep 3, 2026

xavlee commented on Sep 3, 2026

xavlee commented on Sep 23, 2026

geoffreyclaude commented on Oct 19, 2026

Metadata

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Issue actions