[EPIC] Implement Range Partitioning

[gene-bordegaray]

Problem

Partitioning::Range now exists as a physical partitioning model, but optimizer and planning code still need to preserve and use range metadata consistently. The goal of this epic is to make range-partitioned plans avoid unnecessary repartitioning while keeping multi-input correctness explicit.

Related discussion: #21992.

Primary implementation order

• Implement Range Repartitioning #22397
• Add range partitioning compatibility helpers #22589
• Add logical range partitioning representation #22778
• Support declared output partitioning for file/listing scans #22645
• Support planning logical range repartitioning #22786
• Support logical protobuf serialization for range repartitioning #22787
• Support physical execution of Range repartitioning #23137
• Replace / rename HashPartitioned distribution as KeyPartitioned #23236
• Allow Partitioning::Range to satisfy aggregation Distribution::KeyPartitioned requirements #23191
• Allow co-partitioned range inputs to satisfy inner partitioned hash joins #23183
• Allow Partitioning::Range to satisfy PartitionedTopK Distribution::KeyPartitioned requirements #23290
• Allow Partitioning::Range to satisfy window Distribution::KeyPartitioned requirements #23289
• Audit multi-child Distribution::KeyPartitioned requirements before general Range satisfaction #23451 , Allow co-partitioned Partitioning::Range inputs for sort merge joins #23478 , Allow co-partitioned Partitioning::Range inputs for symmetric hash joins #23479 (these were decided to be put into one PR)
• Allow co-partitioned Partitioning::Range inputs for left-side hash joins #23452
• Allow co-partitioned Partitioning::Range inputs for right-side hash joins #23453
• Allow co-partitioned Partitioning::Range inputs for full hash joins #23454
• Allow Partitioning::Range to satisfy Distribution::KeyPartitioned generally #23266
• Preserve Partitioning::Range output partitioning through joins #23450
• Support dynamic filters for range co-partitioned joins #23376
• Allow InterleaveExec to preserve Partitioning::Range #23455

Optimizer and integration follow-ups

These are related but do not need to block the primary join and distribution-satisfaction path.

• Complete physical execution planning of range partitioning #23230
• Represent Hive/value partitioning without claiming Partitioning::Hash #23436
• Expose Range Partitioning Across FFI Boundaries #22394
• Support Substrait exchange output for range repartitioning #22788
• Support functions date_bin and date_trunc in range partition satisfaction #23569
• Add config-matrix tests in enforce_distribution.rs for range-satisfaction settings #23572

[comphead]

Thanks @gene-bordegaray for driving this. Range Partition would help for ballista for distributed sorting

For reference Comet already supports RangePartitioning apache/datafusion-comet#1862

[gene-bordegaray]

Thanks @gene-bordegaray for driving this. Range Partition would help for ballista for distributed sorting

For reference Comet already supports RangePartitioning apache/datafusion-comet#1862

Yes, my pleasure. And thank you for the reference. I will check it out 😄

[stuhood]

@gene-bordegaray :

1. today's discussion (thanks!)
2. the design doc that I've been working on
3. reviewing Add ListingOptions::output_partitioning and FileScanConfig::output_partitioning for pre-defined file partitioning #22657

...have highlighted one of the tradeoffs of the current definition of Range partitioning that I wanted to call out:

---

As defined, Partitioning::Range contains no overlapping partitions. This is aligned with Partitioning::Hash (I can't really imagine what overlap would mean there), but allowing for overlapping output partitioning from a TableProvider, while continuing to require that input partitioning is total and non-overlapping (is this a enum Distribution...?) would have some really interesting properties.

In order to produce a reasonable number of Range partitions which match the other side of the join, the underlying data will either:

1. if DISJOINT - already need to be over-partitioned into a series of disjoint ranges, in which case the creator of the TableProvider would merge/aggregate a few of the underlying files/partitions into the declared Ranges (e.g. go from 1000 underlying files/partitions to 16 declared Partitioning::Range partitions)
2. if NOT-DISJOINT - if the underlying files/partitions are overlapping, then the TableProvider will need to:
   1. dynamically choose partition boundaries (using statistics about the amount of overlap, size, etc)
   2. bucket the files into those partitions
   3. push down or execute range filters for cases where a file spans a partition boundary (to guarantee that data is emitted in exactly one partition)

---

All of this is fine, and we're prepared to do it in our implementation. But if Partitioning::Range supported overlapping output partitions, things would look a little bit different:

In that case instead, a wrapper/sub-trait around the TableProvider might implement the steps from the NOT-DISJOINT case above: it would take the underlying overlapped partitions, consume their statistics (generically), and push down static per-file filters based on the chosen boundaries, and then coalesce the outputs of all files within a partition. Basically, it would generically implement the bucketing-into-partitions.

The benefit of this would be one upstream implementation of this range filtering... and possibly also some amount of sharing of the coordination required between the two sides of a join, which we have been planning to implement with an optimizer rule inserted at the appropriate spot to inspect both sides of the join, and align their Partitioning::Range declaration based on their shared statistics.

[gene-bordegaray]

I may not have time to respond today but am thinking about this and on my list to plan. Thank you @stuhood