Skip to content

GH-47393: [C++][Acero] Support for multi threaded input: SortedMergeNode Version 2 - #51141

Open
gitmodimo wants to merge 5 commits into
apache:mainfrom
gitmodimo:sorted_merge_rework
Open

GH-47393: [C++][Acero] Support for multi threaded input: SortedMergeNode Version 2#51141
gitmodimo wants to merge 5 commits into
apache:mainfrom
gitmodimo:sorted_merge_rework

Conversation

@gitmodimo

@gitmodimogitmodimo commented Sep 2, 2026

Copy link
Copy Markdown
Contributor

Rationale for this change

SortedMerge uses a dedicated worker thread and assumes input batches arrive in logical order. This prevents safe threaded execution and leaves backpressure, completion, stopping, and error propagation split across separate execution paths. The earlier implementation in #47394 added per-input sequencing and backpressure while retaining the existing worker thread; this PR is a new version that addresses #47393 with a single executor-managed implementation.

What changes are included in this PR?

  • Replace the worker thread, process queue, and poison-pill shutdown with one executor-managed merge flow. Per-input SerialSequencingQueues restore logical ExecBatch::index order for threaded, serial, and compile-time no-thread builds.
  • Add bounded upstream backpressure and downstream pause/resume handling. Completion flushes buffered rows, stopping propagates upstream, and sink-driven resume cannot race scheduler teardown.
  • Select multi-row runs up to the earliest row visible on another input, allowing one selection to span multiple timestamps. Runs retain their original ExecBatch payloads until direct materialization with Arrow builders. This supports scalar payloads and bypasses Boolean slices with non-zero offsets reported in [C++][Acero] materializeColumn for boolean type overflow #48072.
  • Validate ascending input order and propagate validation, materialization, and downstream errors through normal ExecPlan status handling.

Are these changes tested?

Yes. All tests pass in threaded and ARROW_ENABLE_THREADING=OFF builds. Coverage includes signed keys crossing zero, scalar payloads, deterministic sequencing under jittered delivery, and downstream backpressure, resume, and stop.

Are there any user-facing changes?

SortedMerge now supports concurrent input delivery, upstream and downstream backpressure, scalar payloads. Inputs that violate their declared ascending order return an Invalid status.
No API changes.

@gitmodimo

Copy link
Copy Markdown
ContributorAuthor

@ursabot please benchmark

@rok

rok commented Sep 2, 2026

Copy link
Copy Markdown
Member

Benchmark runs are scheduled for commit 8470111. Watch https://buildkite.com/apache-arrow and https://conbench.arrow-dev.org for updates. A comment will be posted here when the runs are complete.

@gitmodimo

Copy link
Copy Markdown
ContributorAuthor

Since codebench doesn't seem to work here are my local benchmark run on i9-14900K:

ExecutionInputsBatch sizeBaselineReworkedSpeedup
Serial21,024269.740 ms111.019 ms2.43×
Serial864841.132 ms154.442 ms5.45×
Serial81,024785.382 ms113.241 ms6.94×
Serial816,384771.922 ms108.003 ms7.15×
Serial321,0242,848.301 ms128.857 ms22.10×
Threaded21,024258.336 ms130.043 ms1.99×
Threaded864928.467 ms204.125 ms4.55×
Threaded81,024802.374 ms118.101 ms6.79×
Threaded816,384766.673 ms116.209 ms6.60×
Threaded321,0242,807.121 ms122.661 ms22.89×

Sign up for freeto join this conversation on GitHub. Already have an account? Sign in to comment

Labels

Projects

None yet

Development

Successfully merging this pull request may close these issues.

2 participants

@gitmodimo@rok