Skip to content

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

Open
gitmodimo wants to merge 4 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 4 commits into
apache:mainfrom
gitmodimo:sorted_merge_rework

Conversation

@gitmodimo

@gitmodimo gitmodimo 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
Contributor Author

@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
Contributor Author

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

Execution Inputs Batch size Baseline Reworked Speedup
Serial 2 1,024 269.740 ms 111.019 ms 2.43×
Serial 8 64 841.132 ms 154.442 ms 5.45×
Serial 8 1,024 785.382 ms 113.241 ms 6.94×
Serial 8 16,384 771.922 ms 108.003 ms 7.15×
Serial 32 1,024 2,848.301 ms 128.857 ms 22.10×
Threaded 2 1,024 258.336 ms 130.043 ms 1.99×
Threaded 8 64 928.467 ms 204.125 ms 4.55×
Threaded 8 1,024 802.374 ms 118.101 ms 6.79×
Threaded 8 16,384 766.673 ms 116.209 ms 6.60×
Threaded 32 1,024 2,807.121 ms 122.661 ms 22.89×

Sign up for free to 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