tests on range partitioned data

See

Line 120 in a6cff42

    
                     │       t1: DataSourceExec: file_groups={4 groups: [[/testdata/join/parquet/dim/d_dkey=B/data0.parquet], [/testdata/join/parquet/dim/d_dkey=D/data0.parquet], [], []]}, projection=[env, service, host, d_dkey], output_partitioning=Hash([d_dkey@3], 4), file_type=parquet, predicate=service@1 = log, pruning_predicate=service_null_count@2 != row_count@3 AND service_min@0 <= log AND log <= service_max@1, required_guarantees=[service in (log)]

It's not valid in datafusion to claim your data source is distributed by hash(expr), so this test isn't useful. We should migrate it to a distribution which maps to range partitioning.

There's several open questions that can only really be solved with tests

file_scan_config_scale_up_leaf_node in src/events/defaults/file_scan_config.rs if the file scan is claiming range partitioning and we round robin files, that can cause correctness issues right? See

datafusion-distributed/src/events/defaults/file_scan_config.rs

Line 61 in e818acb

rebalance_round_robin(all_partitioned_files, partition_count * ev.task_count)

Does NetworkShuffleExec work with Range partitioning? Can we add a test with a Range shuffle?

Notably, this errors:

datafusion-distributed/src/execution_plans/common.rs

Line 17 in a6cff42

pub(super) fn scale_partitioning(

Taking this. I'll migrate the join test off the invalid hash(expr) data-source distribution and add coverage for range-partitioned sources (file-scan scale-up, target_partitions vs range partition count, and scale_partitioning).

@shinzoxD

@shinzoxD thanks! You may want to base on top of #540. I'll upgrade and merge that once the release is finalized: apache/datafusion#22393

upgrade to datafusion 55 (#540)

upgrade to datafusion 55 (datafusion-contrib#540)

I have started looking at this one in the context of paradedb#68

take

upgrade to datafusion 55 (datafusion-contrib#540)

@Rich-T-kid

@Rich-T-kid are you planning on working on this? if not, we can free this up for somebody else.

no I don't think I can work on this in the near future.

tests: migrate test on pre-partitioned data to use range partitioning (…

I played around with this here

Original Data Partitions

1. Split up the ranges across tasks

I think this might just work out of the box 🤔... not 100% sure. Range-Range shuffles are super rare. They only occur when there's a join on two sets of differently Range-partitioned data. We need to support it nonetheless.

Range-Hash shuffles will work just fine I think. They should use the existing network shuffle algo.

2. Use the same Ranges across tasks

Honestly, this is a non starter. I doubt any user wants overlapping partitions in multiple leaf tasks. Feel free to push back if I'm wrong.

Thanks for looking at this!

Range-Range shuffles are super rare. They only occur when there's a join on two sets of differently Range-partitioned data. We need to support it nonetheless.

We use these quite a bit! See the testcases I added in #734 and fixed in #730.

Based roughly atop those two branches (sorry that I haven't updated them in a long while: have been busy!), ParadeDB 0.26.x gets these types of plans (search for DataFusion Physical Plan in page): https://github.com/paradedb/paradedb/blob/9aa24da283d6532d0c9c47f846d72894a6a25d4b/pg_search/tests/pg_regress/expected/mpp_range_partitioning.out

We use these quite a bit! See the testcases I added in #734 and fixed in #730.

Oh interesting! Sorry I missed these. I can take a look tomorrow

I took a look at your PRs. I think we've converged on the same upstream limitations, namely the one I outlined in apache/datafusion#26509. I'm going to try to put up a PR for that.

In #730, you take this approach, opting for 1 partition per worker.

Partitioning::Range(4, split points: [10, 20, 30]) 

Split into 4 workers
Partitioning::Unknown(1) // implicitly Range(x < 10)
Partitioning::Unknown(1) // implicitly Range(10 <= x < 20)
Partitioning::Unknown(1) // implicitly Range(20 <= x < 30)
Partitioning::Unknown(1) // implicitly Range(x >= 40)

Pros:

Partitioning across workers is well defined

Cons:

Only 1 partition per worker, meaning you may use too many workers

In #782, I opted for this

Partitioning::Range(4, split points: [10, 20, 30]) 

Split into 2 workers, assuming target_partitions=2
Partitioning::Range(2) // implicitly Range(2, split points: [10])
Partitioning::Range(2) // implicitly Range(2, split points: [30])

Pros:

Support parallelism within workers

Cons:

Partitions are poorly defined. Task 1 has ranges x < 10 and x >=10. Task 2 has ranges x < 30 and x >= 30. These are overlapping partitions. When we coalese them, we cannot recreate the original range partitioning because information was lost. We could store the missing split point out of band, but this is a bit hacky IMO. The other problem would be that in range-range shuffles, I cannot intelligently route certain input partitions to certain output partitions (ie. reducing the number of connections / size of an N*N shuffles)

In apache/datafusion#26509, I propose that we should have end bounds for range partitions (even just MIN_USIZE and MAX_USIZE) so we can preserve the most information.

That way, we can support something like this

Partitioning::Range(4, split points: [10, 20, 30]) 

Split into 2 workers, assuming target_partitions=2
Partitioning::Range(2) // implicitly Range(2, ranges: [x < 10, 10 <= x < 20)
Partitioning::Range(2) // implicitly Range(2, ranges: [20 <= x < 30, x >= 30)

Now, during a coalesce, you can recreate the original partitioning. During a shuffle you know exactly what input ranges need to be routed to what output ranges.

@jayshrivastava

Thanks for writing this up @jayshrivastava!

One related alternative/workaround worth considering is scaling up leaf scans the same way #782 handles Partitioning::Hash: give every worker task the full, unmodified Partitioning::Range(N, split_points) with N FileGroups, where only the slice of FileGroups assigned to that worker is non-empty and the remaining FileGroups are empty:

Partitioning::Range(4, split_points: [10, 20, 30])

Split into 2 workers:
  Task 0: Range(4, split_points: [10, 20, 30])  file_groups: [A, B, [], []]
  Task 1: Range(4, split_points: [10, 20, 30])  file_groups: [[], [], C, D]

DataFusion operators (mostly?) already handle empty partitions cleanly:

No split points are lost between workers, and uneven partition_count / task_count divisions don't need duplicate split points.
Task-local co-partitioned joins and dynamic filter routing work out of the box because both sides of the join on a worker still have identical Range(N, split_points).
More generally: it's a case of skew, and so over time statistics will likely improve to handle empty partitions in more cases.

The main tradeoff with keeping all N partitions per worker is that each worker instantiates N partition streams (with N - N/W of them immediately returning None), and NetworkCoalesceExec currently concatenates all W * N streams unless taught to collapse/route disjoint partitions.

Also, your comment on apache/datafusion#26509 (comment) about supporting empty partitions seems directly aligned with this: if empty partitions are supported (via metadata, duplicate split points, or empty groups), they can also act as the lower/upper caps on a worker's active range without necessarily needing separate min/max bound fields.

@stuhood

@stuhood That's a good approach too that I considered as well. One concrete thing that may be a annoying is that we might have Partitioning::Range(100) and padding that many file groups in each worker is a lot of padding. We also might have to pad on either side of a Range-Range shuffle, which I think will require a lot of extra code paths in DFD.

I have a hunch that splitting up the ranges across workers will be more maintainable/clean and easier to reason about, but it requires 1 upstream change so it might take a bit longer to land. I published apache/datafusion#26593 to add bounds upstream. I'm hoping that gets in to df56 and we can move relatively fast here :)

@stuhood That's a good approach too that I considered as well. One concrete thing that may be a annoying is that we might have Partitioning::Range(100) and padding that many file groups in each worker is a lot of padding. We also might have to pad on either side of a Range-Range shuffle, which I think will require a lot of extra code paths in DFD.

Hm. Is there reason to believe that if this hasn't been a problem for Partitioning::Hash that it would be a problem for Partitioning::Range?

I can maybe see the argument for apache/datafusion#26593, but it does feel like it would add complexity. Might there be a way to share the "pruned"/"empty" partition logic across all Partitioning types, so that Partitioning::Hash could also declare empty partitions?

For all data sources, this repo has the invariant that all task variants have the same number of partitions. For Unknown partitioning, we pad empty file groups. For Range, we can either drop the invariant or find a workaround. In #782, I use a workaround mentioned in the PR description. Hash partitioning a data source is not supported.

Navigation Menu

Repository navigation

tests on range partitioned data #628

Original Data Partitions

1. Split up the ranges across tasks

2. Use the same Ranges across tasks

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

tests on range partitioned data #628

Description

Activity

shinzoxD commented on Aug 15, 2026

jayshrivastava commented on Aug 17, 2026

stuhood commented on Sep 2, 2026

Rich-T-kid commented on Sep 4, 2026

gabotechs commented on Oct 2, 2026

Rich-T-kid commented on Oct 2, 2026

jayshrivastava commented on Oct 19, 2026

Original Data Partitions

1. Split up the ranges across tasks

2. Use the same Ranges across tasks

stuhood commented on Oct 19, 2026

jayshrivastava commented on Oct 19, 2026

jayshrivastava commented on Oct 22, 2026

stuhood commented on Oct 22, 2026

jayshrivastava commented on Oct 23, 2026

stuhood commented on Oct 23, 2026

jayshrivastava commented on Oct 25, 2026

Metadata

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Issue actions