Wait for process_task to be marked finished

Rafał Hibner · Rafał Hibner · commit b46311ce69e1 · 2025-07-03T14:57:19.000Z
EndFromProcessThread spawns function using asof_join_node that can outlive
the node itself. Wait for task to be marked finished.
Also mark finished in case node was never started.
diff --git a/cpp/src/arrow/acero/asof_join_node.cc b/cpp/src/arrow/acero/asof_join_node.cc
@@ -1151,6 +1151,7 @@ class AsofJoinNode : public ExecNode {
       process_thread_.join();
     }
 #endif
+    process_task_.Wait();
   }
 
   const std::vector<col_index_t>& indices_of_on_key() { return indices_of_on_key_; }
@@ -1593,7 +1594,8 @@ AsofJoinNode::AsofJoinNode(ExecPlan* plan, NodeVector inputs,
 #ifdef ARROW_ENABLE_THREADING
       ,
       process_(),
-      process_thread_()
+      process_thread_(),
+      process_task_(Future<>::MakeFinished())
 #endif
 {
   for (auto& key_hasher : key_hashers_) {

Original file line number	Diff line number	Diff line change
`@@ -1151,6 +1151,7 @@ class AsofJoinNode : public ExecNode {`
`1151`	`1151`	`process_thread_.join();`
`1152`	`1152`	`}`
`1153`	`1153`	`#endif`
	`1154`	`+ process_task_.Wait();`
`1154`	`1155`	`}`
`1155`	`1156`
`1156`	`1157`	`const std::vector<col_index_t>& indices_of_on_key() { return indices_of_on_key_; }`
`@@ -1593,7 +1594,8 @@ AsofJoinNode::AsofJoinNode(ExecPlan* plan, NodeVector inputs,`
`1593`	`1594`	`#ifdef ARROW_ENABLE_THREADING`
`1594`	`1595`	`,`
`1595`	`1596`	`process_(),`
`1596`		`- process_thread_()`
	`1597`	`+ process_thread_(),`
	`1598`	`+ process_task_(Future<>::MakeFinished())`
`1597`	`1599`	`#endif`
`1598`	`1600`	`{`
`1599`	`1601`	`for (auto& key_hasher : key_hashers_) {`