Adjust iterativelength task as well

Author: Dtenwolde

src/core/operator/bfs_state.cpp

@@ -1,6 +1,8 @@
 #include "duckpgq/core/operator/bfs_state.hpp"
 
 #include <duckpgq/core/operator/event/shortest_path_event.hpp>
+#include <duckpgq/core/operator/event/iterative_length_event.hpp>
+
 #include <duckpgq/core/utils/compressed_sparse_row.hpp>
 #include <duckpgq/core/utils/duckpgq_barrier.hpp>
 #include <duckpgq/core/utils/duckpgq_utils.hpp>
@@ -159,7 +161,6 @@ void BFSState::InitializeLanes() {
 
 void BFSState::ScheduleBFSBatch(Pipeline &pipeline, Event &event, const PhysicalPathFinding *op) {
   if (mode == "iterativelength") {
-    throw NotImplementedException("Iterative length has not been implemented yet");
     event.InsertEvent(
     make_shared_ptr<IterativeLengthEvent>(shared_from_this(), pipeline, *op));
   } else if (mode == "shortestpath") {

src/core/operator/event/iterative_length_event.cpp

@@ -5,29 +5,25 @@
 namespace duckpgq {
 namespace core {
 
-ParallelIterativeEvent::ParallelIterativeEvent(GlobalBFSState &gstate_p, Pipeline &pipeline_p, const PhysicalPathFinding &op_p)
-      : BasePipelineEvent(pipeline_p), gstate(gstate_p), op(op_p) {}
+IterativeLengthEvent::IterativeLengthEvent(shared_ptr<BFSState> gbfs_state_p,
+                          Pipeline &pipeline_p, const PhysicalPathFinding &op_p)
+    : BasePipelineEvent(pipeline_p), gbfs_state(std::move(gbfs_state_p)), op(op_p) {
 
+}
 
-void ParallelIterativeEvent::Schedule() {
-  auto &context = pipeline->GetClientContext();
 
+void IterativeLengthEvent::Schedule() {
+  auto &context = pipeline->GetClientContext();
+  // std::cout << gbfs_state->csr->ToString();
   vector<shared_ptr<Task>> bfs_tasks;
-  size_t threads_to_schedule = std::min(gstate.num_threads, (idx_t)gstate.global_task_queue.size());
-
-  for (idx_t tnum = 0; tnum < threads_to_schedule; tnum++) {
-    bfs_tasks.push_back(make_uniq<PhysicalIterativeTask>(
-        shared_from_this(), context, gstate, tnum, op));
+  for (idx_t tnum = 0; tnum < gbfs_state->scheduled_threads; tnum++) {
+    bfs_tasks.push_back(make_uniq<IterativeLengthTask>(
+        shared_from_this(), context, gbfs_state, tnum, op));
   }
   SetTasks(std::move(bfs_tasks));
 }
 
-void ParallelIterativeEvent::FinishEvent() {
-
-  // // if remaining pairs, schedule the BFS for the next batch
-  // if (bfs_state->started_searches < gstate.global_pairs->Count()) {
-  //   op.ScheduleBFSEvent(*pipeline, *this, gstate);
-  // }
+void IterativeLengthEvent::FinishEvent() {
 }
 
 } // namespace core

src/core/operator/task/iterative_length_task.cpp

@@ -4,14 +4,17 @@
 namespace duckpgq {
 namespace core {
 
-PhysicalIterativeTask::PhysicalIterativeTask(shared_ptr<Event> event_p, ClientContext &context,
-                        PathFindingGlobalSinkState &state, idx_t worker_id,
-                                             const PhysicalOperator &op_p)
-      : ExecutorTask(context, std::move(event_p), op_p), context(context),
-        state(state), worker_id(worker_id) {}
-
-bool PhysicalIterativeTask::SetTaskRange() {
-  auto task = state.global_bfs_state->FetchTask();
+IterativeLengthTask::IterativeLengthTask(shared_ptr<Event> event_p,
+                                   ClientContext &context,
+                                   shared_ptr<BFSState> &state, idx_t worker_id,
+                                   const PhysicalOperator &op_p)
+: ExecutorTask(context, std::move(event_p), op_p), context(context),
+  state(state), worker_id(worker_id) {
+  left = right = UINT64_MAX; // NOLINT
+}
+
+bool IterativeLengthTask::SetTaskRange() {
+  auto task = state->FetchTask();
   if (task == nullptr) {
     return false;
   }
@@ -21,14 +24,13 @@ bool PhysicalIterativeTask::SetTaskRange() {
 }
 
 
-  TaskExecutionResult PhysicalIterativeTask::ExecuteTask(TaskExecutionMode mode) {
-    auto &bfs_state = state.global_bfs_state;
-    auto &barrier = bfs_state->barrier;
+  TaskExecutionResult IterativeLengthTask::ExecuteTask(TaskExecutionMode mode) {
+    auto &barrier = state->barrier;
     do {
 
       IterativeLength();
       barrier->Wait([&]() {
-            bfs_state->ResetTaskIndex();  // Reset task index safely
+            state->ResetTaskIndex();  // Reset task index safely
         });
 
       barrier->Wait();
@@ -38,11 +40,11 @@ bool PhysicalIterativeTask::SetTaskRange() {
       }
 
       barrier->Wait([&]() {
-        bfs_state->ResetTaskIndex();  // Reset task index safely
+        state->ResetTaskIndex();  // Reset task index safely
       });
 
       barrier->Wait();
-    } while (bfs_state->change);
+    } while (state->change);
 
     if (worker_id == 0) {
       UnReachableSet();
@@ -52,19 +54,17 @@ bool PhysicalIterativeTask::SetTaskRange() {
     return TaskExecutionResult::TASK_FINISHED;
   }
 
-  void PhysicalIterativeTask::IterativeLength() {
-    auto &bfs_state = state.global_bfs_state;
-    auto &seen = bfs_state->seen;
-    auto &visit = bfs_state->iter & 1 ? bfs_state->visit1 : bfs_state->visit2;
-    auto &next = bfs_state->iter & 1 ? bfs_state->visit2 : bfs_state->visit1;
-    auto &barrier = bfs_state->barrier;
-    int64_t *v = (int64_t *)state.csr->v;
-    vector<int64_t> &e = state.csr->e;
-    auto &change = bfs_state->change;
+  void IterativeLengthTask::IterativeLength() {
+    auto &seen = state->seen;
+    auto &visit = state->iter & 1 ? state->visit1 : state->visit2;
+    auto &next = state->iter & 1 ? state->visit2 : state->visit1;
+    auto &barrier = state->barrier;
+    int64_t *v = (int64_t *)state->csr->v;
+    vector<int64_t> &e = state->csr->e;
+    auto &change = state->change;
 
     // Attempt to get a task range
     bool has_tasks = SetTaskRange();
-    // std::cout << "Worker " << worker_id << ": Has tasks = " << has_tasks << std::endl;
 
     // Clear `next` array regardless of task availability
     for (auto i = left; i < right; i++) {
@@ -81,7 +81,7 @@ bool PhysicalIterativeTask::SetTaskRange() {
                 for (auto offset = v[i]; offset < v[i + 1]; offset++) {
                     auto n = e[offset];
                     {
-                        std::lock_guard<std::mutex> lock(bfs_state->element_locks[n]);
+                        std::lock_guard<std::mutex> lock(state->element_locks[n]);
                         next[n] |= visit[i];
                     }
                 }
@@ -93,7 +93,7 @@ bool PhysicalIterativeTask::SetTaskRange() {
     // Synchronize at the end of the main processing
     barrier->Wait([&]() {
         // std::cout << "Worker " << worker_id << ": Resetting task index." << std::endl;
-        bfs_state->ResetTaskIndex();
+        state->ResetTaskIndex();
     });
     barrier->Wait();
 
@@ -114,47 +114,44 @@ bool PhysicalIterativeTask::SetTaskRange() {
 
     // Final synchronization after processing
     barrier->Wait([&]() {
-        // std::cout << "Worker " << worker_id << ": Resetting task index at second barrier." << std::endl;
-        bfs_state->ResetTaskIndex();
+        state->ResetTaskIndex();
     });
     barrier->Wait();
 }
 
-  void PhysicalIterativeTask::ReachDetect() const {
-    auto &bfs_state = state.global_bfs_state;
-    auto result_data = FlatVector::GetData<int64_t>(bfs_state->result.data[0]);
+  void IterativeLengthTask::ReachDetect() const {
+    auto result_data = FlatVector::GetData<int64_t>(state->pf_results->data[0]);
 
     // detect lanes that finished
     for (int64_t lane = 0; lane < LANE_LIMIT; lane++) {
-      int64_t search_num = bfs_state->lane_to_num[lane];
+      int64_t search_num = state->lane_to_num[lane];
       if (search_num >= 0) { // active lane
-        int64_t dst_pos = bfs_state->vdata_dst.sel->get_index(search_num);
-        if (bfs_state->seen[bfs_state->dst[dst_pos]][lane]) {
+        int64_t dst_pos = state->vdata_dst.sel->get_index(search_num);
+        if (state->seen[state->dst[dst_pos]][lane]) {
           result_data[search_num] =
-              bfs_state->iter; /* found at iter => iter = path length */
-          bfs_state->lane_to_num[lane] = -1; // mark inactive
-          bfs_state->active--;
+              state->iter; /* found at iter => iter = path length */
+          state->lane_to_num[lane] = -1; // mark inactive
+          state->active--;
         }
       }
     }
-    if (bfs_state->active == 0) {
-      bfs_state->change = false;
+    if (state->active == 0) {
+      state->change = false;
     }
     // into the next iteration
-    bfs_state->iter++;
+    state->iter++;
   }
 
-  void PhysicalIterativeTask::UnReachableSet() const {
-    auto &bfs_state = state.global_bfs_state;
-    auto result_data = FlatVector::GetData<int64_t>(bfs_state->result.data[0]);
-    auto &result_validity = FlatVector::Validity(bfs_state->result.data[0]);
+  void IterativeLengthTask::UnReachableSet() const {
+    auto result_data = FlatVector::GetData<int64_t>(state->pf_results->data[0]);
+    auto &result_validity = FlatVector::Validity(state->pf_results->data[0]);
 
     for (int64_t lane = 0; lane < LANE_LIMIT; lane++) {
-      int64_t search_num = bfs_state->lane_to_num[lane];
+      int64_t search_num = state->lane_to_num[lane];
       if (search_num >= 0) { // active lane
         result_validity.SetInvalid(search_num);
         result_data[search_num] = (int64_t)-1; /* no path */
-        bfs_state->lane_to_num[lane] = -1;     // mark inactive
+        state->lane_to_num[lane] = -1;     // mark inactive
       }
     }
   }

src/include/duckpgq/core/operator/task/iterative_length_task.hpp

@@ -7,11 +7,11 @@
 namespace duckpgq {
 namespace core {
 
-class PhysicalIterativeTask : public ExecutorTask {
+class IterativeLengthTask : public ExecutorTask {
 public:
-  PhysicalIterativeTask(shared_ptr<Event> event_p, ClientContext &context,
-                        PathFindingGlobalSinkState &state, idx_t worker_id,
-                        const PhysicalOperator &op_p);
+  IterativeLengthTask(shared_ptr<Event> event_p, ClientContext &context,
+                           shared_ptr<BFSState> &state, idx_t worker_id,
+                           const PhysicalOperator &op_p);
 
   TaskExecutionResult ExecuteTask(TaskExecutionMode mode) override;
 private:

test/sql/path_finding/parallel_path_length.test

@@ -9,7 +9,7 @@ statement ok
 select setseed(0.42);
 
 statement ok
-set disabled_optimizers = 'compressed_materialization';
+set threads=1;
 
 query I
 SELECT current_setting('experimental_path_finding_operator');