Add tmp schedule strategy.

src/executor/operator/physical_sink.cpp

@@ -131,19 +131,6 @@ void PhysicalSink::FillSinkStateFromLastOperatorState(MaterializeSinkState *mate
             }
             break;
         }
-        case PhysicalOperatorType::kKnnScan: {
-            throw ExecutorException("KnnScan shouldn't be here");
-            KnnScanOperatorState *knn_output_state = static_cast<KnnScanOperatorState *>(task_op_state);
-            if (knn_output_state->data_block_array_.empty()) {
-                Error<ExecutorException>("Empty knn scan output");
-            }
-
-            for (auto &data_block : knn_output_state->data_block_array_) {
-                materialize_sink_state->data_block_array_.emplace_back(Move(data_block));
-            }
-            knn_output_state->data_block_array_.clear();
-            break;
-        }
         case PhysicalOperatorType::kAggregate: {
             AggregateOperatorState *agg_output_state = static_cast<AggregateOperatorState *>(task_op_state);
             if (agg_output_state->data_block_array_.empty()) {

src/main/resource_manager.cppm

@@ -30,7 +30,7 @@ public:
         return cpu_count;
     }
 
-    inline u64 GetCpuResource() { return GetCpuResource(4); }
+    inline u64 GetCpuResource() { return GetCpuResource(Thread::hardware_concurrency()); }
 
     inline u64 GetMemoryResource(u64 memory_size) {
         total_memory_ -= memory_size;

src/scheduler/task_scheduler.cpp

@@ -30,6 +30,7 @@ import query_context;
 import plan_fragment;
 import fragment_context;
 import default_values;
+import physical_operator_type;
 
 module task_scheduler;
 
@@ -93,8 +94,20 @@ void TaskScheduler::Schedule(QueryContext *query_context, const Vector<FragmentT
     //    if the first op isn't SCAN op, fragment task source type is kQueue and a task_result_queue need to be created.
     //    According to the fragment output type to set the correct fragment task sink type.
     //    Set the queue of parent fragment task.
-    // ScheduleOneWorkerIfPossible(query_context, tasks, plan_fwosfasdfragment);
-    ScheduleRoundRobin(query_context, tasks, plan_fragment);
+    if (plan_fragment->GetOperators().empty()) {
+        Error<SchedulerException>("Empty fragment");
+    }
+    auto *last_operator = plan_fragment->GetOperators()[0];
+    switch (last_operator->operator_type()) {
+        case PhysicalOperatorType::kCreateIndexFinish: {
+            ScheduleRoundRobin(query_context, tasks, plan_fragment);
+            break;
+        }
+        default: {
+            ScheduleOneWorkerIfPossible(query_context, tasks, plan_fragment);
+            break;
+        }
+    }
 }
 
 void TaskScheduler::ScheduleOneWorkerPerQuery(QueryContext *query_context, const Vector<FragmentTask *> &tasks, PlanFragment *plan_fragment) {