Parallel create index

src/executor/fragment_builder.cpp

@@ -229,8 +229,8 @@ void FragmentBuilder::BuildFragments(PhysicalOperator *phys_op, PlanFragment *cu
             if (phys_op->left() != nullptr or phys_op->right() != nullptr) {
                 Error<SchedulerException>(Format("{} shouldn't have child.", phys_op->GetName()));
             }
-            PhysicalKnnScan* knn_scan = static_cast<PhysicalKnnScan*>(phys_op);
-            if(knn_scan->TaskCount() == 1) {
+            PhysicalKnnScan *knn_scan = static_cast<PhysicalKnnScan *>(phys_op);
+            if (knn_scan->TaskCount() == 1) {
                 current_fragment_ptr->SetFragmentType(FragmentType::kSerialMaterialize);
             } else {
                 current_fragment_ptr->SetFragmentType(FragmentType::kParallelMaterialize);
@@ -262,6 +262,51 @@ void FragmentBuilder::BuildFragments(PhysicalOperator *phys_op, PlanFragment *cu
             }
             return;
         }
+        case PhysicalOperatorType::kCreateIndexPrepare: {
+            if (phys_op->left() != nullptr || phys_op->right() != nullptr) {
+                Error<SchedulerException>(Format("Invalid input node of {}", phys_op->GetName()));
+            }
+            current_fragment_ptr->AddOperator(phys_op);
+            current_fragment_ptr->SetFragmentType(FragmentType::kSerialMaterialize);
+            current_fragment_ptr->SetSourceNode(query_context_ptr_, SourceType::kEmpty, phys_op->GetOutputNames(), phys_op->GetOutputTypes());
+            return;
+        }
+        case PhysicalOperatorType::kCreateIndexDo: {
+            if (phys_op->left() == nullptr || phys_op->right() != nullptr) {
+                Error<SchedulerException>(Format("Invalid input node of {}", phys_op->GetName()));
+            }
+            current_fragment_ptr->AddOperator(phys_op);
+            current_fragment_ptr->SetFragmentType(FragmentType::kParallelMaterialize);
+            current_fragment_ptr->SetSourceNode(query_context_ptr_, SourceType::kLocalQueue, phys_op->GetOutputNames(), phys_op->GetOutputTypes());
+
+            auto next_plan_fragment = MakeUnique<PlanFragment>(GetFragmentId());
+            next_plan_fragment->SetSinkNode(query_context_ptr_,
+                                            SinkType::kLocalQueue,
+                                            phys_op->left()->GetOutputNames(),
+                                            phys_op->left()->GetOutputTypes());
+            BuildFragments(phys_op->left(), next_plan_fragment.get());
+
+            current_fragment_ptr->AddChild(Move(next_plan_fragment));
+            return;
+        }
+        case PhysicalOperatorType::kCreateIndexFinish: {
+            if (phys_op->left() == nullptr || phys_op->right() != nullptr) {
+                Error<SchedulerException>(Format("Invalid input node of {}", phys_op->GetName()));
+            }
+            current_fragment_ptr->AddOperator(phys_op);
+            current_fragment_ptr->SetFragmentType(FragmentType::kSerialMaterialize);
+            current_fragment_ptr->SetSourceNode(query_context_ptr_, SourceType::kLocalQueue, phys_op->GetOutputNames(), phys_op->GetOutputTypes());
+
+            auto next_plan_fragment = MakeUnique<PlanFragment>(GetFragmentId());
+            next_plan_fragment->SetSinkNode(query_context_ptr_,
+                                            SinkType::kLocalQueue,
+                                            phys_op->left()->GetOutputNames(),
+                                            phys_op->left()->GetOutputTypes());
+            BuildFragments(phys_op->left(), next_plan_fragment.get());
+
+            current_fragment_ptr->AddChild(Move(next_plan_fragment));
+            return;
+        }
         default: {
             LOG_ERROR(Format("Invalid operator type: {} in Fragment Builder", phys_op->GetName()));
             break;

src/executor/operator/physical_create_index.cpp

@@ -33,7 +33,7 @@ void PhysicalCreateIndex::Init() {}
 
 bool PhysicalCreateIndex::Execute(QueryContext *query_context, OperatorState *operator_state) {
     auto *txn = query_context->GetTxn();
-    Status status = txn->CreateIndex(*schema_name_, *table_name_, index_def_ptr_, conflict_type_);
+    Status status = txn->CreateIndex(*schema_name_, *table_name_, index_def_ptr_, conflict_type_, false);
     if (!status.ok()) {
         operator_state->error_message_ = Move(status.msg_);
     }

src/executor/operator/physical_create_index_do.cpp

@@ -0,0 +1,76 @@
+// Copyright(C) 2023 InfiniFlow, Inc. All rights reserved.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+module;
+
+import stl;
+import parser;
+import catalog;
+import physical_operator_type;
+import physical_operator;
+import query_context;
+import operator_state;
+import load_meta;
+
+import index_def;
+import create_index_data;
+import base_table_ref;
+import status;
+import infinity_exception;
+import buffer_handle;
+import index_hnsw;
+import index_base;
+import hnsw_common;
+import dist_func_l2;
+import dist_func_ip;
+import hnsw_alg;
+import lvq_store;
+import plain_store;
+import buffer_manager;
+import txn_store;
+import third_party;
+import logger;
+
+module physical_create_index_do;
+
+namespace infinity {
+PhysicalCreateIndexDo::PhysicalCreateIndexDo(u64 id,
+                                             UniquePtr<PhysicalOperator> left,
+                                             SharedPtr<BaseTableRef> base_table_ref,
+                                             SharedPtr<String> index_name,
+                                             SharedPtr<Vector<String>> output_names,
+                                             SharedPtr<Vector<SharedPtr<DataType>>> output_types,
+                                             SharedPtr<Vector<LoadMeta>> load_metas)
+    : PhysicalOperator(PhysicalOperatorType::kCreateIndexDo, Move(left), nullptr, id, load_metas), base_table_ref_(base_table_ref),
+      index_name_(index_name), output_names_(output_names), output_types_(output_types) {}
+
+void PhysicalCreateIndexDo::Init() {}
+
+// FIXME: fetch and add a block one time
+bool PhysicalCreateIndexDo::Execute(QueryContext *query_context, OperatorState *operator_state) {
+    auto *txn = query_context->GetTxn();
+    auto *create_index_do_state = static_cast<CreateIndexDoOperatorState *>(operator_state);
+    auto &create_index_idxes = create_index_do_state->create_index_shared_data_->create_index_idxes_;
+
+    auto status = txn->CreateIndexDo(*base_table_ref_->schema_name(), *base_table_ref_->table_name(), *index_name_, create_index_idxes);
+    if (!status.ok()) {
+        operator_state->error_message_ = Move(status.msg_);
+        return false;
+    }
+    operator_state->SetComplete();
+
+    return true;
+}
+
+}; // namespace infinity

src/executor/operator/physical_create_index_do.cppm

@@ -0,0 +1,62 @@
+// Copyright(C) 2023 InfiniFlow, Inc. All rights reserved.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+module;
+
+import stl;
+import parser;
+import physical_operator_type;
+import physical_operator;
+import query_context;
+import operator_state;
+import load_meta;
+
+import index_def;
+import base_table_ref;
+
+export module physical_create_index_do;
+
+namespace infinity {
+
+export class PhysicalCreateIndexDo : public PhysicalOperator {
+public:
+    PhysicalCreateIndexDo(u64 id,
+                          UniquePtr<PhysicalOperator> left,
+                          SharedPtr<BaseTableRef> base_table_ref,
+                          SharedPtr<String> index_name,
+                          SharedPtr<Vector<String>> output_names,
+                          SharedPtr<Vector<SharedPtr<DataType>>> output_types,
+                          SharedPtr<Vector<LoadMeta>> load_metas);
+
+public:
+    void Init() override;
+
+    bool Execute(QueryContext *query_context, OperatorState *operator_state) override;
+
+    SizeT TaskletCount() override { return 0; }
+
+    SharedPtr<Vector<String>> GetOutputNames() const override { return output_names_; }
+
+    SharedPtr<Vector<SharedPtr<DataType>>> GetOutputTypes() const override { return output_types_; }
+
+public:
+    // for create fragemnt context
+    const SharedPtr<BaseTableRef> base_table_ref_{};
+    const SharedPtr<String> index_name_{};
+
+    const SharedPtr<Vector<String>> output_names_{};
+    const SharedPtr<Vector<SharedPtr<DataType>>> output_types_{};
+};
+
+} // namespace infinity

src/executor/operator/physical_create_index_finish.cpp

@@ -0,0 +1,56 @@
+// Copyright(C) 2023 InfiniFlow, Inc. All rights reserved.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+module;
+
+import stl;
+import parser;
+import physical_operator_type;
+import physical_operator;
+import query_context;
+import operator_state;
+import load_meta;
+import index_def;
+import wal;
+
+module physical_create_index_finish;
+
+namespace infinity {
+PhysicalCreateIndexFinish::PhysicalCreateIndexFinish(u64 id,
+                                                     UniquePtr<PhysicalOperator> left,
+                                                     SharedPtr<String> db_name,
+                                                     SharedPtr<String> table_name,
+                                                     SharedPtr<IndexDef> index_def,
+                                                     SharedPtr<Vector<String>> output_names,
+                                                     SharedPtr<Vector<SharedPtr<DataType>>> output_types,
+                                                     SharedPtr<Vector<LoadMeta>> load_metas)
+    : PhysicalOperator(PhysicalOperatorType::kCreateIndexFinish, Move(left), nullptr, id, load_metas), db_name_(db_name), table_name_(table_name),
+      index_def_(index_def), output_names_(output_names), output_types_(output_types) {}
+
+void PhysicalCreateIndexFinish::Init() {}
+
+bool PhysicalCreateIndexFinish::Execute(QueryContext *query_context, OperatorState *operator_state) {
+    auto *txn = query_context->GetTxn();
+    auto *create_index_finish_op_state = static_cast<CreateIndexFinishOperatorState *>(operator_state);
+
+    if (create_index_finish_op_state->input_complete_) {
+        txn->AddWalCmd(MakeShared<WalCmdCreateIndex>(*db_name_, *table_name_, index_def_));
+
+        operator_state->SetComplete();
+        return true;
+    }
+    return false;
+}
+
+} // namespace infinity

src/executor/operator/physical_create_index_finish.cppm

@@ -0,0 +1,60 @@
+// Copyright(C) 2023 InfiniFlow, Inc. All rights reserved.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+module;
+
+import stl;
+import parser;
+import physical_operator_type;
+import physical_operator;
+import query_context;
+import operator_state;
+import load_meta;
+import index_def;
+
+export module physical_create_index_finish;
+
+namespace infinity {
+export class PhysicalCreateIndexFinish : public PhysicalOperator {
+public:
+    PhysicalCreateIndexFinish(u64 id,
+                              UniquePtr<PhysicalOperator> left,
+                              SharedPtr<String> db_name,
+                              SharedPtr<String> table_name,
+                              SharedPtr<IndexDef> index_def,
+                              SharedPtr<Vector<String>> output_names,
+                              SharedPtr<Vector<SharedPtr<DataType>>> output_types,
+                              SharedPtr<Vector<LoadMeta>> load_metas);
+
+public:
+    void Init() override;
+
+    bool Execute(QueryContext *query_context, OperatorState *operator_state) override;
+
+    SizeT TaskletCount() override { return 1; }
+
+    SharedPtr<Vector<String>> GetOutputNames() const override { return output_names_; }
+
+    SharedPtr<Vector<SharedPtr<DataType>>> GetOutputTypes() const override { return output_types_; }
+
+public:
+    const SharedPtr<String> db_name_{};
+    const SharedPtr<String> table_name_{};
+    const SharedPtr<IndexDef> index_def_{};
+
+    const SharedPtr<Vector<String>> output_names_{};
+    const SharedPtr<Vector<SharedPtr<DataType>>> output_types_{};
+};
+
+} // namespace infinity