Move BackpressureController from Asofasof_join_node and sorted_merge_node to common backpressure

Author: Rafał Hibner

cpp/src/arrow/acero/asof_join_node.cc

@@ -46,6 +46,7 @@
 #ifndef NDEBUG
 #  include "arrow/compute/function_internal.h"
 #endif
+#include "arrow/acero/backpressure.h"
 #include "arrow/acero/time_series_util.h"
 #include "arrow/compute/key_hash_internal.h"
 #include "arrow/compute/light_array_internal.h"
@@ -459,21 +460,6 @@ class KeyHasher {
   arrow::util::TempVectorStack stack_;
 };
 
-class BackpressureController : public BackpressureControl {
- public:
-  BackpressureController(ExecNode* node, ExecNode* output,
-                         std::atomic<int32_t>& backpressure_counter)
-      : node_(node), output_(output), backpressure_counter_(backpressure_counter) {}
-
-  void Pause() override { node_->PauseProducing(output_, ++backpressure_counter_); }
-  void Resume() override { node_->ResumeProducing(output_, ++backpressure_counter_); }
-
- private:
-  ExecNode* node_;
-  ExecNode* output_;
-  std::atomic<int32_t>& backpressure_counter_;
-};
-
 class InputState : public util::SerialSequencingQueue::Processor {
   // InputState corresponds to an input
   // Input record batches are queued up in InputState until processed and

cpp/src/arrow/acero/backpressure.cc

@@ -16,7 +16,19 @@
 // under the License.
 
 #include "arrow/acero/backpressure.h"
+#include "arrow/acero/exec_plan.h"
+
 namespace arrow::acero {
+BackpressureController::BackpressureController(ExecNode* node, ExecNode* output,
+                                               std::atomic<int32_t>& backpressure_counter)
+    : node_(node), output_(output), backpressure_counter_(backpressure_counter) {}
+void BackpressureController::Pause() {
+  node_->PauseProducing(output_, ++backpressure_counter_);
+}
+void BackpressureController::Resume() {
+  node_->ResumeProducing(output_, ++backpressure_counter_);
+}
+
 BackpressureCombiner::BackpressureCombiner(
     std::unique_ptr<BackpressureControl> backpressure_control)
     : backpressure_control_(std::move(backpressure_control)) {}

cpp/src/arrow/acero/backpressure.h

@@ -21,6 +21,21 @@
 #include <mutex>
 namespace arrow::acero {
 
+// Generic backpressure controller for ExecNode
+class BackpressureController : public BackpressureControl {
+ public:
+  BackpressureController(ExecNode* node, ExecNode* output,
+                         std::atomic<int32_t>& backpressure_counter);
+
+  void Pause() override;
+  void Resume() override;
+
+ private:
+  ExecNode* node_;
+  ExecNode* output_;
+  std::atomic<int32_t>& backpressure_counter_;
+};
+
 // Provides infrastructure of combining multiple backpressure sources and propagate the
 // result into BackpressureControl There are two types of Source: strong - pause on any
 // strong Source within controller

cpp/src/arrow/acero/sorted_merge_node.cc

@@ -24,6 +24,7 @@
 #include <unordered_map>
 #include <vector>
 
+#include "arrow/acero/backpressure.h"
 #include "arrow/acero/concurrent_queue_internal.h"
 #include "arrow/acero/exec_plan.h"
 #include "arrow/acero/exec_plan_internal.h"
@@ -82,21 +83,6 @@ using col_index_t = int;
 constexpr bool kNewTask = true;
 constexpr bool kPoisonPill = false;
 
-class BackpressureController : public BackpressureControl {
- public:
-  BackpressureController(ExecNode* node, ExecNode* output,
-                         std::atomic<int32_t>& backpressure_counter)
-      : node_(node), output_(output), backpressure_counter_(backpressure_counter) {}
-
-  void Pause() override { node_->PauseProducing(output_, ++backpressure_counter_); }
-  void Resume() override { node_->ResumeProducing(output_, ++backpressure_counter_); }
-
- private:
-  ExecNode* node_;
-  ExecNode* output_;
-  std::atomic<int32_t>& backpressure_counter_;
-};
-
 /// InputState corresponds to an input. Input record batches are queued up in InputState
 /// until processed and turned into output record batches.
 class InputState {