[EPIC] Implement Range Partitioning

[gene-bordegaray]

Is your feature request related to a problem or challenge?

Partitioning::Range provides a way to represent range-partitioned physical layouts, but the optimizer and planning currently have guards that drop / reject this partitioning.

That is because the initial PR was purely mechanical, but follow-up work should preserve range partitioning where possible.

This will serve as a parent issue as more composable issues are made.

Describe the solution you'd like

Make physical optimizer and planning code range partitioning aware and preserve its metadata.

Some places to cover:

• Implement Range Repartitioning #22397
• Projection pushdown through RepartitionExec
• Sort pushdown through RepartitionExec
• Join output partitioning when one or both inputs are range partitioned
• Dynamic filter pushdowns / routing

Describe alternatives you've considered

None.

Additional context

• Context: [DISCUSSION] Extending Partitioning to Support More Variants #21992

[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 declared file scan output 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