Add stop in backpressure combiner

Author: Rafał Hibner

cpp/src/arrow/acero/backpressure.cc

@@ -57,7 +57,15 @@ void BackpressureCombiner::Resume(Source* output) {
   }
 }
 
+void BackpressureCombiner::Stop() {
+  std::lock_guard<std::mutex> lg(mutex_);
+  stopped = true;
+  backpressure_control_->Resume();
+  paused = false;
+}
+
 void BackpressureCombiner::UpdatePauseStateUnlocked() {
+  if (stopped) return;
   bool should_be_paused = (paused_count_ > 0);
   if (!pause_on_any_) {
     should_be_paused = should_be_paused && (paused_count_ == paused_.size());

cpp/src/arrow/acero/backpressure.h

@@ -77,6 +77,8 @@ class ARROW_ACERO_EXPORT BackpressureCombiner {
     std::vector<BackpressureCombiner*> connections_;
   };
 
+  void Stop();
+
  private:
   friend class Source;
   void Pause(Source* output);
@@ -89,6 +91,7 @@ class ARROW_ACERO_EXPORT BackpressureCombiner {
   std::unordered_map<Source*, bool> paused_;
   size_t paused_count_{0};
   bool paused{false};
+  bool stopped{false};
 };
 
 }  // namespace arrow::acero