fix: fix memory leak when cancel segcore task (#29431)

#29430

Signed-off-by: luzhang <luzhang@zilliz.com>
Co-authored-by: luzhang <luzhang@zilliz.com>

internal/core/src/exec/Driver.h

@@ -207,6 +207,11 @@ class Driver : public std::enable_shared_from_this<Driver> {
     Init(std::unique_ptr<DriverContext> driver_ctx,
          std::vector<std::unique_ptr<Operator>> operators);
 
+    void
+    CloseByTask() {
+        Close();
+    }
+
  private:
     Driver() = default;
 

internal/core/src/exec/Task.cpp

@@ -129,6 +129,13 @@ Task::CreateDriversLocked(std::shared_ptr<Task>& self,
     }
 }
 
+void
+Task::Terminate(TaskState state) {
+    for (auto& driver : drivers_) {
+        driver->CloseByTask();
+    }
+}
+
 RowVectorPtr
 Task::Next(ContinueFuture* future) {
     // NOTE: Task::Next is single-threaded execution

internal/core/src/exec/Task.h

@@ -149,8 +149,7 @@ class Task : public std::enable_shared_from_this<Task> {
     }
 
     void
-    Terminate(TaskState state) {
-    }
+    Terminate(TaskState state);
 
     std::exception_ptr
     error() const {
@@ -167,6 +166,11 @@ class Task : public std::enable_shared_from_this<Task> {
         num_finished_drivers_++;
     }
 
+    void
+    RequestCancel() {
+        Terminate(TaskState::kCanceled);
+    }
+
  private:
     std::string uuid_;
 

internal/core/src/query/visitors/ExecPlanNodeVisitor.cpp

@@ -116,6 +116,7 @@ ExecPlanNodeVisitor::ExecuteExprNodeInternal(
                 if (cache_offset_vec->size() == 0) {
                     auto active_count = segment->get_active_count(timestamp_);
                     bitset_holder.resize(active_count);
+                    task->RequestCancel();
                     break;
                 }
                 auto cache_offset_vec_ptr =