Support DML,DDL to use inputNode (#103)

* Support DML,DDL to use inputNode

* address comment

* rebase upstream

* reabse upstream

* update

* address comments

* add test

Co-authored-by: dutor <440396+dutor@users.noreply.github.com>

src/context/ValidateContext.h

@@ -38,6 +38,26 @@ public:
         spaces_.emplace_back(std::move(space));
     }
 
+    const ColsDef& getVar(const std::string& var) const {
+        static const ColsDef kEmptyCols;
+        if (!existVar(var)) {
+            return kEmptyCols;
+        }
+        return vars_.at(var);
+    }
+
+    bool existVar(const std::string& var) const {
+        return vars_.find(var) != vars_.end();
+    }
+
+    void addSpace(const std::string &spaceName) {
+        createSpaces_.emplace(spaceName);
+    }
+
+    bool hasSpace(const std::string &spaceName) const {
+        return createSpaces_.find(spaceName) != createSpaces_.end();
+    }
+
     void registerVariable(std::string var, ColsDef cols) {
         vars_.emplace(std::move(var), std::move(cols));
     }
@@ -50,20 +70,21 @@ public:
         return spaces_.back();
     }
 
-    const ColsDef& getVar(const std::string& var) const {
-        static const ColsDef kEmptyCols;
-        if (!existVar(var)) {
-            return kEmptyCols;
-        }
-        return vars_.at(var);
+    AnnoVarGenerator* varGen() const {
+        return varGen_.get();
     }
 
-    bool existVar(const std::string& var) const {
-        return vars_.find(var) != vars_.end();
+    void addSchema(const std::string& name,
+                   const std::shared_ptr<const meta::NebulaSchemaProvider> &schema) {
+        schemas_.emplace(name, schema);
     }
 
-    AnnoVarGenerator* varGen() const {
-        return varGen_.get();
+    std::shared_ptr<const meta::NebulaSchemaProvider> getSchema(const std::string &name) const {
+        auto find = schemas_.find(name);
+        if (find == schemas_.end()) {
+            return std::shared_ptr<const meta::NebulaSchemaProvider>();
+        }
+        return find->second;
     }
 
 private:
@@ -72,6 +93,10 @@ private:
     // vars_ saves all named variable
     std::unordered_map<std::string, ColsDef>            vars_;
     std::unique_ptr<AnnoVarGenerator>                   varGen_;
+    using Schemas = std::unordered_map<std::string,
+          std::shared_ptr<const meta::NebulaSchemaProvider>>;
+    Schemas                                             schemas_;
+    std::unordered_set<std::string>                     createSpaces_;
 };
 }  // namespace graph
 }  // namespace nebula

src/exec/admin/SwitchSpaceExecutor.cpp

@@ -13,11 +13,23 @@ namespace nebula {
 namespace graph {
 
 folly::Future<Status> SwitchSpaceExecutor::execute() {
+    dumpLog();
+
     auto *spaceToNode = asNode<SwitchSpace>(node());
-    qctx_->rctx()->session()->setSpace(spaceToNode->getSpaceName(), spaceToNode->getSpaceId());
-    LOG(INFO) << "Graph space switched to `" << spaceToNode->getSpaceName()
-              << "', space id: " << spaceToNode->getSpaceId();
-    return start();
+    auto spaceName = spaceToNode->getSpaceName();
+    return qctx()->getMetaClient()->getSpace(spaceName)
+            .via(runner())
+            .then([spaceName, this](StatusOr<meta::cpp2::SpaceItem> resp) {
+                if (!resp.ok()) {
+                    LOG(ERROR) << resp.status();
+                    return resp.status();
+                }
+                auto spaceId = resp.value().get_space_id();
+                qctx_->rctx()->session()->setSpace(spaceName, spaceId);
+                LOG(INFO) << "Graph space switched to `" << spaceName
+                          << "', space id: " << spaceId;
+                return Status::OK();
+            });
 }
 
 }   // namespace graph

src/exec/maintain/CreateEdgeExecutor.cpp

@@ -17,9 +17,9 @@ folly::Future<Status> CreateEdgeExecutor::execute() {
 
 folly::Future<Status> CreateEdgeExecutor::createEdge() {
     dumpLog();
-
+    auto space = qctx_->rctx()->session()->space();
     auto *ceNode = asNode<CreateEdge>(node());
-    return qctx()->getMetaClient()->createEdgeSchema(ceNode->space(),
+    return qctx()->getMetaClient()->createEdgeSchema(space,
             ceNode->getName(), ceNode->getSchema(), ceNode->getIfNotExists())
         .via(runner())
         .then([](StatusOr<bool> resp) {

src/exec/maintain/CreateTagExecutor.cpp

@@ -18,8 +18,9 @@ folly::Future<Status> CreateTagExecutor::execute() {
 folly::Future<Status> CreateTagExecutor::createTag() {
     dumpLog();
 
+    auto space = qctx_->rctx()->session()->space();
     auto *ctNode = asNode<CreateTag>(node());
-    return qctx()->getMetaClient()->createTagSchema(ctNode->space(),
+    return qctx()->getMetaClient()->createTagSchema(space,
             ctNode->getName(), ctNode->getSchema(), ctNode->getIfNotExists())
         .via(runner())
         .then([](StatusOr<bool> resp) {

src/exec/maintain/DescEdgeExecutor.cpp

@@ -19,8 +19,9 @@ folly::Future<Status> DescEdgeExecutor::execute() {
 folly::Future<Status> DescEdgeExecutor::descEdge() {
     dumpLog();
 
+    auto space = qctx_->rctx()->session()->space();
     auto *deNode = asNode<DescEdge>(node());
-    return qctx()->getMetaClient()->getEdgeSchema(deNode->getSpaceId(), deNode->getName())
+    return qctx()->getMetaClient()->getEdgeSchema(space, deNode->getName())
             .via(runner())
             .then([this](StatusOr<meta::cpp2::Schema> resp) {
                 if (!resp.ok()) {

src/exec/maintain/DescTagExecutor.cpp

@@ -19,8 +19,9 @@ folly::Future<Status> DescTagExecutor::execute() {
 folly::Future<Status> DescTagExecutor::descTag() {
     dumpLog();
 
+    auto space = qctx_->rctx()->session()->space();
     auto *dtNode = asNode<DescTag>(node());
-    return qctx()->getMetaClient()->getTagSchema(dtNode->getSpaceId(), dtNode->getName())
+    return qctx()->getMetaClient()->getTagSchema(space, dtNode->getName())
             .via(runner())
             .then([this](StatusOr<meta::cpp2::Schema> resp) {
                 if (!resp.ok()) {

src/exec/mutate/InsertEdgesExecutor.cpp

@@ -22,8 +22,7 @@ folly::Future<Status> InsertEdgesExecutor::execute() {
 folly::Future<Status> InsertEdgesExecutor::insertEdges() {
     dumpLog();
     auto *ieNode = asNode<InsertEdges>(node());
-
-    return qctx()->getStorageClient()->addEdges(ieNode->space(),
+    return qctx()->getStorageClient()->addEdges(ieNode->getSpace(),
             ieNode->getEdges(), ieNode->getPropNames(), ieNode->getOverwritable())
         .via(runner())
         .then([this](storage::StorageRpcResponse<storage::cpp2::ExecResponse> resp) {

src/exec/mutate/InsertVerticesExecutor.cpp

@@ -22,7 +22,7 @@ folly::Future<Status> InsertVerticesExecutor::insertVertices() {
     dumpLog();
 
     auto *ivNode = asNode<InsertVertices>(node());
-    return qctx()->getStorageClient()->addVertices(ivNode->space(),
+    return qctx()->getStorageClient()->addVertices(ivNode->getSpace(),
             ivNode->getVertices(), ivNode->getPropNames(), ivNode->getOverwritable())
         .via(runner())
         .then([this](storage::StorageRpcResponse<storage::cpp2::ExecResponse> resp) {

src/mock/StorageCache.cpp

@@ -18,7 +18,7 @@ StorageCache::StorageCache(uint16_t metaPort) {
     meta::MetaClientOptions options;
     options.serviceName_ = "StorageCache";
     metaClient_ = std::make_unique<meta::MetaClient>(threadPool,
-            std::move(hostStatus).value(), options);
+                                                     std::move(hostStatus).value(), options);
     metaClient_->waitForMetadReady();
 
     mgr_ = std::make_unique<meta::ServerBasedSchemaManager>();
@@ -37,13 +37,18 @@ Status StorageCache::addVertices(const storage::cpp2::AddVerticesRequest& req) {
         spaceDataInfo = &cache_[spaceId];
     }
 
-    std::unordered_map<TagID, std::vector<std::string>> propNames = req.get_prop_names();
+    std::unordered_map<TagID, std::vector<std::string>> propNames;
+    for (auto &prop : req.get_prop_names()) {
+        propNames[prop.first] = prop.second;
+    }
+
     auto &parts = req.get_parts();
     for (auto &part : parts) {
         for (auto &vertex : part.second) {
             auto vId = vertex.get_id();
             auto findV = spaceDataInfo->vertices.find(vId);
-            std::unordered_map<TagID, PropertyInfo> *vertexInfo = nullptr;
+            std::unordered_map<TagID,
+                               std::unordered_map<std::string, Value>> *vertexInfo = nullptr;
             if (findV != spaceDataInfo->vertices.end()) {
                 vertexInfo = &findV->second;
             } else {
@@ -57,15 +62,12 @@ Status StorageCache::addVertices(const storage::cpp2::AddVerticesRequest& req) {
                 if (propValues.size() != propNames[tagId].size()) {
                     return Status::Error("Wrong size");
                 }
-
-                std::unordered_map<std::string, int32_t> propIndexes;
-                for (auto i = 0u; i < propValues.size(); i++) {
-                    propIndexes[propNames[tagId][i]] = i;
+                auto ret = getTagWholeValue(spaceId, tagId, propValues, propNames[tagId]);
+                if (!ret.ok()) {
+                    LOG(ERROR) << ret.status();
+                    return ret.status();
                 }
-
-                (*vertexInfo)[tagId].propNames = propNames[tagId];
-                (*vertexInfo)[tagId].propValues = std::move(propValues);
-                (*vertexInfo)[tagId].propIndexes = std::move(propIndexes);
+                (*vertexInfo)[tagId] = std::move(ret).value();
             }
         }
     }
@@ -88,57 +90,103 @@ Status StorageCache::addEdges(const storage::cpp2::AddEdgesRequest& req) {
     auto &parts = req.get_parts();
     for (auto &part : parts) {
         for (auto &edge : part.second) {
-            auto key = edge.get_key();
-            auto edgeKey = getEdgeKey(key.get_src(),
-                    key.get_edge_type(), key.get_ranking(), key.get_dst());
-            PropertyInfo propertyInfo;
-            propertyInfo.propNames = propNames;
-            for (auto i = 0u; i < propertyInfo.propNames.size(); i++) {
-                propertyInfo.propIndexes[propertyInfo.propNames[i]] = i;
-            }
-            propertyInfo.propValues = edge.get_props();
-            if (propertyInfo.propValues.size() != propertyInfo.propNames.size()) {
-                LOG(ERROR) << "Wrong size, propValues.size : " << propertyInfo.propValues.size()
-                           << ", propNames.size : " << propertyInfo.propNames.size();
+            storage::cpp2::EdgeKey edgeKey;
+            edgeKey.set_src(edge.key.get_src());
+            auto edgeType = edge.key.get_edge_type();
+            edgeKey.set_edge_type(edgeType);
+            edgeKey.set_ranking(edge.key.get_ranking());
+            edgeKey.set_dst(edge.key.get_dst());
+            auto propValues = edge.get_props();
+            if (propValues.size() != propNames.size()) {
+                LOG(ERROR) << "Wrong size, propValues.size : " << propValues.size()
+                           << ", propNames.size : " << propNames.size();
                 return Status::Error("Wrong size");
             }
-            spaceDataInfo->edges[edgeKey] = std::move(propertyInfo);
+            auto ret = getEdgeWholeValue(spaceId, edgeType, propValues, propNames);
+            if (!ret.ok()) {
+                LOG(ERROR) << ret.status();
+                return ret.status();
+            }
+            spaceDataInfo->edges[edgeKey] = std::move(ret).value();
         }
     }
     return Status::OK();
 }
 
-StatusOr<std::vector<DataSet>>
-StorageCache::getProps(const storage::cpp2::GetPropRequest&) {
-    return {};
-#if 0
-    folly::RWSpinLock::ReadHolder holder(lock_);
-    std::vector<storage::cpp2::VertexPropData> data;
-    auto spaceId = req.get_space_id();
-    auto findIt = cache_.find(spaceId);
-    if (findIt == cache_.end()) {
-        return Status::Error("SpaceID `%d' not found", spaceId);
+StatusOr<std::unordered_map<std::string, Value>>
+StorageCache::getTagWholeValue(const GraphSpaceID spaceId,
+                               const TagID tagId,
+                               const std::vector<Value>& props,
+                               const std::vector<std::string> &names) {
+    if (props.size() != names.size()) {
+        return Status::Error("Wrong size between props and names");
     }
-    auto vertices = findIt->second.vertices;
-    auto parts = req.get_parts();
-    std::vector<storage::cpp2::VertexProp> props;
-    if (req.__isset.vertex_props) {
-        props = req.get_vertex_props();
+    auto schema = mgr_->getTagSchema(spaceId, tagId);
+    if (schema == nullptr) {
+        return Status::Error("TagId `%d' not exist", tagId);
     }
+    return getPropertyInfo(schema, props, names);
+}
 
-    for (auto &part : parts) {
-        for (auto &vId : part.second) {
-            auto vFindIt = vertices.find(vId);
-            if (vFindIt == vertices.end()) {
-                return Status::Error("VertexId `%s' not found", vId.c_str());
+StatusOr<std::unordered_map<std::string, Value>>
+StorageCache::getEdgeWholeValue(const GraphSpaceID spaceId,
+                                const EdgeType edgeType,
+                                const std::vector<Value>& props,
+                                const std::vector<std::string> &names) {
+    if (props.size() != names.size()) {
+        return Status::Error("Wrong size between props and names");
+    }
+    auto schema = mgr_->getEdgeSchema(spaceId, std::abs(edgeType));
+    if (schema == nullptr) {
+        return Status::Error("EdgeType `%d' not exist", edgeType);
+    }
+    return getPropertyInfo(schema, props, names);
+}
+
+StatusOr<std::unordered_map<std::string, Value>>
+StorageCache::getPropertyInfo(std::shared_ptr<const meta::NebulaSchemaProvider> schema,
+                              const std::vector<Value>& props,
+                              const std::vector<std::string> &names) {
+    auto number = schema->getNumFields();
+    if (number == 0 && props.size() == 0) {
+        return {};
+    }
+    if (number == 0 && props.size() != 0) {
+        return Status::Error("Wrong value about empty schema");
+    }
+
+    std::bitset<32> indexBitSet;
+    std::unordered_map<std::string, Value> propertyInfo;
+    auto index = 0u;
+    for (auto &item : names) {
+        auto filed = schema->field(item);
+        if (filed != nullptr) {
+            indexBitSet.set(index);
+            if (!filed->nullable() && props[index].isNull()) {
+                return Status::Error("Wrong null type `%s'", item.c_str());
             }
-            storage::cpp2::VertexPropData vertex;
-            vertex.set_id(vId);
-            vertex.set_props(vFindIt.second);
-            vertex.set_names();
+            propertyInfo.emplace(item, props[index]);
+        } else {
+            return Status::Error("Wrong prop name `%s'", item.c_str());
         }
+        index++;
     }
-#endif
+
+    if (schema->getNumFields() != indexBitSet.count()) {
+        for (auto i = 0u; i < schema->getNumFields(); i++) {
+            if (indexBitSet[i] != 1) {
+                auto field = schema->field(i);
+                if (field != nullptr && !field->hasDefault()) {
+                    return Status::Error("Prop name `%s' without default value",
+                                         field->name());
+                }
+                VLOG(1) << "Add default value, filed name: " << field->name();
+                propertyInfo.emplace(field->name(), field->defaultValue());
+            }
+        }
+    }
+
+    return propertyInfo;
 }
 }  // namespace graph
 }  // namespace nebula

src/mock/StorageCache.h

@@ -16,6 +16,22 @@
 namespace nebula {
 namespace graph {
 
+struct EdgeHasher {
+    std::size_t operator()(const storage::cpp2::EdgeKey& k) const {
+        std::size_t hash_val = 0;
+        hash_val ^= ((std::hash<std::string>()(k.get_src())) << 1);
+        hash_val ^= ((std::hash<int32_t>()(k.get_edge_type())) << 1);
+        hash_val ^= ((std::hash<int64_t>()(k.get_ranking())) << 1);
+        hash_val ^= ((std::hash<std::string>()(k.get_dst())) << 1);
+        return hash_val;
+    }
+};
+
+using VerticesInfo = std::unordered_map<VertexID,
+                     std::unordered_map<TagID, std::unordered_map<std::string, Value>>>;
+using EdgesInfo = std::unordered_map<storage::cpp2::EdgeKey,
+                                     std::unordered_map<std::string, Value>, EdgeHasher>;
+
 class StorageCache final {
 public:
     explicit StorageCache(uint16_t metaPort);
@@ -26,47 +42,30 @@ public:
 
     Status addEdges(const storage::cpp2::AddEdgesRequest& req);
 
-    StatusOr<std::vector<DataSet>> getProps(const storage::cpp2::GetPropRequest&);
-
 private:
-    std::string getEdgeKey(VertexID srcId,
-                           EdgeType type,
-                           EdgeRanking rank,
-                           VertexID dstId) {
-        std::string key;
-        key.reserve(srcId.size() + sizeof(EdgeType) + sizeof(EdgeRanking) + dstId.size());
-        key.append(srcId.data(), srcId.size())
-           .append(reinterpret_cast<const char*>(&type), sizeof(EdgeType))
-           .append(reinterpret_cast<const char*>(&rank), sizeof(EdgeRanking))
-           .append(dstId.data(), dstId.size());
-        return key;
-    }
-
-    std::string srcEdgePrefix(VertexID srcId,
-                              EdgeType type) {
-        std::string key;
-        key.reserve(srcId.size() + sizeof(EdgeType));
-        key.append(srcId.data(), srcId.size())
-           .append(reinterpret_cast<const char*>(&type), sizeof(EdgeType));
-        return key;
-    }
-
-    std::vector<std::string> getTagPropNamesFromCache(const GraphSpaceID spaceId,
-                                                      const TagID tagId);
-
-    std::vector<std::string> getEdgePropNamesFromCache(const GraphSpaceID spaceId,
-                                                       const EdgeType edgeType);
+    StatusOr<std::unordered_map<std::string, Value>>
+    getTagWholeValue(const GraphSpaceID spaceId,
+                     const TagID tagId,
+                     const std::vector<Value>& props,
+                     const std::vector<std::string> &names);
+
+    StatusOr<std::unordered_map<std::string, Value>>
+    getEdgeWholeValue(const GraphSpaceID spaceId,
+                      const EdgeType edgeType,
+                      const std::vector<Value>& props,
+                      const std::vector<std::string> &names);
+
+    StatusOr<std::unordered_map<std::string, Value>>
+    getPropertyInfo(std::shared_ptr<const meta::NebulaSchemaProvider> schema,
+                    const std::vector<Value>& props,
+                    const std::vector<std::string> &names);
 
 private:
-    struct PropertyInfo {
-        std::vector<std::string>                   propNames;
-        std::vector<Value>                         propValues;
-        std::unordered_map<std::string, int32_t>   propIndexes;
-    };
-
     struct SpaceDataInfo {
-        std::unordered_map<VertexID, std::unordered_map<TagID, PropertyInfo>>        vertices;
-        std::unordered_map<std::string, PropertyInfo>                                edges;
+        SpaceDataInfo() = default;
+        ~SpaceDataInfo() = default;
+        VerticesInfo     vertices;
+        EdgesInfo        edges;
     };
 
     std::unordered_map<GraphSpaceID, SpaceDataInfo>   cache_;

src/mock/test/SchemaTest.cpp

@@ -112,12 +112,19 @@ TEST_F(SchemaTest, TestTag) {
         ASSERT_TRUE(resp.__isset.data);
         ASSERT_EQ(expect, *resp.get_data());
     }
+    {
+        cpp2::ExecutionResponse resp;
+        std::string query = "CREATE SPACE A; USE A; "
+                            "CREATE TAG tag1(name STRING, age INT8, grade FIXED_STRING(10));";
+        auto code = client_->execute(query, resp);
+        ASSERT_EQ(cpp2::ErrorCode::SUCCEEDED, code);
+    }
 }
 
 TEST_F(SchemaTest, TestEdge) {
     {
         cpp2::ExecutionResponse resp;
-        std::string query = "CREATE EDGE schoolmate(start int, end int);";
+        std::string query = "USE space_for_default; CREATE EDGE schoolmate(start int, end int);";
         auto code = client_->execute(query, resp);
         ASSERT_EQ(cpp2::ErrorCode::SUCCEEDED, code);
     }
@@ -140,13 +147,26 @@ TEST_F(SchemaTest, TestEdge) {
         expect.rows = std::move(rows);
         ASSERT_EQ(expect, *resp.get_data());
     }
+    {
+        cpp2::ExecutionResponse resp;
+        std::string query = "CREATE SPACE B; USE B; "
+                            "CREATE EDGE edge1(name STRING, age INT8, grade FIXED_STRING(10));";
+        auto code = client_->execute(query, resp);
+        ASSERT_EQ(cpp2::ErrorCode::SUCCEEDED, code);
+    }
 }
 
 TEST_F(SchemaTest, TestInsert) {
     sleep(FLAGS_heartbeat_interval_secs + 1);
     {
         cpp2::ExecutionResponse resp;
-        std::string query = "INSERT VERTEX student(name, age, grade) "
+        std::string query = "USE space_for_default;";
+        auto code = client_->execute(query, resp);
+        ASSERT_EQ(cpp2::ErrorCode::SUCCEEDED, code);
+    }
+    {
+        cpp2::ExecutionResponse resp;
+        std::string query = "USE space_for_default; INSERT VERTEX student(name, age, grade) "
                             "VALUES \"Tom\":(\"Tom\", 18, \"three\");";
         auto code = client_->execute(query, resp);
         ASSERT_EQ(cpp2::ErrorCode::SUCCEEDED, code);

src/mock/test/TestMock.cpp

@@ -167,7 +167,7 @@ TEST_F(MockServerTest, TestMeta) {
     sleep(FLAGS_heartbeat_interval_secs + 1);
 }
 
-TEST_F(MockServerTest, TestStorage) {
+TEST_F(MockServerTest, DISABLED_TestStorage) {
     auto storageClient = gEnv->getStorageClient();
     GraphSpaceID space = 1;
     std::vector<Value> props;
@@ -192,3 +192,4 @@ TEST_F(MockServerTest, TestStorage) {
 
 }   // namespace graph
 }   // namespace nebula
+

src/planner/Admin.h

@@ -58,7 +58,7 @@ private:
                 PlanNode* input,
                 meta::SpaceDesc props,
                 bool ifNotExists)
-    : SingleInputNode(plan, Kind::kCreateSpace, input) {
+        : SingleInputNode(plan, Kind::kCreateSpace, input) {
         props_ = std::move(props);
         ifNotExists_ = ifNotExists;
     }
@@ -96,7 +96,7 @@ private:
     DescSpace(ExecutionPlan* plan,
               PlanNode* input,
               std::string spaceName)
-    : SingleInputNode(plan, Kind::kDescSpace, input) {
+        : SingleInputNode(plan, Kind::kDescSpace, input) {
         spaceName_ = std::move(spaceName);
     }
 

src/planner/Maintain.h

@@ -7,38 +7,22 @@
 #ifndef PLANNER_MAINTAIN_H_
 #define PLANNER_MAINTAIN_H_
 
-#include "Query.h"
+#include "planner/Query.h"
 #include "common/interface/gen-cpp2/meta_types.h"
 #include "common/clients/meta/MetaClient.h"
 
 namespace nebula {
 namespace graph {
-// TODO: All DDLs, DMLs and DQLs could be used in a single query
 // which would make them in a single and big execution plan
-class SchemaNode : public SingleInputNode {
-public:
-    GraphSpaceID space() const {
-        return space_;
-    }
-
-protected:
-    SchemaNode(ExecutionPlan* plan, Kind kind, PlanNode* input, const GraphSpaceID space)
-        : SingleInputNode(plan, kind, input), space_(space) {}
-
-protected:
-    GraphSpaceID        space_;
-};
-
-class CreateSchemaNode : public SchemaNode {
+class CreateSchemaNode : public SingleInputNode {
 protected:
     CreateSchemaNode(ExecutionPlan* plan,
-                     Kind kind,
                      PlanNode* input,
-                     GraphSpaceID space,
+                     Kind kind,
                      std::string name,
                      meta::cpp2::Schema schema,
                      bool ifNotExists)
-        : SchemaNode(plan, kind, input, space)
+        : SingleInputNode(plan, kind, input)
         , name_(std::move(name))
         , schema_(std::move(schema))
         , ifNotExists_(ifNotExists) {}
@@ -66,13 +50,11 @@ class CreateTag final : public CreateSchemaNode {
 public:
     static CreateTag* make(ExecutionPlan* plan,
                            PlanNode* input,
-                           GraphSpaceID space,
                            std::string tagName,
                            meta::cpp2::Schema schema,
                            bool ifNotExists) {
     return new CreateTag(plan,
                          input,
-                         space,
                          std::move(tagName),
                          std::move(schema),
                          ifNotExists);
@@ -85,31 +67,27 @@ public:
 private:
     CreateTag(ExecutionPlan* plan,
               PlanNode* input,
-              GraphSpaceID space,
               std::string tagName,
               meta::cpp2::Schema schema,
               bool ifNotExists)
-    : CreateSchemaNode(plan,
-                       Kind::kCreateTag,
-                       input,
-                       space,
-                       std::move(tagName),
-                       std::move(schema),
-                       ifNotExists) {
-    }
+        : CreateSchemaNode(plan,
+                           input,
+                           Kind::kCreateTag,
+                           std::move(tagName),
+                           std::move(schema),
+                           ifNotExists) {
+        }
 };
 
 class CreateEdge final : public CreateSchemaNode {
 public:
     static CreateEdge* make(ExecutionPlan* plan,
                             PlanNode* input,
-                            GraphSpaceID space,
                             std::string edgeName,
                             meta::cpp2::Schema schema,
                             bool ifNotExists) {
     return new CreateEdge(plan,
                           input,
-                          space,
                           std::move(edgeName),
                           std::move(schema),
                           ifNotExists);
@@ -122,28 +100,26 @@ public:
 private:
     CreateEdge(ExecutionPlan* plan,
                PlanNode* input,
-               GraphSpaceID space,
                std::string edgeName,
                meta::cpp2::Schema schema,
                bool ifNotExists)
         : CreateSchemaNode(plan,
-                           Kind::kCreateEdge,
                            input,
-                           space,
+                           Kind::kCreateEdge,
                            std::move(edgeName),
                            std::move(schema),
                            ifNotExists) {
         }
 };
 
-class AlterTag final : public PlanNode {
+class AlterTag final : public SingleInputNode {
 public:
     std::string explain() const override {
         return "AlterTag";
     }
 };
 
-class AlterEdge final : public PlanNode {
+class AlterEdge final : public SingleInputNode {
 public:
     std::string explain() const override {
         return "AlterEdge";
@@ -153,25 +129,19 @@ public:
 class DescSchema : public SingleInputNode {
 protected:
     DescSchema(ExecutionPlan* plan,
-               Kind kind,
                PlanNode* input,
-               GraphSpaceID space,
+               Kind kind,
                std::string name)
         : SingleInputNode(plan, kind, input)
-        , space_(space)
-        , name_(std::move(name)) {}
+        , name_(std::move(name)) {
+    }
 
 public:
     const std::string& getName() const {
         return name_;
     }
 
-    GraphSpaceID getSpaceId() const {
-        return space_;
-    }
-
 protected:
-    GraphSpaceID           space_;
     std::string            name_;
 };
 
@@ -179,9 +149,8 @@ class DescTag final : public DescSchema {
 public:
     static DescTag* make(ExecutionPlan* plan,
                          PlanNode* input,
-                         GraphSpaceID space,
                          std::string tagName) {
-        return new DescTag(plan, input, space, std::move(tagName));
+        return new DescTag(plan, input, std::move(tagName));
     }
 
     std::string explain() const override {
@@ -191,9 +160,8 @@ public:
 private:
     DescTag(ExecutionPlan* plan,
             PlanNode* input,
-            GraphSpaceID space,
             std::string tagName)
-    : DescSchema(plan, Kind::kDescTag, input, space, std::move(tagName)) {
+        : DescSchema(plan, input, Kind::kDescTag, std::move(tagName)) {
     }
 };
 
@@ -201,9 +169,8 @@ class DescEdge final : public DescSchema {
 public:
     static DescEdge* make(ExecutionPlan* plan,
                           PlanNode* input,
-                          GraphSpaceID space,
                           std::string edgeName) {
-        return new DescEdge(plan, input, space, std::move(edgeName));
+        return new DescEdge(plan, input, std::move(edgeName));
     }
 
     std::string explain() const override {
@@ -213,76 +180,75 @@ public:
 private:
     DescEdge(ExecutionPlan* plan,
              PlanNode* input,
-             GraphSpaceID space,
              std::string edgeName)
-    : DescSchema(plan, Kind::kDescEdge, input, space, std::move(edgeName)) {
+        : DescSchema(plan, input, Kind::kDescEdge, std::move(edgeName)) {
     }
 };
 
-class DropTag final : public PlanNode {
+class DropTag final : public SingleInputNode {
 public:
     std::string explain() const override {
         return "DropTag";
     }
 };
 
-class DropEdge final : public PlanNode {
+class DropEdge final : public SingleInputNode {
 public:
     std::string explain() const override {
         return "DropEdge";
     }
 };
 
-class CreateTagIndex final : public PlanNode {
+class CreateTagIndex final : public SingleInputNode {
 public:
     std::string explain() const override {
         return "CreateTagIndex";
     }
 };
 
-class CreateEdgeIndex final : public PlanNode {
+class CreateEdgeIndex final : public SingleInputNode {
 public:
     std::string explain() const override {
         return "CreateEdgeIndex";
     }
 };
 
-class DescribeTagIndex final : public PlanNode {
+class DescribeTagIndex final : public SingleInputNode {
 public:
     std::string explain() const override {
         return "DescribeTagIndex";
     }
 };
 
-class DescribeEdgeIndex final : public PlanNode {
+class DescribeEdgeIndex final : public SingleInputNode {
 public:
     std::string explain() const override {
         return "DescribeEdgeIndex";
     }
 };
 
-class DropTagIndex final : public PlanNode {
+class DropTagIndex final : public SingleInputNode {
 public:
     std::string explain() const override {
         return "DropTagIndex";
     }
 };
 
-class DropEdgeIndex final : public PlanNode {
+class DropEdgeIndex final : public SingleInputNode {
 public:
     std::string explain() const override {
         return "DropEdgeIndex";
     }
 };
 
-class BuildTagIndex final : public PlanNode {
+class BuildTagIndex final : public SingleInputNode {
 public:
     std::string explain() const override {
         return "BuildTagIndex";
     }
 };
 
-class BuildEdgeIndex final : public PlanNode {
+class BuildEdgeIndex final : public SingleInputNode {
 public:
     std::string explain() const override {
         return "BuildEdgeIndex";

src/planner/Mutate.h

@@ -8,23 +8,22 @@
 #define PLANNER_MUTATE_H_
 
 #include "common/interface/gen-cpp2/storage_types.h"
-#include "Query.h"
+#include "planner/Query.h"
 
 /**
  * All mutate-related nodes would put in this file.
  */
 namespace nebula {
 namespace graph {
-// TODO: All DDLs, DMLs and DQLs could be used in a single query
-// which would make them in a single and big execution plan
 class InsertVertices final : public SingleInputNode {
 public:
-    static InsertVertices* make(ExecutionPlan* plan,
-                                PlanNode* input,
-                                GraphSpaceID spaceId,
-                                std::vector<storage::cpp2::NewVertex> vertices,
-                                std::unordered_map<TagID, std::vector<std::string>> tagPropNames,
-                                bool overwritable) {
+    static InsertVertices* make(
+            ExecutionPlan* plan,
+            PlanNode* input,
+            GraphSpaceID spaceId,
+            std::vector<storage::cpp2::NewVertex> vertices,
+            std::unordered_map<TagID, std::vector<std::string>> tagPropNames,
+            bool overwritable) {
         return new InsertVertices(plan,
                                   input,
                                   spaceId,
@@ -37,10 +36,6 @@ public:
         return "InsertVertices";
     }
 
-    GraphSpaceID space() const {
-        return space_;
-    }
-
     const std::vector<storage::cpp2::NewVertex>& getVertices() const {
         return vertices_;
     }
@@ -53,6 +48,10 @@ public:
         return overwritable_;
     }
 
+    GraphSpaceID getSpace() const {
+        return spaceId_;
+    }
+
 private:
     InsertVertices(ExecutionPlan* plan,
                    PlanNode* input,
@@ -60,14 +59,15 @@ private:
                    std::vector<storage::cpp2::NewVertex> vertices,
                    std::unordered_map<TagID, std::vector<std::string>> tagPropNames,
                    bool overwritable)
-    : SingleInputNode(plan, Kind::kInsertVertices, input)
-    , space_(spaceId)
-    , vertices_(std::move(vertices))
-    , tagPropNames_(std::move(tagPropNames))
-    , overwritable_(overwritable) {}
+        : SingleInputNode(plan, Kind::kInsertVertices, input)
+        , spaceId_(spaceId)
+        , vertices_(std::move(vertices))
+        , tagPropNames_(std::move(tagPropNames))
+        , overwritable_(overwritable) {
+    }
 
 private:
-    GraphSpaceID                                               space_;
+    GraphSpaceID                                               spaceId_{-1};
     std::vector<storage::cpp2::NewVertex>                      vertices_;
     std::unordered_map<TagID, std::vector<std::string>>        tagPropNames_;
     bool                                                       overwritable_;
@@ -93,10 +93,6 @@ public:
         return "InsertEdges";
     }
 
-    GraphSpaceID space() const {
-        return space_;
-    }
-
     const std::vector<std::string>& getPropNames() const {
         return propNames_;
     }
@@ -109,6 +105,10 @@ public:
         return overwritable_;
     }
 
+    GraphSpaceID getSpace() const {
+        return spaceId_;
+    }
+
 private:
     InsertEdges(ExecutionPlan* plan,
                 PlanNode* input,
@@ -116,41 +116,42 @@ private:
                 std::vector<storage::cpp2::NewEdge> edges,
                 std::vector<std::string> propNames,
                 bool overwritable)
-    : SingleInputNode(plan, Kind::kInsertEdges, input)
-    , space_(spaceId)
-    , edges_(std::move(edges))
-    , propNames_(std::move(propNames))
-    , overwritable_(overwritable) {}
+        : SingleInputNode(plan, Kind::kInsertEdges, input)
+        , spaceId_(spaceId)
+        , edges_(std::move(edges))
+        , propNames_(std::move(propNames))
+        , overwritable_(overwritable) {
+    }
 
 private:
-    GraphSpaceID                               space_;
+    GraphSpaceID                               spaceId_{-1};
     std::vector<storage::cpp2::NewEdge>        edges_;
     std::vector<std::string>                   propNames_;
     bool                                       overwritable_;
 };
 
-class UpdateVertex final : public PlanNode {
+class UpdateVertex final : public SingleInputNode {
 public:
     std::string explain() const override {
         return "UpdateVertex";
     }
 };
 
-class UpdateEdge final : public PlanNode {
+class UpdateEdge final : public SingleInputNode {
 public:
     std::string explain() const override {
         return "UpdateEdge";
     }
 };
 
-class DeleteVertex final : public PlanNode {
+class DeleteVertex final : public SingleInputNode {
 public:
     std::string explain() const override {
         return "DeleteVertex";
     }
 };
 
-class DeleteEdge final : public PlanNode {
+class DeleteEdge final : public SingleInputNode {
 public:
     std::string explain() const override {
         return "DeleteEdge";

src/planner/Query.cpp

@@ -4,7 +4,7 @@
  * attached with Common Clause Condition 1.0, found in the LICENSES directory.
  */
 
-#include "Query.h"
+#include "planner/Query.h"
 
 #include <folly/String.h>
 

src/planner/Query.h

@@ -797,35 +797,27 @@ private:
 class SwitchSpace final : public SingleInputNode {
 public:
     static SwitchSpace* make(ExecutionPlan* plan,
-                                        PlanNode* input,
-                                        std::string spaceName,
-                                        GraphSpaceID spaceId) {
-        return new SwitchSpace(plan, input, spaceName, spaceId);
+                             PlanNode* input,
+                             std::string spaceName) {
+        return new SwitchSpace(plan, input, spaceName);
     }
 
     const std::string& getSpaceName() const {
         return spaceName_;
     }
 
-    GraphSpaceID getSpaceId() const {
-        return spaceId_;
-    }
-
     std::string explain() const override;
 
 private:
     SwitchSpace(ExecutionPlan* plan,
-                           PlanNode* input,
-                           std::string spaceName,
-                           GraphSpaceID spaceId)
+                PlanNode* input,
+                std::string spaceName)
         : SingleInputNode(plan, Kind::kSwitchSpace, input) {
         spaceName_ = std::move(spaceName);
-        spaceId_ = spaceId;
     }
 
 private:
     std::string     spaceName_;
-    GraphSpaceID    spaceId_{-1};
 };
 
 class Dedup final : public SingleInputNode {

src/service/ClientSession.h

@@ -56,7 +56,7 @@ public:
 private:
     // ClientSession could only be created via SessionManager
     friend class SessionManager;
-    friend class ValidatorTest;
+    friend class ValidatorTestBase;
     ClientSession() = default;
     explicit ClientSession(int64_t id);
 

src/util/ObjectPool.h

@@ -39,6 +39,10 @@ public:
         return obj;
     }
 
+    bool empty() const {
+        return objects_.empty();
+    }
+
 private:
     struct Element {
         void *obj;

src/util/SchemaUtil.cpp

@@ -79,6 +79,21 @@ Status SchemaUtil::validateProps(const std::vector<SchemaPropItem*> &schemaProps
     return Status::OK();
 }
 
+// static
+std::shared_ptr<const meta::NebulaSchemaProvider>
+SchemaUtil::generateSchemaProvider(const SchemaVer ver, const meta::cpp2::Schema &schema) {
+    auto schemaPtr = std::make_shared<meta::NebulaSchemaProvider>(ver);
+    for (auto col : schema.get_columns()) {
+        bool hasDef = col.__isset.default_value;
+        schemaPtr->addField(col.get_name(),
+                            col.get_type(),
+                            col.__isset.type_length ? *col.get_type_length() : 0,
+                            col.__isset.nullable ? *col.get_nullable() : false,
+                            hasDef ? *col.get_default_value() : Value());
+    }
+    return schemaPtr;
+}
+
 // static
 StatusOr<nebula::Value> SchemaUtil::toSchemaValue(const meta::cpp2::PropertyType type,
                                                   const Value &v) {