Create sorted merge node

cpp/src/arrow/acero/asof_join_node.cc

@@ -566,6 +566,7 @@ class InputState {
     return queue_.UnsyncFront();
   }
 
+// TODO(jeraguilon): consolidate
 #define LATEST_VAL_CASE(id, val)                     \
   case Type::id: {                                   \
     using T = typename TypeIdTraits<Type::id>::Type; \

cpp/src/arrow/acero/backpressure_handler.h

@@ -1,3 +1,7 @@
+<<<<<<< HEAD
+=======
+
+>>>>>>> b34c999b6 (Create sorted merge node)
 // Licensed to the Apache Software Foundation (ASF) under one
 // or more contributor license agreements.  See the NOTICE file
 // distributed with this work for additional information
@@ -16,6 +20,10 @@
 // under the License.
 
 #pragma once
+<<<<<<< HEAD
+=======
+#include "arrow/acero/exec_plan.h"
+>>>>>>> b34c999b6 (Create sorted merge node)
 #include "arrow/acero/options.h"
 
 #include <memory>
@@ -24,15 +32,26 @@ namespace arrow::acero {
 
 class BackpressureHandler {
  private:
+<<<<<<< HEAD
   BackpressureHandler(size_t low_threshold, size_t high_threshold,
                       std::unique_ptr<BackpressureControl> backpressure_control)
       : low_threshold_(low_threshold),
+=======
+  BackpressureHandler(ExecNode* input, size_t low_threshold, size_t high_threshold,
+                      std::unique_ptr<BackpressureControl> backpressure_control)
+      : input_(input),
+        low_threshold_(low_threshold),
+>>>>>>> b34c999b6 (Create sorted merge node)
         high_threshold_(high_threshold),
         backpressure_control_(std::move(backpressure_control)) {}
 
  public:
   static Result<BackpressureHandler> Make(
+<<<<<<< HEAD
       size_t low_threshold, size_t high_threshold,
+=======
+      ExecNode* input, size_t low_threshold, size_t high_threshold,
+>>>>>>> b34c999b6 (Create sorted merge node)
       std::unique_ptr<BackpressureControl> backpressure_control) {
     if (low_threshold >= high_threshold) {
       return Status::Invalid("low threshold (", low_threshold,
@@ -41,7 +60,11 @@ class BackpressureHandler {
     if (backpressure_control == NULLPTR) {
       return Status::Invalid("null backpressure control parameter");
     }
+<<<<<<< HEAD
     BackpressureHandler backpressure_handler(low_threshold, high_threshold,
+=======
+    BackpressureHandler backpressure_handler(input, low_threshold, high_threshold,
+>>>>>>> b34c999b6 (Create sorted merge node)
                                              std::move(backpressure_control));
     return std::move(backpressure_handler);
   }
@@ -54,7 +77,20 @@ class BackpressureHandler {
     }
   }
 
+<<<<<<< HEAD
+ private:
+=======
+  Status ForceShutdown() {
+    // It may be unintuitive to call Resume() here, but this is to avoid a deadlock.
+    // Since acero's executor won't terminate if any one node is paused, we need to
+    // force resume the node before stopping production.
+    backpressure_control_->Resume();
+    return input_->StopProducing();
+  }
+
  private:
+  ExecNode* input_;
+>>>>>>> b34c999b6 (Create sorted merge node)
   size_t low_threshold_;
   size_t high_threshold_;
   std::unique_ptr<BackpressureControl> backpressure_control_;

cpp/src/arrow/acero/concurrent_queue.h

@@ -140,6 +140,11 @@ class BackpressureConcurrentQueue : public ConcurrentQueue<T> {
     return ConcurrentQueue<T>::TryPopUnlocked();
   }
 
+<<<<<<< HEAD
+=======
+  Status ForceShutdown() { return handler_.ForceShutdown(); }
+
+>>>>>>> b34c999b6 (Create sorted merge node)
  private:
   BackpressureHandler handler_;
 };