perf(algo,llm): run in-memory compute and LLM extraction tasks on the waiting thread

algo/src/common/in_mem_gds_utils.cpp

@@ -32,8 +32,9 @@ void InMemGDSUtils::runParallelCompute(InMemParallelCompute& vc, common::offset_
     const auto task =
         std::make_shared<InMemParallelComputeTask>(maxThreads, vc, sharedState, tableId);
     sharedState->morselDispatcher.init(maxOffset);
-    common::TaskScheduler::Get(*clientContext)
-        ->scheduleTaskAndWaitOrError(task, context, true /* launchNewWorkerThread */);
+    // Algorithms run inside a task, so this thread holds a scheduler slot and works on the task
+    // it waits for rather than starting a thread to take its place.
+    common::TaskScheduler::Get(*clientContext)->scheduleTaskAndWaitOrError(task, context);
 }
 } // namespace algo_extension
 } // namespace lbug

llm/src/function/ai_extract.cpp

@@ -199,8 +199,10 @@ void execFunc(const std::vector<std::shared_ptr<ValueVector>>& parameters,
         {4, clientContext->getMaxNumThreadForExec(), static_cast<uint64_t>(jobs.size())});
     auto task = std::make_shared<ExtractTask>(std::max<uint64_t>(1, workerCount), jobs, bindData,
         clientContext, bindData.apiKey);
-    TaskScheduler::Get(*clientContext)
-        ->scheduleTaskAndWaitOrError(task, clientContext, true /* launchNewWorkerThread */);
+    // A thread running the query holds a scheduler slot and works on the task it waits for. A
+    // call folded at bind time usually holds none; it runs its one-row task in a free slot or
+    // waits for a worker.
+    TaskScheduler::Get(*clientContext)->scheduleTaskAndWaitOrError(task, clientContext);
 
     result.resetAuxiliaryBuffer();
     for (const auto& job : jobs) {

llm/src/function/ai_extract_record.cpp

@@ -468,7 +468,10 @@ void execFunc(const std::vector<std::shared_ptr<ValueVector>>& parameters,
         std::min<uint64_t>({4, ctx->getMaxNumThreadForExec(), static_cast<uint64_t>(jobs.size())});
     auto task = std::make_shared<ExtractRecordTask>(std::max<uint64_t>(1, workerCount), jobs,
         bindData, ctx, apiKey);
-    TaskScheduler::Get(*ctx)->scheduleTaskAndWaitOrError(task, ctx, true);
+    // A thread running the query holds a scheduler slot and works on the task it waits for. A
+    // call folded at bind time usually holds none; it runs its one-row task in a free slot or
+    // waits for a worker.
+    TaskScheduler::Get(*ctx)->scheduleTaskAndWaitOrError(task, ctx);
 
     // Write results back into the STRUCT child vectors.
     result.resetAuxiliaryBuffer();