Handle PauseProducing in asof_join

Author: Rafał Hibner

cpp/src/arrow/acero/asof_join_node.cc

@@ -1108,6 +1108,7 @@ class AsofJoinNode : public ExecNode {
         EndFromProcessThread();
         return;
       }
+      backpressure_future_.Wait();
       if (!Process()) {
         return;
       }
@@ -1520,8 +1521,51 @@ class AsofJoinNode : public ExecNode {
     return Status::OK();
   }
 
-  void PauseProducing(ExecNode* output, int32_t counter) override {}
-  void ResumeProducing(ExecNode* output, int32_t counter) override {}
+  void PauseProducing(ExecNode* output, int32_t counter) override {
+    std::lock_guard<std::mutex> lg(backpressure_mutex_);
+    if (counter <= last_backpressure_counter_) {
+      return;
+    }
+    last_backpressure_counter_ = counter;
+    if (!backpressure_future_.is_finished()) {
+      // Could happen if we get something like Pause(1) Pause(3) Resume(2)
+      return;
+    }
+    backpressure_future_ = Future<>::Make();
+  }
+  void ResumeProducing(ExecNode* output, int32_t counter) override {
+    Future<> to_finish;
+    {
+      std::lock_guard<std::mutex> lg(backpressure_mutex_);
+      if (counter <= last_backpressure_counter_) {
+        return;
+      }
+      last_backpressure_counter_ = counter;
+      if (backpressure_future_.is_finished()) {
+        return;
+      }
+      to_finish = backpressure_future_;
+      backpressure_future_ = Future<>::MakeFinished();
+    }
+    to_finish.MarkFinished();
+  }
+
+  Status StopProducing() override {
+    // GH-35837: ensure node is not paused
+    Future<> to_finish;
+    {
+      std::lock_guard<std::mutex> lg(backpressure_mutex_);
+      if (!backpressure_future_.is_finished()) {
+        to_finish = backpressure_future_;
+        backpressure_future_ = Future<>::MakeFinished();
+      }
+    }
+    if (to_finish.is_valid()) {
+      to_finish.MarkFinished();
+    }
+    // only then stop
+    return ExecNode::StopProducing();
+  }
 
   Status StopProducingImpl() override {
 #ifdef ARROW_ENABLE_THREADING
@@ -1549,6 +1593,11 @@ class AsofJoinNode : public ExecNode {
   // Each input state corresponds to an input table
   std::vector<std::unique_ptr<InputState>> state_;
   std::mutex gate_;
+
+  std::mutex backpressure_mutex_;
+  std::atomic<int32_t> last_backpressure_counter_{0};
+  Future<> backpressure_future_ = Future<>::MakeFinished();
+
   TolType tolerance_;
 #ifndef NDEBUG
   std::ostream* debug_os_;