temp

infiniflow · Dec 26, 2023 · 5057481 · 5057481
1 parent 536967a
commit 5057481
Show file tree

Hide file tree

Showing 71 changed files with 560 additions and 47 deletions.
diff --git a/python/hello_infinity.py b/python/hello_infinity.py
@@ -19,18 +19,12 @@
 def main():
     infinity_obj = infinity.connect(REMOTE_HOST)
     db = infinity_obj.get_database("default")
-    db.drop_table("my_table", if_exists=True)
+    db.drop_table("my_table2", if_exists=True)
     table = db.create_table(
-        "my_table", {"num": "integer", "body": "varchar", "vec": "vector,5,float"}, None)
-    table.insert(
-        [{"num": 1, "body": "undesirable, unnecessary, and harmful", "vec": [1.0] * 5}])
-    table.insert(
-        [{"num": 2, "body": "publisher=US National Office for Harmful Algal Blooms", "vec": [4.0] * 5}])
-    table.insert(
-        [{"num": 3, "body": "in the case of plants, growth and chemical", "vec": [7.0] * 5}])
-
-    res = table.output(["*"]).knn("vec", [3.0] * 5, "float", "ip", 2).to_pl()
-    print(res)
+        "my_table2", {"c1": "integer", "c2": "float"}, None)
+    for i in range(8):
+        table.insert(
+            [{"c1": i, "c2": float(i)}])
 
 
 if __name__ == '__main__':

diff --git a/src/executor/fragment_builder.cpp b/src/executor/fragment_builder.cpp
@@ -168,7 +168,7 @@ void FragmentBuilder::BuildFragments(PhysicalOperator *phys_op, PlanFragment *cu
             break;
         }
         case PhysicalOperatorType::kFusion:
-        case PhysicalOperatorType::kMergeParallelAggregate:
+        case PhysicalOperatorType::kMergeAggregate:
         case PhysicalOperatorType::kMergeHash:
         case PhysicalOperatorType::kMergeLimit:
         case PhysicalOperatorType::kMergeTop:

diff --git a/src/executor/operator/physcial_drop_view.cppm b/src/executor/operator/physcial_drop_view.cppm
@@ -21,6 +21,7 @@ import operator_state;
 import physical_operator;
 import physical_operator_type;
 import load_meta;
+import infinity_exception;
 
 export module physical_drop_view;
 
@@ -35,15 +36,19 @@ public:
                               u64 id,
                               SharedPtr<Vector<LoadMeta>> load_metas)
         : PhysicalOperator(PhysicalOperatorType::kDropView, nullptr, nullptr, id, load_metas), schema_name_(Move(schema_name)),
-          view_name_(Move(view_name)), conflict_type_(conflict_type), output_names_(Move(output_names)),
-          output_types_(Move(output_types)) {}
+          view_name_(Move(view_name)), conflict_type_(conflict_type), output_names_(Move(output_names)), output_types_(Move(output_types)) {}
 
     ~PhysicalDropView() override = default;
 
     void Init() override;
 
     bool Execute(QueryContext *query_context, OperatorState *operator_state) final;
 
+    SizeT TaskletCount() override {
+        Error<NotImplementException>("TaskletCount not Implement");
+        return 0;
+    }
+
     inline SharedPtr<Vector<String>> GetOutputNames() const final { return output_names_; }
 
     inline SharedPtr<Vector<SharedPtr<DataType>>> GetOutputTypes() const final { return output_types_; }

diff --git a/src/executor/operator/physical_aggregate.cppm b/src/executor/operator/physical_aggregate.cppm
@@ -24,6 +24,7 @@ import data_table;
 import hash_table;
 import base_expression;
 import load_meta;
+import infinity_exception;
 
 export module physical_aggregate;
 
@@ -52,6 +53,11 @@ public:
 
     bool Execute(QueryContext *query_context, OperatorState *operator_state) final;
 
+    SizeT TaskletCount() override {
+        Error<NotImplementException>("TaskletCount not Implement");
+        return 0;
+    }
+
     void GroupByInputTable(const SharedPtr<DataTable> &input_table, SharedPtr<DataTable> &output_table);
 
     void GenerateGroupByResult(const SharedPtr<DataTable> &input_table, SharedPtr<DataTable> &output_table);

diff --git a/src/executor/operator/physical_alter.cppm b/src/executor/operator/physical_alter.cppm
@@ -21,6 +21,7 @@ import operator_state;
 import physical_operator;
 import physical_operator_type;
 import load_meta;
+import infinity_exception;
 
 export module physical_alter;
 
@@ -41,6 +42,11 @@ public:
 
     bool Execute(QueryContext *query_context, OperatorState *operator_state) final;
 
+    SizeT TaskletCount() override {
+        Error<NotImplementException>("TaskletCount not Implement");
+        return 0;
+    }
+
     inline SharedPtr<Vector<String>> GetOutputNames() const final { return output_names_; }
 
     inline SharedPtr<Vector<SharedPtr<DataType>>> GetOutputTypes() const final { return output_types_; }

diff --git a/src/executor/operator/physical_command.cppm b/src/executor/operator/physical_command.cppm
@@ -22,6 +22,7 @@ import query_context;
 import parser;
 import operator_state;
 import load_meta;
+import infinity_exception;
 
 export module physical_command;
 
@@ -43,6 +44,11 @@ public:
 
     bool Execute(QueryContext *query_context, OperatorState *operator_state) override;
 
+    SizeT TaskletCount() override {
+        Error<NotImplementException>("TaskletCount not Implement");
+        return 0;
+    }
+
     inline SharedPtr<Vector<String>> GetOutputNames() const override { return output_names_; }
 
     inline SharedPtr<Vector<SharedPtr<DataType>>> GetOutputTypes() const override { return output_types_; }

diff --git a/src/executor/operator/physical_create_collection.cppm b/src/executor/operator/physical_create_collection.cppm
@@ -21,6 +21,7 @@ import operator_state;
 import physical_operator;
 import physical_operator_type;
 import load_meta;
+import infinity_exception;
 
 export module physical_create_collection;
 
@@ -43,6 +44,11 @@ public:
 
     bool Execute(QueryContext *query_context, OperatorState *operator_state) final;
 
+    SizeT TaskletCount() override {
+        Error<NotImplementException>("TaskletCount not Implement");
+        return 0;
+    }
+
     inline SharedPtr<Vector<String>> GetOutputNames() const final { return output_names_; }
 
     inline SharedPtr<Vector<SharedPtr<DataType>>> GetOutputTypes() const final { return output_types_; }

diff --git a/src/executor/operator/physical_create_index.cppm b/src/executor/operator/physical_create_index.cppm
@@ -22,6 +22,7 @@ import physical_operator;
 import physical_operator_type;
 import index_def;
 import load_meta;
+import infinity_exception;
 
 export module physical_create_index;
 
@@ -32,6 +33,11 @@ public:
 
     bool Execute(QueryContext *query_context, OperatorState *operator_state) override;
 
+    SizeT TaskletCount() override {
+        Error<NotImplementException>("TaskletCount not Implement");
+        return 0;
+    }
+
     SharedPtr<Vector<String>> GetOutputNames() const override { return output_names_; }
 
     SharedPtr<Vector<SharedPtr<DataType>>> GetOutputTypes() const override { return output_types_; }

diff --git a/src/executor/operator/physical_create_schema.cppm b/src/executor/operator/physical_create_schema.cppm
@@ -22,6 +22,7 @@ import physical_operator;
 import physical_operator_type;
 import index_def;
 import load_meta;
+import infinity_exception;
 
 export module physical_create_schema;
 
@@ -44,6 +45,11 @@ public:
 
     bool Execute(QueryContext *query_context, OperatorState *operator_state) final;
 
+    SizeT TaskletCount() override {
+        Error<NotImplementException>("TaskletCount not Implement");
+        return 0;
+    }
+
     inline SharedPtr<Vector<String>> GetOutputNames() const final { return output_names_; }
 
     inline SharedPtr<Vector<SharedPtr<DataType>>> GetOutputTypes() const final { return output_types_; }

diff --git a/src/executor/operator/physical_create_table.cppm b/src/executor/operator/physical_create_table.cppm
@@ -23,6 +23,7 @@ import physical_operator_type;
 import index_def;
 import table_def;
 import load_meta;
+import infinity_exception;
 
 export module physical_create_table;
 
@@ -54,6 +55,10 @@ public:
 
     bool Execute(QueryContext *query_context, OperatorState *operator_state) final;
 
+    SizeT TaskletCount() override {
+        Error<NotImplementException>("TaskletCount not Implement");
+        return 0;
+    }
     inline SharedPtr<Vector<String>> GetOutputNames() const final { return output_names_; }
 
     inline SharedPtr<Vector<SharedPtr<DataType>>> GetOutputTypes() const final { return output_types_; }

diff --git a/src/executor/operator/physical_create_view.cppm b/src/executor/operator/physical_create_view.cppm
@@ -22,6 +22,7 @@ import physical_operator;
 import physical_operator_type;
 import index_def;
 import load_meta;
+import infinity_exception;
 
 export module physical_create_view;
 
@@ -43,6 +44,11 @@ public:
 
     bool Execute(QueryContext *query_context, OperatorState *operator_state) final;
 
+    SizeT TaskletCount() override {
+        Error<NotImplementException>("TaskletCount not Implement");
+        return 0;
+    }
+
     inline const SharedPtr<CreateViewInfo> &bound_select_statement() const { return create_view_info_; };
 
     inline SharedPtr<Vector<String>> GetOutputNames() const final { return output_names_; }

diff --git a/src/executor/operator/physical_cross_product.cppm b/src/executor/operator/physical_cross_product.cppm
@@ -22,6 +22,7 @@ import physical_operator;
 import physical_operator_type;
 import data_table;
 import load_meta;
+import infinity_exception;
 
 export module physical_cross_product;
 
@@ -38,6 +39,11 @@ public:
 
     bool Execute(QueryContext *query_context, OperatorState *operator_state) final;
 
+    SizeT TaskletCount() override {
+        Error<NotImplementException>("TaskletCount not Implement");
+        return 0;
+    }
+
     SharedPtr<Vector<String>> GetOutputNames() const final;
 
     SharedPtr<Vector<SharedPtr<DataType>>> GetOutputTypes() const final;

diff --git a/src/executor/operator/physical_delete.cppm b/src/executor/operator/physical_delete.cppm
@@ -22,6 +22,7 @@ import physical_operator;
 import physical_operator_type;
 import table_collection_entry;
 import load_meta;
+import infinity_exception;
 
 export module physical_delete;
 
@@ -38,6 +39,11 @@ public:
 
     bool Execute(QueryContext *query_context, OperatorState *operator_state) final;
 
+    SizeT TaskletCount() override {
+        Error<NotImplementException>("TaskletCount not Implement");
+        return 0;
+    }
+
     inline SharedPtr<Vector<String>> GetOutputNames() const final { return output_names_; }
 
     inline SharedPtr<Vector<SharedPtr<DataType>>> GetOutputTypes() const final { return output_types_; }

diff --git a/src/executor/operator/physical_drop_collection.cppm b/src/executor/operator/physical_drop_collection.cppm
@@ -21,6 +21,7 @@ import operator_state;
 import physical_operator;
 import physical_operator_type;
 import load_meta;
+import infinity_exception;
 
 export module physical_drop_collection;
 
@@ -42,6 +43,11 @@ public:
 
     bool Execute(QueryContext *query_context, OperatorState *operator_state) final;
 
+    SizeT TaskletCount() override {
+        Error<NotImplementException>("TaskletCount not Implement");
+        return 0;
+    }
+
     inline SharedPtr<Vector<String>> GetOutputNames() const final { return output_names_; }
 
     inline SharedPtr<Vector<SharedPtr<DataType>>> GetOutputTypes() const final { return output_types_; }

diff --git a/src/executor/operator/physical_drop_index.cppm b/src/executor/operator/physical_drop_index.cppm
@@ -22,6 +22,7 @@ import query_context;
 import parser;
 import operator_state;
 import load_meta;
+import infinity_exception;
 
 export module physical_drop_index;
 
@@ -47,6 +48,11 @@ public:
 
     bool Execute(QueryContext *query_context, OperatorState *operator_state) override;
 
+    SizeT TaskletCount() override {
+        Error<NotImplementException>("TaskletCount not Implement");
+        return 0;
+    }
+
     inline SharedPtr<Vector<String>> GetOutputNames() const override { return output_names_; }
 
     inline SharedPtr<Vector<SharedPtr<DataType>>> GetOutputTypes() const override { return output_types_; }

diff --git a/src/executor/operator/physical_drop_schema.cppm b/src/executor/operator/physical_drop_schema.cppm
@@ -21,6 +21,7 @@ import operator_state;
 import physical_operator;
 import physical_operator_type;
 import load_meta;
+import infinity_exception;
 
 export module physical_drop_schema;
 
@@ -43,6 +44,11 @@ public:
 
     bool Execute(QueryContext *query_context, OperatorState *operator_state) final;
 
+    SizeT TaskletCount() override {
+        Error<NotImplementException>("TaskletCount not Implement");
+        return 0;
+    }
+
     inline SharedPtr<Vector<String>> GetOutputNames() const final { return output_names_; }
 
     inline SharedPtr<Vector<SharedPtr<DataType>>> GetOutputTypes() const final { return output_types_; }

diff --git a/src/executor/operator/physical_drop_table.cppm b/src/executor/operator/physical_drop_table.cppm
@@ -21,6 +21,7 @@ import operator_state;
 import physical_operator;
 import physical_operator_type;
 import load_meta;
+import infinity_exception;
 
 export module physical_drop_table;
 
@@ -45,6 +46,11 @@ public:
 
     bool Execute(QueryContext *query_context, OperatorState *operator_state) final;
 
+    SizeT TaskletCount() override {
+        Error<NotImplementException>("TaskletCount not Implement");
+        return 0;
+    }
+
     inline SharedPtr<Vector<String>> GetOutputNames() const final { return output_names_; }
 
     inline SharedPtr<Vector<SharedPtr<DataType>>> GetOutputTypes() const final { return output_types_; }

diff --git a/src/executor/operator/physical_dummy_operator.cppm b/src/executor/operator/physical_dummy_operator.cppm
@@ -21,6 +21,7 @@ import operator_state;
 import physical_operator;
 import physical_operator_type;
 import load_meta;
+import infinity_exception;
 
 export module physical_dummy_operator;
 
@@ -36,6 +37,11 @@ public:
 
     bool Execute(QueryContext *query_context, OperatorState *operator_state) final;
 
+    SizeT TaskletCount() override {
+        Error<NotImplementException>("TaskletCount not Implement");
+        return 0;
+    }
+
     inline SharedPtr<Vector<String>> GetOutputNames() const final { return output_names_; }
 
     inline SharedPtr<Vector<SharedPtr<DataType>>> GetOutputTypes() const final { return output_types_; }

diff --git a/src/executor/operator/physical_dummy_scan.cppm b/src/executor/operator/physical_dummy_scan.cppm
@@ -21,14 +21,16 @@ import operator_state;
 import physical_operator;
 import physical_operator_type;
 import load_meta;
+import infinity_exception;
 
 export module physical_dummy_scan;
 
 namespace infinity {
 
 export class PhysicalDummyScan final : public PhysicalOperator {
 public:
-    explicit PhysicalDummyScan(u64 id, SharedPtr<Vector<LoadMeta>> load_metas) : PhysicalOperator(PhysicalOperatorType::kDummyScan, nullptr, nullptr, id, load_metas) {}
+    explicit PhysicalDummyScan(u64 id, SharedPtr<Vector<LoadMeta>> load_metas)
+        : PhysicalOperator(PhysicalOperatorType::kDummyScan, nullptr, nullptr, id, load_metas) {}
 
     ~PhysicalDummyScan() override = default;
 
@@ -40,6 +42,11 @@ public:
 
     inline SharedPtr<Vector<SharedPtr<DataType>>> GetOutputTypes() const final { return output_types_; }
 
+    SizeT TaskletCount() override {
+        Error<NotImplementException>("TaskletCount not Implement");
+        return 0;
+    }
+
 private:
     SharedPtr<Vector<String>> output_names_{};
     SharedPtr<Vector<SharedPtr<DataType>>> output_types_{};