GH-46224: [C++][Acero] Fix the hang in asof join

cpp/src/arrow/acero/asof_join_node.cc

@@ -639,12 +639,10 @@ class InputState : public util::SerialSequencingQueue::Processor {
         // hit the end of the batch, need to get the next batch if possible.
         ++batches_processed_;
         latest_ref_row_ = 0;
-        have_active_batch &= !queue_.TryPop();
-        if (have_active_batch) {
-          DCHECK_GT(queue_.Front()->num_rows(), 0);  // empty batches disallowed
-          memo_.UpdateTime(GetTime(queue_.Front().get(), time_type_id_, time_col_index_,
-                                   0));  // time changed
-        }
+        bool did_pop = queue_.TryPop().has_value();
+        DCHECK(did_pop);
+        ARROW_UNUSED(did_pop);
+        have_active_batch = !queue_.Empty();
       }
     }
     return have_active_batch;

cpp/src/arrow/acero/asof_join_node_test.cc

@@ -44,6 +44,7 @@
 #include "arrow/compute/cast.h"
 #include "arrow/compute/row/row_encoder_internal.h"
 #include "arrow/compute/test_util_internal.h"
+#include "arrow/testing/generator.h"
 #include "arrow/testing/gtest_util.h"
 #include "arrow/testing/matchers.h"
 #include "arrow/testing/random.h"
@@ -1770,5 +1771,58 @@ TEST(AsofJoinTest, DestroyNonStartedAsofJoinNode) {
       DeclarationToStatus(std::move(sink)));
 }
 
+// Reproduction of GH-46224: Hang when all left timestamps are greater than right
+// timestamps.
+TEST(AsofJoinTest, OneSideTsAllGreaterThanTheOther) {
+#if defined(ARROW_VALGRIND) || defined(ADDRESS_SANITIZER)
+  const int rounds = 1;
+#else
+  const int rounds = 42;
+#endif
+  int64_t tolerance = 1;
+  int64_t num_rows_big_ts = 1;
+  int64_t num_rows_small_ts = ExecPlan::kMaxBatchSize + 1;
+  // Make sure the big_ts is outside the horizon of the tolerance regardless of the side.
+  int64_t big_ts = num_rows_small_ts + tolerance + 1;
+
+  // Column of big timestamps.
+  ASSERT_OK_AND_ASSIGN(auto col_big_ts,
+                       gen::Constant(MakeScalar(big_ts))->Generate(num_rows_big_ts));
+  // Column of small timestamps from 0 to num_rows_small_ts - 1.
+  ASSERT_OK_AND_ASSIGN(auto col_small_ts,
+                       gen::Step<int64_t>()->Generate(num_rows_small_ts));
+
+  struct Case {
+    std::shared_ptr<arrow::Array> left_col;
+    std::shared_ptr<arrow::Array> right_col;
+  };
+
+  for (const auto& c : {
+           Case{col_big_ts, col_small_ts},
+           Case{col_small_ts, col_big_ts},
+       }) {
+    auto left_schema = arrow::schema({arrow::field("on", int64())});
+    auto right_schema = arrow::schema({arrow::field("on", int64())});
+
+    ExecBatch left_batch({c.left_col}, c.left_col->length());
+    ExecBatch right_batch({c.right_col}, c.right_col->length());
+    ASSERT_OK_AND_ASSIGN(auto col_null, MakeArrayOfNull(int64(), c.left_col->length()));
+    ExecBatch exp_batch({c.left_col, col_null}, c.left_col->length());
+
+    // Run moderate number of times to ensure that no hangs occur.
+    for (int i = 0; i < rounds; ++i) {
+      AsofJoinNodeOptions opts({{{"on"}, {}}, {{"on"}, {}}}, tolerance);
+      auto left = Declaration("exec_batch_source",
+                              ExecBatchSourceNodeOptions(left_schema, {left_batch}));
+      auto right = Declaration("exec_batch_source",
+                               ExecBatchSourceNodeOptions(right_schema, {right_batch}));
+      auto asof_join = arrow::acero::Declaration{"asofjoin", {left, right}, opts};
+      ASSERT_OK_AND_ASSIGN(auto result,
+                           arrow::acero::DeclarationToExecBatches(std::move(asof_join)));
+      AssertExecBatchesEqualIgnoringOrder(result.schema, {exp_batch}, result.batches);
+    }
+  }
+}
+
 }  // namespace acero
 }  // namespace arrow