Range partition on primitives.

[stuhood]

Which issue does this PR close?

• Closes Improve the performance of range re-partitioning #24597.
• Related to [EPIC] Implement Range Partitioning #22395.

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 / BatchPartitioner and RangeExpr assigns rows to partitions on a row-by-row basis using extract_row_at_idx_to_buf and dynamic ScalarValue comparisons. 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 zero-allocation, vectorizable implementation, achieving 10x–16x speedup on primitive numeric keys and 2x–3x speedup on string and composite keys.

What changes are included in this PR?

• Adds an internal RangeRouter with specialized routing paths:
  • A zero-allocation path for single primitive numeric, temporal, and float columns that binary searches directly over Arrow buffer slices.
  • A (more) vectorizable path using Arrow's RowConverter storing flat Rows buffers for strings, decimals, dictionary arrays, and composite keys.
• Updates RepartitionExec to construct and share an Arc<RangeRouter> across input partitions, avoiding redundant split-point encoding.
• Updates BatchPartitioner and RangeExpr to delegate partition routing to RangeRouter.
• Adds BatchPartitioner::try_new_range_partitioner and deprecates BatchPartitioner::new_range_partitioner.
• Adds a Criterion suite.

Benchmark Results (vs main baseline)

Benchmark	Partitions	Baseline (main)	This PR	Speedup
range_repartition_i64_uniform	8	275.3 µs (29.8 Melem/s)	23.4 µs (349.5 Melem/s)	11.7x
	64	459.8 µs (17.8 Melem/s)	40.9 µs (200.4 Melem/s)	11.2x
	512	1,194.6 µs (6.86 Melem/s)	89.9 µs (91.2 Melem/s)	12.2x
range_repartition_i64_sequential	8	291.8 µs (28.1 Melem/s)	23.2 µs (353.6 Melem/s)	12.6x
	64	478.6 µs (17.1 Melem/s)	38.5 µs (212.9 Melem/s)	12.4x
	512	989.9 µs (8.28 Melem/s)	83.9 µs (97.6 Melem/s)	11.8x
range_repartition_utf8_uniform	8	543.5 µs (15.1 Melem/s)	184.2 µs (44.5 Melem/s)	3.0x
	64	964.7 µs (8.49 Melem/s)	330.5 µs (24.8 Melem/s)	2.9x
	512	1,333.2 µs (6.14 Melem/s)	551.7 µs (14.8 Melem/s)	2.4x
range_repartition_composite_i64	8	344.2 µs (23.8 Melem/s)	114.3 µs (71.7 Melem/s)	3.0x
	64	516.9 µs (15.8 Melem/s)	249.7 µs (32.8 Melem/s)	2.1x
	512	985.3 µs (8.31 Melem/s)	473.6 µs (17.3 Melem/s)	2.1x

Are these changes tested?

Added unit and integration tests.

Are there any user-facing changes?

• Added BatchPartitioner::try_new_range_partitioner(&RangePartitioning, Time) -> Result<Self>.
• Deprecated BatchPartitioner::new_range_partitioner in favor of fallible try_new_range_partitioner.

[codecov-commenter]

Codecov Report

❌ Patch coverage is 81.90743% with 129 lines in your changes missing coverage. Please review.
✅ Project coverage is 81.57%. Comparing base (a274959) to head (612381c).

Files with missing lines	Patch %	Lines
datafusion/physical-plan/src/repartition/range.rs	84.21%	57 Missing and 30 partials ⚠️
datafusion/physical-plan/src/repartition/mod.rs	74.07%	32 Missing and 10 partials ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main   #24598      +/-   ##
==========================================
- Coverage   81.58%   81.57%   -0.01%     
==========================================
  Files        1123     1124       +1     
  Lines      406610   407237     +627     
  Branches   406610   407237     +627     
==========================================
+ Hits       331719   332206     +487     
- Misses      55453    55552      +99     
- Partials    19438    19479      +41     

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

[Dandandan]

Nice results!
I think two easy further optimizations could be using extend and perhaps further optimizing the rowconverter path.
Ok to postpone it to a follow-up PR however you like!

[Dandandan]

2 efficiency comments

[gene-bordegaray]

some optimization things and linear scan path that might be interesting 👍

[gene-bordegaray]

here are my samplys:

• No Linear Scan and no cache RowConverter: https://profiler.firefox.com/from-url/https%3A%2F%2Fgist.githubusercontent.com%2Fgene-bordegaray%2F87a66e96650dd9ec88dc5dac66c549db%2Fraw%2Fprofile_base_baked.json/
• Linear Scan and cache RowConverter in Arc: https://profiler.firefox.com/from-url/https%3A%2F%2Fgist.githubusercontent.com%2Fgene-bordegaray%2Fdccf6a47d4cc100fe5d22197c7513643%2Fraw%2Fprofile_opt_baked.json/

You will see the cachine RowConverter gets rid of the drop_glue::RowConverter then the partition_grouped_take takes over as dominant

[gene-bordegaray]

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

[gene-bordegaray]

starting to get there. I think this is last major round of comments

[gene-bordegaray]

oops did not mean to approve

[stuhood]

Split out the benchmarks into the bottom commit, and applied review feedback in the top commit.

The initial benchmark results were a bit optimistic (they never regressed during review feedback: they were just a bit off when they were initially posted). Now updated.

Thanks for the feedback!

[github-actions]

Thank you for opening this pull request!

Reviewer note: cargo-semver-checks reported the current version number is not SemVer-compatible with the changes in this pull request (compared against the base branch).

Details

     Cloning apache/main
    Building datafusion-physical-plan v55.0.0 (current)
       Built [  41.251s] (current)
     Parsing datafusion-physical-plan v55.0.0 (current)
      Parsed [   0.130s] (current)
    Building datafusion-physical-plan v55.0.0 (baseline)
       Built [  33.886s] (baseline)
     Parsing datafusion-physical-plan v55.0.0 (baseline)
      Parsed [   0.127s] (baseline)
    Checking datafusion-physical-plan v55.0.0 -> v55.0.0 (no change; assume patch)
     Checked [   0.879s] 223 checks: 222 pass, 1 fail, 0 warn, 31 skip

--- failure type_method_marked_deprecated: type method #[deprecated] added ---

Description:
A type method is now #[deprecated]. Downstream crates will get a compiler warning when using this method.
        ref: https://doc.rust-lang.org/reference/attributes/diagnostics.html#the-deprecated-attribute
       impl: https://github.com/obi1kenobi/cargo-semver-checks/tree/v0.50.0/src/lints/type_method_marked_deprecated.ron

Failed in:
  method datafusion_physical_plan::repartition::BatchPartitioner::new_range_partitioner in /home/runner/work/datafusion/datafusion/datafusion/physical-plan/src/repartition/mod.rs:1013

     Summary semver requires new minor version: 0 major and 1 minor checks failed
    Finished [  77.791s] datafusion-physical-plan

[stuhood]

A type method is now #[deprecated]. Downstream crates will get a compiler warning when using this method.

Hm... is this not allowed for commits landing on main? Or is this not a blocking comment.