Range partition on primitives. - #24598
Conversation
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## main #24598 +/- ##
==========================================
+ Coverage 81.43% 81.45% +0.01%
==========================================
Files 1118 1120 +2
Lines 399414 400805 +1391
Branches 399414 400805 +1391
==========================================
+ Hits 325278 326465 +1187
- Misses 55145 55241 +96
- Partials 18991 19099 +108 ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
5579f93 to
a6a8368
Compare
|
Nice results! |
gene-bordegaray
left a comment
gene-bordegaray
left a comment
There was a problem hiding this comment.
some optimization things and linear scan path that might be interesting 👍
|
here are my samplys:
You will see the cachine |
6f4a2df to
ccaab13
Compare
gene-bordegaray
left a comment
gene-bordegaray
left a comment
There was a problem hiding this comment.
i think there is some strctural things that could be improved with responsiblities. I prposed a rough idea. Let me know what you think @stuhood
| let data_types: Vec<DataType> = if !split_points.is_empty() { | ||
| (0..ordering.len()) | ||
| .map(|col_idx| split_points[0].values()[col_idx].data_type()) | ||
| .collect() | ||
| } else { | ||
| vec![] | ||
| }; |
There was a problem hiding this comment.
Should the key types come from the ordering exprs rather than split_points[0]? RowConverter needs an exact type match where compare_rows tolerated decimal precision / timestamp tz differences.
There was a problem hiding this comment.
AFAIK, we don't have the schema here. But after planning, split_points should already be aligned with the expected schema, so that should be fine...?
d87522a to
06849b5
Compare
6 participants
Which issue does this PR close?
Rationale for this change
In query plans that use range re-partitioning (such as matching the range partitioning of underlying base data, or evaluating dynamic range filters in hash joins), range re-partitioning can represent a significant fraction of total query execution time.
As detailed in the issue, the existing implementation in
RepartitionExec/BatchPartitionerandRangeExprassigns rows to partitions on a row-by-row basis usingextract_row_at_idx_to_bufand dynamicScalarValuecomparisons. This allocates per row and has dynamic dispatch overhead, making range re-partitioning 15x–30x slower than hash re-partitioning on standard integer keys.This PR adds a (more) vectorizable, zero-allocation implementation, bringing range re-partitioning to near-parity with hash repartitioning (10x–22x speedup on primitive numeric keys and 2x–3x speedup on string and composite keys).
What changes are included in this PR?
RangeRouterwith specialized routing paths:RowConverterfor strings, decimals, dictionary arrays, and composite keys.BatchPartitionerandRangeExprto delegate partition routing toRangeRouter.Benchmark Comparison (8,192 Rows)
mainbranchrange_expr_routing_i64(Isolated Primitive Routing)range_repartition_i64_uniform(End-to-EndBatchPartitioner)range_repartition_utf8_uniform(String Keys viaRowConverter)range_repartition_composite_i64(Composite(i64, i64)Keys)Are these changes tested?
Yes:
datafusion/physical-plan/src/repartition/range.rscovering ASC/DESC sort options,nulls_first/nulls_last, floats (with total order comparison), strings, and composite keys.datafusion/physical-plan/benches/range_repartition.rs.Are there any user-facing changes?
No.