Parallel merge in SortPreservingMergeExec after sort elimination

[zhuqi-lucas]

Is your feature request related to a problem or challenge?

When SortExec is eliminated via sort pushdown (#21182, statistics-based file reordering), SortPreservingMergeExec becomes the only merge operator, reading directly from I/O-bound DataSourceExec partitions instead of from SortExec's in-memory buffer.

Currently we insert a BufferExec to compensate, but with many partitions the single-threaded K-way merge in SPM can still become a bottleneck — it alternates between partitions, and each partition switch may stall on I/O even with buffering.

Describe the solution you'd like

When SPM reads from I/O-bound sources (no buffering SortExec), split the N input streams into groups and merge each group in parallel, then merge the intermediate results. This creates a tree of merges:

Level 1 (parallel):  merge(s1,s2), merge(s3,s4), merge(s5,s6), merge(s7,s8)
Level 2 (parallel):  merge(m1,m2), merge(m3,m4)
Level 3:             merge(m5,m6) → final output

This would be especially beneficial when:

• Sort elimination removes the buffering SortExec, making SPM I/O-bound
• Many partitions with I/O-bound sources
• Large datasets where single-threaded merge becomes the bottleneck

Additional context

Suggested by @Dandandan in #21182 (comment)

Related: DuckDB's parallel k-way merge

Parent issue: #17348

[2010YOUY01]

When SortExec is eliminated via sort pushdown (statistics-based file reordering), SortPreservingMergeExec reads directly from I/O-bound sources instead of from SortExec's in-memory buffer. Currently SPM does a single K-way merge of all input streams, which can become a bottleneck when there are many partitions.

The current implementation already forms a tree of SPMs similar to what you have described, so there shouldn’t be a major bottleneck I think, though there may still be room for improvement.

Currently, there are two stages of SPM:

• Stage 1: SortExec output, where the merge degree equals the number of in-memory batches
• Stage 2: Final SortPreservingMergeExec, where the merge degree equals the number of partitions

Stage 1 is non-blocking, so the two SPM stages can run in parallel. The main optimization opportunity seems to be balancing the merge degrees across the two stages to avoid one becoming a bottleneck. For example, if SortExec performs a 4-way merge while Stage 2 performs a 32-way merge, Stage 2 is likely to be the bottleneck. This could potentially be improved with a more recursive merging scheme.

Related: DuckDB implements a similar parallel merge sort strategy.

They have a more recent version with a parallel k-way merge algorithm: https://duckdb.org/2025/09/24/sorting-again#k-way-merge-path

[zhuqi-lucas]

Thanks @2010YOUY01 for the context! This issue is specifically about the case after SortExec is eliminated via sort pushdown (#21182). When sort elimination removes SortExec entirely, there is no Stage 1 merge. SPM becomes the only merge operator, reading directly from I/O-bound DataSourceExec partitions.

I've updated the issue description to clarify in case any misunderstanding.

In that scenario, a parallel merge within SPM itself (splitting N input streams into groups and merging in parallel) could help. The DuckDB parallel k-way merge link is great — thanks for sharing!

[Dandandan]

My comment was about the final SortPreservingMergeExec stage indeed, not the internal merging of batches within SortExec.

I think both cases (full sort and merge-only) might benefit from more parallelism in the final stage, but in cases SPM is the only stage there might be substantial more benefit, probably less buffering will be necessary to introduce enough parallelism in the inner streams.

[2010YOUY01]

Thanks @2010YOUY01 for the context! This issue is specifically about the case after SortExec is eliminated via sort pushdown (#21182). When sort elimination removes SortExec entirely, there is no Stage 1 merge. SPM becomes the only merge operator, reading directly from I/O-bound DataSourceExec partitions.

I've updated the issue description to clarify in case any misunderstanding.

In that scenario, a parallel merge within SPM itself (splitting N input streams into groups and merging in parallel) could help. The DuckDB parallel k-way merge link is great — thanks for sharing!

Ah, I see — that's a good optimization opportunity.

Would you mind if I work on this, if you haven’t started on it yet? I’ve been thinking about improving the merge strategy similarly.

[zhuqi-lucas]

Thanks @2010YOUY01 for the context! This issue is specifically about the case after SortExec is eliminated via sort pushdown (#21182). When sort elimination removes SortExec entirely, there is no Stage 1 merge. SPM becomes the only merge operator, reading directly from I/O-bound DataSourceExec partitions.
I've updated the issue description to clarify in case any misunderstanding.
In that scenario, a parallel merge within SPM itself (splitting N input streams into groups and merging in parallel) could help. The DuckDB parallel k-way merge link is great — thanks for sharing!

Ah, I see — that's a good optimization opportunity.

Would you mind if I work on this, if you haven’t started on it yet? I’ve been thinking about improving the merge strategy similarly.

Sure @2010YOUY01 , feel free to take it.