perf: Bring over apache/arrow-rs/9683 radix sort, integrate into ExternalSorter

[mbutrovich]

Which issue does this PR close?

N/A.

Rationale for this change

MSD radix sort on row-encoded keys is 2-3x faster than lexsort_to_indices for most multi-column sorts. See apache/arrow-rs#9683 for benchmarks and design rationale.

The arrow-rs kernel hasn't been released yet, so this PR temporarily hoists the implementation into DataFusion. Once arrow-rs ships it, we replace the local copy with the import.

What changes are included in this PR?

• sorts/radix.rs (new): MSD radix sort kernel + RowConverter wiring, copied from feat(arrow-row): add MSD radix sort kernel for row-encoded keys arrow-rs#9683.
• sorts/sort.rs: sort_batch branches to radix sort when fetch is not set and column types are favorable. ExternalSorter precomputes the radix-vs-lexsort decision once at construction rather than per-batch.
• sorts/stream.rs: IncrementalSortIterator accepts the precomputed use_radix flag.

The heuristic falls back to lexsort_to_indices when:

• fetch is set (partial sort / top-N)
• All sort columns are dictionary-typed (long shared row prefixes)
• Any sort column is a nested type (lists, structs — encoding cost outweighs benefit)

Are these changes tested?

4 new tests:

• Multi-column integer sort verifying radix path correctness
• Heuristic unit test (primitives, dictionaries, lists, mixed)
• Nulls with descending + nulls_first sort options
• 50-iteration fuzz test cross-validating radix output against lexsort

The kernel itself has 17 tests in the arrow-rs PR. The 188 existing sort tests in datafusion-physical-plan continue to pass.

Are there any user-facing changes?

No.

[mbutrovich]

run benchmark sort

[adriangbot]

🤖 Criterion benchmark running (GKE) | trigger
Instance: c4a-highmem-16 (12 vCPU / 65 GiB) | Linux bench-c4217435322-1038-mzpkf 6.12.55+ #1 SMP Sun Feb 1 08:59:41 UTC 2026 aarch64 GNU/Linux

CPU Details (lscpu)

Architecture:                            aarch64
CPU op-mode(s):                          64-bit
Byte Order:                              Little Endian
CPU(s):                                  16
On-line CPU(s) list:                     0-15
Vendor ID:                               ARM
Model name:                              Neoverse-V2
Model:                                   1
Thread(s) per core:                      1
Core(s) per cluster:                     16
Socket(s):                               -
Cluster(s):                              1
Stepping:                                r0p1
BogoMIPS:                                2000.00
Flags:                                   fp asimd evtstrm aes pmull sha1 sha2 crc32 atomics fphp asimdhp cpuid asimdrdm jscvt fcma lrcpc dcpop sha3 sm3 sm4 asimddp sha512 sve asimdfhm dit uscat ilrcpc flagm sb paca pacg dcpodp sve2 sveaes svepmull svebitperm svesha3 svesm4 flagm2 frint svei8mm svebf16 i8mm bf16 dgh rng bti
L1d cache:                               1 MiB (16 instances)
L1i cache:                               1 MiB (16 instances)
L2 cache:                                32 MiB (16 instances)
L3 cache:                                80 MiB (1 instance)
NUMA node(s):                            1
NUMA node0 CPU(s):                       0-15
Vulnerability Gather data sampling:      Not affected
Vulnerability Indirect target selection: Not affected
Vulnerability Itlb multihit:             Not affected
Vulnerability L1tf:                      Not affected
Vulnerability Mds:                       Not affected
Vulnerability Meltdown:                  Not affected
Vulnerability Mmio stale data:           Not affected
Vulnerability Reg file data sampling:    Not affected
Vulnerability Retbleed:                  Not affected
Vulnerability Spec rstack overflow:      Not affected
Vulnerability Spec store bypass:         Mitigation; Speculative Store Bypass disabled via prctl
Vulnerability Spectre v1:                Mitigation; __user pointer sanitization
Vulnerability Spectre v2:                Mitigation; CSV2, BHB
Vulnerability Srbds:                     Not affected
Vulnerability Tsa:                       Not affected
Vulnerability Tsx async abort:           Not affected
Vulnerability Vmscape:                   Not affected

Comparing arrow_rs_9683 (e1e1d09) to 8d91fb0 (merge-base) diff
BENCH_NAME=sort
BENCH_COMMAND=cargo bench --features=parquet --bench sort
BENCH_FILTER=
Results will be posted here when complete

---

File an issue against this benchmark runner

[adriangbot]

🤖 Criterion benchmark completed (GKE) | trigger

Instance: c4a-highmem-16 (12 vCPU / 65 GiB)

CPU Details (lscpu)

Architecture:                            aarch64
CPU op-mode(s):                          64-bit
Byte Order:                              Little Endian
CPU(s):                                  16
On-line CPU(s) list:                     0-15
Vendor ID:                               ARM
Model name:                              Neoverse-V2
Model:                                   1
Thread(s) per core:                      1
Core(s) per cluster:                     16
Socket(s):                               -
Cluster(s):                              1
Stepping:                                r0p1
BogoMIPS:                                2000.00
Flags:                                   fp asimd evtstrm aes pmull sha1 sha2 crc32 atomics fphp asimdhp cpuid asimdrdm jscvt fcma lrcpc dcpop sha3 sm3 sm4 asimddp sha512 sve asimdfhm dit uscat ilrcpc flagm sb paca pacg dcpodp sve2 sveaes svepmull svebitperm svesha3 svesm4 flagm2 frint svei8mm svebf16 i8mm bf16 dgh rng bti
L1d cache:                               1 MiB (16 instances)
L1i cache:                               1 MiB (16 instances)
L2 cache:                                32 MiB (16 instances)
L3 cache:                                80 MiB (1 instance)
NUMA node(s):                            1
NUMA node0 CPU(s):                       0-15
Vulnerability Gather data sampling:      Not affected
Vulnerability Indirect target selection: Not affected
Vulnerability Itlb multihit:             Not affected
Vulnerability L1tf:                      Not affected
Vulnerability Mds:                       Not affected
Vulnerability Meltdown:                  Not affected
Vulnerability Mmio stale data:           Not affected
Vulnerability Reg file data sampling:    Not affected
Vulnerability Retbleed:                  Not affected
Vulnerability Spec rstack overflow:      Not affected
Vulnerability Spec store bypass:         Mitigation; Speculative Store Bypass disabled via prctl
Vulnerability Spectre v1:                Mitigation; __user pointer sanitization
Vulnerability Spectre v2:                Mitigation; CSV2, BHB
Vulnerability Srbds:                     Not affected
Vulnerability Tsa:                       Not affected
Vulnerability Tsx async abort:           Not affected
Vulnerability Vmscape:                   Not affected

Details

group                                          arrow_rs_9683                           main
-----                                          -------------                           ----
merge sorted f64                               1.00      3.8±0.01ms        ? ?/sec     1.00      3.8±0.01ms        ? ?/sec
merge sorted i64                               1.00      3.7±0.01ms        ? ?/sec     1.00      3.7±0.02ms        ? ?/sec
merge sorted mixed dictionary tuple            1.01      8.0±0.04ms        ? ?/sec     1.00      7.9±0.03ms        ? ?/sec
merge sorted mixed tuple                       1.01      8.1±0.02ms        ? ?/sec     1.00      8.0±0.02ms        ? ?/sec
merge sorted mixed tuple with utf8 view        1.02      7.7±0.02ms        ? ?/sec     1.00      7.6±0.02ms        ? ?/sec
merge sorted utf8 dictionary                   1.01      3.8±0.02ms        ? ?/sec     1.00      3.8±0.02ms        ? ?/sec
merge sorted utf8 dictionary tuple             1.01      5.4±0.02ms        ? ?/sec     1.00      5.4±0.02ms        ? ?/sec
merge sorted utf8 high cardinality             1.00      5.6±0.02ms        ? ?/sec     1.00      5.6±0.01ms        ? ?/sec
merge sorted utf8 low cardinality              1.00      3.8±0.01ms        ? ?/sec     1.00      3.8±0.01ms        ? ?/sec
merge sorted utf8 tuple                        1.00      8.5±0.05ms        ? ?/sec     1.00      8.5±0.02ms        ? ?/sec
merge sorted utf8 view high cardinality        1.00      4.7±0.02ms        ? ?/sec     1.00      4.7±0.02ms        ? ?/sec
merge sorted utf8 view low cardinality         1.00      2.7±0.01ms        ? ?/sec     1.00      2.7±0.01ms        ? ?/sec
merge sorted utf8 view tuple                   1.00      8.2±0.03ms        ? ?/sec     1.00      8.1±0.04ms        ? ?/sec
sort f64                                       1.00      5.4±0.02ms        ? ?/sec     1.01      5.4±0.02ms        ? ?/sec
sort i64                                       1.00      5.2±0.02ms        ? ?/sec     1.02      5.3±0.02ms        ? ?/sec
sort merge f64                                 1.05      3.8±0.01ms        ? ?/sec     1.00      3.7±0.01ms        ? ?/sec
sort merge i64                                 1.07      3.8±0.01ms        ? ?/sec     1.00      3.6±0.01ms        ? ?/sec
sort merge mixed dictionary tuple              1.03      7.8±0.02ms        ? ?/sec     1.00      7.5±0.02ms        ? ?/sec
sort merge mixed tuple                         1.05      8.1±0.03ms        ? ?/sec     1.00      7.7±0.02ms        ? ?/sec
sort merge mixed tuple with utf8 view          1.03      7.9±0.03ms        ? ?/sec     1.00      7.7±0.03ms        ? ?/sec
sort merge utf8 dictionary                     1.00      3.4±0.01ms        ? ?/sec     1.00      3.4±0.01ms        ? ?/sec
sort merge utf8 dictionary tuple               1.00      5.1±0.02ms        ? ?/sec     1.00      5.1±0.03ms        ? ?/sec
sort merge utf8 high cardinality               1.05      5.7±0.03ms        ? ?/sec     1.00      5.4±0.02ms        ? ?/sec
sort merge utf8 low cardinality                1.08      4.0±0.02ms        ? ?/sec     1.00      3.7±0.01ms        ? ?/sec
sort merge utf8 tuple                          1.02      8.9±0.04ms        ? ?/sec     1.00      8.7±0.03ms        ? ?/sec
sort merge utf8 view high cardinality          1.03      4.9±0.01ms        ? ?/sec     1.00      4.8±0.01ms        ? ?/sec
sort merge utf8 view low cardinality           1.13      2.9±0.01ms        ? ?/sec     1.00      2.5±0.02ms        ? ?/sec
sort merge utf8 view tuple                     1.03      8.4±0.04ms        ? ?/sec     1.00      8.2±0.03ms        ? ?/sec
sort mixed dictionary tuple                    1.01     11.2±0.04ms        ? ?/sec     1.00     11.1±0.04ms        ? ?/sec
sort mixed tuple                               1.01     10.6±0.04ms        ? ?/sec     1.00     10.5±0.05ms        ? ?/sec
sort mixed tuple with utf8 view                1.02     10.2±0.06ms        ? ?/sec     1.00     10.0±0.04ms        ? ?/sec
sort partitioned f64                           2.60  346.4±163.88µs        ? ?/sec     1.00  133.0±129.92µs        ? ?/sec
sort partitioned i64                           3.00   405.9±95.39µs        ? ?/sec     1.00  135.3±176.23µs        ? ?/sec
sort partitioned mixed dictionary tuple        1.49  748.2±109.88µs        ? ?/sec     1.00  501.9±110.71µs        ? ?/sec
sort partitioned mixed tuple                   2.16  712.8±144.83µs        ? ?/sec     1.00  330.0±137.80µs        ? ?/sec
sort partitioned mixed tuple with utf8 view    1.30  1216.7±149.09µs        ? ?/sec    1.00   939.2±77.99µs        ? ?/sec
sort partitioned utf8 dictionary               1.00  158.6±151.43µs        ? ?/sec     1.02  161.2±222.13µs        ? ?/sec
sort partitioned utf8 dictionary tuple         1.00  455.5±114.36µs        ? ?/sec     1.00   453.4±87.81µs        ? ?/sec
sort partitioned utf8 high cardinality         2.20   429.3±54.86µs        ? ?/sec     1.00  194.9±182.40µs        ? ?/sec
sort partitioned utf8 low cardinality          2.59   535.1±97.33µs        ? ?/sec     1.00   206.3±66.53µs        ? ?/sec
sort partitioned utf8 tuple                    1.23  1509.3±61.18µs        ? ?/sec     1.00  1223.9±115.89µs        ? ?/sec
sort partitioned utf8 view high cardinality    1.30   799.5±54.02µs        ? ?/sec     1.00  613.1±137.72µs        ? ?/sec
sort partitioned utf8 view low cardinality     3.57   500.6±69.08µs        ? ?/sec     1.00  140.3±124.84µs        ? ?/sec
sort partitioned utf8 view tuple               1.34  1390.1±129.11µs        ? ?/sec    1.00  1036.3±130.97µs        ? ?/sec
sort utf8 dictionary                           1.00   1103.7±7.62µs        ? ?/sec     1.00   1099.3±8.30µs        ? ?/sec
sort utf8 dictionary tuple                     1.01      8.3±0.04ms        ? ?/sec     1.00      8.3±0.04ms        ? ?/sec
sort utf8 high cardinality                     1.00      7.6±0.03ms        ? ?/sec     1.01      7.7±0.02ms        ? ?/sec
sort utf8 low cardinality                      1.00      6.0±0.02ms        ? ?/sec     1.01      6.0±0.02ms        ? ?/sec
sort utf8 tuple                                1.00     10.8±0.03ms        ? ?/sec     1.01     10.9±0.05ms        ? ?/sec
sort utf8 view high cardinality                1.00      6.7±0.02ms        ? ?/sec     1.00      6.7±0.03ms        ? ?/sec
sort utf8 view low cardinality                 1.00      4.5±0.03ms        ? ?/sec     1.02      4.5±0.02ms        ? ?/sec
sort utf8 view tuple                           1.00     10.5±0.04ms        ? ?/sec     1.00     10.5±0.04ms        ? ?/sec

Resource Usage

base (merge-base)

Metric	Value
Wall time	667.1s
Peak memory	4.6 GiB
Avg memory	4.5 GiB
CPU user	1623.6s
CPU sys	72.2s
Peak spill	0 B

branch

Metric	Value
Wall time	671.2s
Peak memory	4.6 GiB
Avg memory	4.4 GiB
CPU user	1871.7s
CPU sys	57.0s
Peak spill	0 B

---

File an issue against this benchmark runner

[mbutrovich]

Benchmark results show no improvement, which caused me to look at the sorting state machine a bit more.

ExternalSorter only concats batches into a single large sort when total buffered data is < 1MB (sort_in_place_threshold_bytes). Above that, it sorts each small batch (1024–8192 rows) individually and merges them. Radix sort needs a large batch to amortize RowConverter encoding overhead, but at small batch sizes the encoding cost dominates any radix advantage.

We can't simply raise the threshold because concat_batches temporarily 2x's memory, and RowConverter adds another large allocation on top. There's also no arrow-rs API to incrementally append to a Rows buffer across batches, so we can't amortize encoding as data arrives.

Shelving the DataFusion integration and focusing on landing the arrow-rs kernel for now.

[mbutrovich]

I'm also not sure why we default to batch size of 1024 for the sort benchmarks.