migrate ddl to task schedular framework

kuzudb · Jan 11, 2023 · 5a4d0c6 · 5a4d0c6
1 parent 5754eaf
commit 5a4d0c6
Show file tree

Hide file tree

Showing 27 changed files with 182 additions and 93 deletions.
diff --git a/src/binder/bind/bind_ddl.cpp b/src/binder/bind/bind_ddl.cpp
@@ -22,7 +22,7 @@ unique_ptr<BoundStatement> Binder::bindCreateNodeClause(const Statement& stateme
     auto primaryKeyIdx = bindPrimaryKey(
         createNodeClause.getPKColName(), createNodeClause.getPropertyNameDataTypes());
     return make_unique<BoundCreateNodeClause>(
-        tableName, move(boundPropertyNameDataTypes), primaryKeyIdx);
+        tableName, std::move(boundPropertyNameDataTypes), primaryKeyIdx);
 }
 
 unique_ptr<BoundStatement> Binder::bindCreateRelClause(const Statement& statement) {
@@ -40,7 +40,7 @@ unique_ptr<BoundStatement> Binder::bindCreateRelClause(const Statement& statemen
         srcDstTableIDs.emplace_back(bindNodeTableID(srcTableName), bindNodeTableID(dstTableName));
     }
     return make_unique<BoundCreateRelClause>(
-        tableName, move(propertyNameDataTypes), relMultiplicity, srcDstTableIDs);
+        tableName, std::move(propertyNameDataTypes), relMultiplicity, srcDstTableIDs);
 }
 
 unique_ptr<BoundStatement> Binder::bindDropTable(const Statement& statement) {

diff --git a/src/include/binder/bound_statement_result.h b/src/include/binder/bound_statement_result.h
@@ -43,6 +43,12 @@ class BoundStatementResult {
         return result;
     }
 
+    inline shared_ptr<Expression> getSingleExpressionToCollect() {
+        auto expressionsToCollect = getExpressionsToCollect();
+        assert(expressionsToCollect.size() == 1);
+        return expressionsToCollect[0];
+    }
+
     inline unique_ptr<BoundStatementResult> copy() {
         return make_unique<BoundStatementResult>(columns, expressionsToCollectPerColumn);
     }

diff --git a/src/include/planner/logical_plan/logical_operator/logical_create_node_table.h b/src/include/planner/logical_plan/logical_operator/logical_create_node_table.h
@@ -8,15 +8,16 @@ namespace planner {
 class LogicalCreateNodeTable : public LogicalCreateTable {
 public:
     LogicalCreateNodeTable(string tableName, vector<PropertyNameDataType> propertyNameDataTypes,
-        uint32_t primaryKeyIdx)
+        uint32_t primaryKeyIdx, shared_ptr<Expression> outputExpression)
         : LogicalCreateTable{LogicalOperatorType::CREATE_NODE_TABLE, std::move(tableName),
-              std::move(propertyNameDataTypes)},
+              std::move(propertyNameDataTypes), std::move(outputExpression)},
           primaryKeyIdx{primaryKeyIdx} {}
 
     inline uint32_t getPrimaryKeyIdx() const { return primaryKeyIdx; }
 
     inline unique_ptr<LogicalOperator> copy() override {
-        return make_unique<LogicalCreateNodeTable>(tableName, propertyNameDataTypes, primaryKeyIdx);
+        return make_unique<LogicalCreateNodeTable>(
+            tableName, propertyNameDataTypes, primaryKeyIdx, outputExpression);
     }
 
 private:

diff --git a/src/include/planner/logical_plan/logical_operator/logical_create_rel_table.h b/src/include/planner/logical_plan/logical_operator/logical_create_rel_table.h
@@ -9,9 +9,10 @@ namespace planner {
 class LogicalCreateRelTable : public LogicalCreateTable {
 public:
     LogicalCreateRelTable(string tableName, vector<PropertyNameDataType> propertyNameDataTypes,
-        RelMultiplicity relMultiplicity, vector<pair<table_id_t, table_id_t>> srcDstTableIDs)
+        RelMultiplicity relMultiplicity, vector<pair<table_id_t, table_id_t>> srcDstTableIDs,
+        shared_ptr<Expression> outputExpression)
         : LogicalCreateTable{LogicalOperatorType::CREATE_REL_TABLE, std::move(tableName),
-              std::move(propertyNameDataTypes)},
+              std::move(propertyNameDataTypes), std::move(outputExpression)},
           relMultiplicity{relMultiplicity}, srcDstTableIDs{std::move(srcDstTableIDs)} {}
 
     inline RelMultiplicity getRelMultiplicity() const { return relMultiplicity; }
@@ -20,7 +21,7 @@ class LogicalCreateRelTable : public LogicalCreateTable {
 
     inline unique_ptr<LogicalOperator> copy() override {
         return make_unique<LogicalCreateRelTable>(
-            tableName, propertyNameDataTypes, relMultiplicity, srcDstTableIDs);
+            tableName, propertyNameDataTypes, relMultiplicity, srcDstTableIDs, outputExpression);
     }
 
 private:

diff --git a/src/include/planner/logical_plan/logical_operator/logical_create_table.h b/src/include/planner/logical_plan/logical_operator/logical_create_table.h
@@ -8,9 +8,9 @@ namespace planner {
 class LogicalCreateTable : public LogicalDDL {
 public:
     LogicalCreateTable(LogicalOperatorType operatorType, string tableName,
-        vector<PropertyNameDataType> propertyNameDataTypes)
-        : LogicalDDL{operatorType, std::move(tableName)}, propertyNameDataTypes{
-                                                              std::move(propertyNameDataTypes)} {}
+        vector<PropertyNameDataType> propertyNameDataTypes, shared_ptr<Expression> outputExpression)
+        : LogicalDDL{operatorType, std::move(tableName), std::move(outputExpression)},
+          propertyNameDataTypes{std::move(propertyNameDataTypes)} {}
 
     inline vector<PropertyNameDataType> getPropertyNameDataTypes() const {
         return propertyNameDataTypes;

diff --git a/src/include/planner/logical_plan/logical_operator/logical_ddl.h b/src/include/planner/logical_plan/logical_operator/logical_ddl.h
@@ -7,17 +7,26 @@ namespace planner {
 
 class LogicalDDL : public LogicalOperator {
 public:
-    LogicalDDL(LogicalOperatorType operatorType, string tableName)
-        : LogicalOperator{operatorType}, tableName{std::move(tableName)} {}
+    LogicalDDL(
+        LogicalOperatorType operatorType, string tableName, shared_ptr<Expression> outputExpression)
+        : LogicalOperator{operatorType}, tableName{std::move(tableName)},
+          outputExpression{std::move(outputExpression)} {}
 
     inline string getTableName() const { return tableName; }
+    inline shared_ptr<Expression> getOutputExpression() const { return outputExpression; }
 
     inline string getExpressionsForPrinting() const override { return tableName; }
 
-    inline void computeSchema() override { schema = make_unique<Schema>(); }
+    inline void computeSchema() override {
+        schema = make_unique<Schema>();
+        auto groupPos = schema->createGroup();
+        schema->insertToGroupAndScope(outputExpression, groupPos);
+        schema->setGroupAsSingleState(groupPos);
+    }
 
 protected:
     string tableName;
+    shared_ptr<Expression> outputExpression;
 };
 
 } // namespace planner

diff --git a/src/include/planner/logical_plan/logical_operator/logical_drop_property.h b/src/include/planner/logical_plan/logical_operator/logical_drop_property.h
@@ -6,18 +6,19 @@ namespace kuzu {
 namespace planner {
 
 class LogicalDropProperty : public LogicalDDL {
-
 public:
-    explicit LogicalDropProperty(table_id_t tableID, property_id_t propertyID, string tableName)
-        : LogicalDDL{LogicalOperatorType::DROP_PROPERTY, std::move(tableName)}, tableID{tableID},
-          propertyID{propertyID} {}
+    explicit LogicalDropProperty(table_id_t tableID, property_id_t propertyID, string tableName,
+        shared_ptr<Expression> outputExpression)
+        : LogicalDDL{LogicalOperatorType::DROP_PROPERTY, std::move(tableName),
+              std::move(outputExpression)},
+          tableID{tableID}, propertyID{propertyID} {}
 
     inline table_id_t getTableID() const { return tableID; }
 
     inline property_id_t getPropertyID() const { return propertyID; }
 
     inline unique_ptr<LogicalOperator> copy() override {
-        return make_unique<LogicalDropProperty>(tableID, propertyID, tableName);
+        return make_unique<LogicalDropProperty>(tableID, propertyID, tableName, outputExpression);
     }
 
 private:

diff --git a/src/include/planner/logical_plan/logical_operator/logical_drop_table.h b/src/include/planner/logical_plan/logical_operator/logical_drop_table.h
@@ -6,15 +6,17 @@ namespace kuzu {
 namespace planner {
 
 class LogicalDropTable : public LogicalDDL {
-
 public:
-    explicit LogicalDropTable(table_id_t tableID, string tableName)
-        : LogicalDDL{LogicalOperatorType::DROP_TABLE, std::move(tableName)}, tableID{tableID} {}
+    explicit LogicalDropTable(
+        table_id_t tableID, string tableName, shared_ptr<Expression> outputExpression)
+        : LogicalDDL{LogicalOperatorType::DROP_TABLE, std::move(tableName),
+              std::move(outputExpression)},
+          tableID{tableID} {}
 
     inline table_id_t getTableID() const { return tableID; }
 
     inline unique_ptr<LogicalOperator> copy() override {
-        return make_unique<LogicalDropTable>(tableID, tableName);
+        return make_unique<LogicalDropTable>(tableID, tableName, outputExpression);
     }
 
 private:

diff --git a/src/include/processor/mapper/plan_mapper.h b/src/include/processor/mapper/plan_mapper.h
@@ -1,6 +1,7 @@
 #pragma once
 
 #include "binder/expression/node_expression.h"
+#include "common/statement_type.h"
 #include "planner/logical_plan/logical_plan.h"
 #include "processor/mapper/expression_mapper.h"
 #include "processor/operator/result_collector.h"

diff --git a/src/include/processor/operator/ddl/create_node_table.h b/src/include/processor/operator/ddl/create_node_table.h
@@ -9,23 +9,25 @@ namespace processor {
 class CreateNodeTable : public CreateTable {
 public:
     CreateNodeTable(Catalog* catalog, string tableName,
-        vector<PropertyNameDataType> propertyNameDataTypes, uint32_t primaryKeyIdx, uint32_t id,
-        const string& paramsString, NodesStatisticsAndDeletedIDs* nodesStatisticsAndDeletedIDs)
+        vector<PropertyNameDataType> propertyNameDataTypes, uint32_t primaryKeyIdx,
+        const DataPos& outputPos, uint32_t id, const string& paramsString,
+        NodesStatisticsAndDeletedIDs* nodesStatistics)
         : CreateTable{PhysicalOperatorType::CREATE_NODE_TABLE, catalog, std::move(tableName),
-              std::move(propertyNameDataTypes), id, paramsString},
-          primaryKeyIdx{primaryKeyIdx}, nodesStatisticsAndDeletedIDs{nodesStatisticsAndDeletedIDs} {
-    }
+              std::move(propertyNameDataTypes), outputPos, id, paramsString},
+          primaryKeyIdx{primaryKeyIdx}, nodesStatistics{nodesStatistics} {}
+
+    void executeDDLInternal() override;
 
-    string execute() override;
+    std::string getOutputMsg() override;
 
     unique_ptr<PhysicalOperator> clone() override {
         return make_unique<CreateNodeTable>(catalog, tableName, propertyNameDataTypes,
-            primaryKeyIdx, id, paramsString, nodesStatisticsAndDeletedIDs);
+            primaryKeyIdx, outputPos, id, paramsString, nodesStatistics);
     }
 
 private:
     uint32_t primaryKeyIdx;
-    NodesStatisticsAndDeletedIDs* nodesStatisticsAndDeletedIDs;
+    NodesStatisticsAndDeletedIDs* nodesStatistics;
 };
 
 } // namespace processor

diff --git a/src/include/processor/operator/ddl/create_rel_table.h b/src/include/processor/operator/ddl/create_rel_table.h
@@ -10,18 +10,20 @@ class CreateRelTable : public CreateTable {
 public:
     CreateRelTable(Catalog* catalog, string tableName,
         vector<PropertyNameDataType> propertyNameDataTypes, RelMultiplicity relMultiplicity,
-        vector<pair<table_id_t, table_id_t>> srcDstTableIDs, uint32_t id,
+        vector<pair<table_id_t, table_id_t>> srcDstTableIDs, const DataPos& outputPos, uint32_t id,
         const string& paramsString, RelsStatistics* relsStatistics)
         : CreateTable{PhysicalOperatorType::CREATE_REL_TABLE, catalog, std::move(tableName),
-              std::move(propertyNameDataTypes), id, paramsString},
+              std::move(propertyNameDataTypes), outputPos, id, paramsString},
           relMultiplicity{relMultiplicity}, srcDstTableIDs{std::move(srcDstTableIDs)},
           relsStatistics{relsStatistics} {}
 
-    string execute() override;
+    void executeDDLInternal() override;
+
+    std::string getOutputMsg() override;
 
     unique_ptr<PhysicalOperator> clone() override {
         return make_unique<CreateRelTable>(catalog, tableName, propertyNameDataTypes,
-            relMultiplicity, srcDstTableIDs, id, paramsString, relsStatistics);
+            relMultiplicity, srcDstTableIDs, outputPos, id, paramsString, relsStatistics);
     }
 
 private:

diff --git a/src/include/processor/operator/ddl/create_table.h b/src/include/processor/operator/ddl/create_table.h
@@ -8,12 +8,17 @@ namespace processor {
 class CreateTable : public DDL {
 public:
     CreateTable(PhysicalOperatorType operatorType, Catalog* catalog, string tableName,
-        vector<PropertyNameDataType> propertyNameDataTypes, uint32_t id, const string& paramsString)
-        : DDL{operatorType, catalog, id, paramsString}, tableName{std::move(tableName)},
-          propertyNameDataTypes{move(propertyNameDataTypes)} {}
+        vector<PropertyNameDataType> propertyNameDataTypes, const DataPos& outputPos, uint32_t id,
+        const string& paramsString)
+        : DDL{operatorType, catalog, outputPos, id, paramsString}, tableName{std::move(tableName)},
+          propertyNameDataTypes{std::move(propertyNameDataTypes)} {}
 
     virtual ~CreateTable() override = default;
 
+    void executeDDLInternal() override = 0;
+
+    std::string getOutputMsg() override = 0;
+
 protected:
     string tableName;
     vector<PropertyNameDataType> propertyNameDataTypes;

diff --git a/src/include/processor/operator/ddl/ddl.h b/src/include/processor/operator/ddl/ddl.h
@@ -12,21 +12,28 @@ namespace processor {
 
 class DDL : public PhysicalOperator {
 public:
-    DDL(PhysicalOperatorType operatorType, Catalog* catalog, uint32_t id,
+    DDL(PhysicalOperatorType operatorType, Catalog* catalog, const DataPos& outputPos, uint32_t id,
         const string& paramsString)
-        : PhysicalOperator{operatorType, id, paramsString}, catalog{catalog} {}
+        : PhysicalOperator{operatorType, id, paramsString}, catalog{catalog}, outputPos{outputPos} {
+    }
     virtual ~DDL() override = default;
 
     inline bool isSource() const override { return true; }
 
-    virtual string execute() = 0;
+    void initLocalStateInternal(ResultSet* resultSet, ExecutionContext* context) override;
 
-    bool getNextTuplesInternal() override {
-        throw InternalException("getNextTupleInternal() should not be called on DDL operator.");
-    }
+    bool getNextTuplesInternal() override;
+
+protected:
+    virtual std::string getOutputMsg() = 0;
+    virtual void executeDDLInternal() = 0;
 
 protected:
     Catalog* catalog;
+    DataPos outputPos;
+    ValueVector* outputVector;
+
+    bool hasExecuted = false;
 };
 
 } // namespace processor

diff --git a/src/include/processor/operator/ddl/drop_property.h b/src/include/processor/operator/ddl/drop_property.h
@@ -11,15 +11,18 @@ namespace processor {
 class DropProperty : public DDL {
 public:
     DropProperty(Catalog* catalog, table_id_t tableID, property_id_t propertyID,
-        StorageManager& storageManager, uint32_t id, const string& paramsString)
-        : DDL{PhysicalOperatorType::DROP_PROPERTY, catalog, id, paramsString}, tableID{tableID},
-          propertyID{propertyID}, storageManager{storageManager} {}
+        StorageManager& storageManager, const DataPos& outputPos, uint32_t id,
+        const string& paramsString)
+        : DDL{PhysicalOperatorType::DROP_PROPERTY, catalog, outputPos, id, paramsString},
+          tableID{tableID}, propertyID{propertyID}, storageManager{storageManager} {}
 
-    string execute() override;
+    void executeDDLInternal() override;
+
+    std::string getOutputMsg() override;
 
     unique_ptr<PhysicalOperator> clone() override {
         return make_unique<DropProperty>(
-            catalog, tableID, propertyID, storageManager, id, paramsString);
+            catalog, tableID, propertyID, storageManager, outputPos, id, paramsString);
     }
 
 protected:

diff --git a/src/include/processor/operator/ddl/drop_table.h b/src/include/processor/operator/ddl/drop_table.h
@@ -10,20 +10,18 @@ namespace processor {
 
 class DropTable : public DDL {
 public:
-    DropTable(Catalog* catalog, table_id_t tableID, StorageManager& storageManager, uint32_t id,
-        const string& paramsString)
-        : DDL{PhysicalOperatorType::DROP_TABLE, catalog, id, paramsString}, tableID{tableID},
-          storageManager{storageManager} {}
-
-    string execute() override {
-        auto tableSchema = catalog->getReadOnlyVersion()->getTableSchema(tableID);
-        catalog->removeTableSchema(tableSchema);
-        return StringUtils::string_format("%sTable: %s has been dropped.",
-            tableSchema->isNodeTable ? "Node" : "Rel", tableSchema->tableName.c_str());
-    }
+    DropTable(Catalog* catalog, table_id_t tableID, StorageManager& storageManager,
+        const DataPos& outputPos, uint32_t id, const string& paramsString)
+        : DDL{PhysicalOperatorType::DROP_TABLE, catalog, outputPos, id, paramsString},
+          tableID{tableID}, storageManager{storageManager} {}
+
+    void executeDDLInternal() override;
+
+    std::string getOutputMsg() override;
 
     unique_ptr<PhysicalOperator> clone() override {
-        return make_unique<DropTable>(catalog, tableID, storageManager, id, paramsString);
+        return make_unique<DropTable>(
+            catalog, tableID, storageManager, outputPos, id, paramsString);
     }
 
 protected:

diff --git a/src/include/processor/operator/physical_operator.h b/src/include/processor/operator/physical_operator.h
@@ -56,6 +56,7 @@ enum class PhysicalOperatorType : uint8_t {
 class PhysicalOperatorUtils {
 public:
     static std::string operatorTypeToString(PhysicalOperatorType operatorType);
+    static bool isDDLOperator(PhysicalOperatorType operatorType);
 };
 
 struct OperatorMetrics {

diff --git a/src/include/processor/physical_plan.h b/src/include/processor/physical_plan.h
@@ -2,24 +2,23 @@
 
 #include <memory>
 
-#include "common/statement_type.h"
 #include "processor/operator/physical_operator.h"
 
 namespace kuzu {
 namespace processor {
 
 class PhysicalPlan {
 public:
-    explicit PhysicalPlan(
-        unique_ptr<PhysicalOperator> lastOperator, common::StatementType statementType)
-        : lastOperator{std::move(lastOperator)}, statementType{statementType} {}
+    explicit PhysicalPlan(unique_ptr<PhysicalOperator> lastOperator)
+        : lastOperator{std::move(lastOperator)} {}
 
-    inline bool isDDL() const { return StatementTypeUtils::isDDL(statementType); }
-    inline bool isCopyCSV() const { return StatementTypeUtils::isCopyCSV(statementType); }
+    inline bool isCopyCSV() const {
+        return lastOperator->getOperatorType() == PhysicalOperatorType::COPY_NODE_CSV ||
+               lastOperator->getOperatorType() == PhysicalOperatorType::COPY_REL_CSV;
+    }
 
 public:
     unique_ptr<PhysicalOperator> lastOperator;
-    common::StatementType statementType;
 };
 
 class PhysicalPlanUtil {