Blog post about Range Partitioning

[alamb]

Is your feature request related to a problem or challenge?

There is a lot of great work from @gene-bordegaray @jayshrivastava and others in

• [EPIC] Implement Range Partitioning #22395

As always, telling the community about it will both be good for others who have the same use case, as well as the authors to raise their visibility in the commonly

@gene-bordegaray contributed a section writing about this in the DataFusion 55 blog post (see details below)

• Add DataFusion 55.0.0 release blog post datafusion-site#203

However, the 55 release post is non ideal because:

1. There is limited space to explain the range partitioning ideas or sufficient backstory
2. The range partitioning isn't yet complete (e.g. there is no way to use it via SQL yet)

Describe the solution you'd like

Thus I suggest a complete self contained blog post about range partitioning

As always, I would structure this as a "informative post with low key DataFusion pitch" -- something like this outline

1. An understandable representative usecase (so readers can understand if it applies to them)
2. Background / technical intuition of how range partitioning works and why it is valuable (rather than e.g. hash partiitioning) as well as drawbacks
3. How to use it in DataFusion
4. A bit about how it is implemented (if relevant)

Describe alternatives you've considered

Here is the original text about range partitioning from the post (I need to trim it down for length)

Details

Support for Range Partitioning

Range partitioning assigns rows to partitions by ordered key ranges, so
partition 0 holds the lowest keys, partition 1 the next range, and so on. It is
the layout of time-series tables written one file per day or hour, and of tables
partitioned by ID range.

DataFusion 55 adds range partitioning ([#22395], design discussion
[#21992]). A range partitioning declares an ordering and a list of split
points. Partition i holds the keys that fall between split point i-1 and
split point i:

ordering     = [date ASC NULLS LAST]
split_points = [(2022-01-01), (2023-01-01)]

partition 0: date < 2022-01-01
partition 1: 2022-01-01 <= date < 2023-01-01
partition 2: date >= 2023-01-01

Compound keys work the same way, with split points compared lexicographically.
DataFusion does not validate the layout: a source that declares range
partitioning is responsible for placing every row in the partition its split
points describe. Wrong split points produce skew or missing join matches rather
than an error. The physical Partitioning::Range variant landed in [#22207],
the logical representation in [#22777], and execution plus planning in
[#23231] and [#23617].

Declaring a layout. There is no SQL syntax for this yet. A table declares
its partitioning through ListingOptions::with_output_partitioning or
FileScanConfig::with_output_partitioning ([#22657]):

let output_partitioning = Partitioning::Range(RangePartitioning::try_new(
    vec![col("range_key").sort(true, true)],
    vec![
        SplitPoint::new(vec![ScalarValue::Int32(Some(10))]),
        SplitPoint::new(vec![ScalarValue::Int32(Some(20))]),
        SplitPoint::new(vec![ScalarValue::Int32(Some(30))]),
    ],
)?);

let options = ListingOptions::new(Arc::new(ParquetFormat::default()))
    .with_output_partitioning(Some(output_partitioning));

What it buys you. The payoff is the RepartitionExec the planner no longer
inserts. A declared range layout now satisfies Distribution::KeyPartitioned
([#23680]), so aggregates ([#23239]), partitioned hash joins for inner
([#23184]), left ([#23487]), right ([#23484]), and full ([#23583]) types, sort
merge and symmetric hash joins ([#23480]), window functions ([#23416]),
and InterleaveExec ([#23623]) can all run
directly on the declared partitions:

> EXPLAIN SELECT range_key, SUM(value) FROM range_partitioned GROUP BY range_key;

AggregateExec: mode=SinglePartitioned, gby=[range_key@0 as range_key], aggr=[sum(range_partitioned.value)]
  DataSourceExec: output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=parquet

Joins need both sides on the same layout, not merely on some range layout:

> EXPLAIN SELECT l.range_key, l.value, r.value
  FROM range_partitioned l JOIN range_partitioned r ON l.range_key = r.range_key;

HashJoinExec: mode=Partitioned, join_type=Inner, on=[(range_key@0, range_key@0)]
  DataSourceExec: output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=parquet
  DataSourceExec: output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=parquet

Dynamic filter pushdown also now works for range-partitioned joins, routing
build-side filters to the correct probe partition using the range split points
([#23854]).

Thanks to [@gene-bordegaray], [@saadtajwar], [@peterxcli], [@stuhood],
[@gmhelmold], [@mattp5657], [@mithuncy], [@JSOD11], [@EdsonPetry],
[@Rich-T-kid], and [@blinding-pixels] for driving this substantial community
effort.

Additional context

No response

[gene-bordegaray]

I am in! i am gong to make an epic with follow up rane work which can be a part of this 🙇

[saadtajwar]

If we think a section regarding the logic of the physical execution of range partitioning could be helpful to add to the blog post, I'm happy to contribute a section for that :) #binary_search_finally_used_in_real_life

[stuhood]

If the timing works out, we'd be happy to contribute a blurb about how we're using it.

[peterxcli]

I can help to write or review the dynamic filter part!