diff --git a/src/query/v2/multiframe.cpp b/src/query/v2/multiframe.cpp index 4829addb2..88262e514 100644 --- a/src/query/v2/multiframe.cpp +++ b/src/query/v2/multiframe.cpp @@ -24,7 +24,7 @@ static_assert(std::forward_iterator); static_assert(std::forward_iterator); static_assert(std::forward_iterator); -MultiFrame::MultiFrame(int64_t size_of_frame, size_t number_of_frames, utils::MemoryResource *execution_memory) +MultiFrame::MultiFrame(size_t size_of_frame, size_t number_of_frames, utils::MemoryResource *execution_memory) : frames_(utils::pmr::vector( number_of_frames, FrameWithValidity(size_of_frame, execution_memory), execution_memory)) { MG_ASSERT(number_of_frames > 0); diff --git a/src/query/v2/multiframe.hpp b/src/query/v2/multiframe.hpp index 0b6896422..37e6fea11 100644 --- a/src/query/v2/multiframe.hpp +++ b/src/query/v2/multiframe.hpp @@ -30,7 +30,7 @@ class MultiFrame { friend class ValidFramesReader; friend class InvalidFramesPopulator; - MultiFrame(int64_t size_of_frame, size_t number_of_frames, utils::MemoryResource *execution_memory); + MultiFrame(size_t size_of_frame, size_t number_of_frames, utils::MemoryResource *execution_memory); ~MultiFrame() = default; MultiFrame(const MultiFrame &other); diff --git a/src/query/v2/plan/operator.cpp b/src/query/v2/plan/operator.cpp index eeb5cd6b4..c909cc466 100644 --- a/src/query/v2/plan/operator.cpp +++ b/src/query/v2/plan/operator.cpp @@ -2376,6 +2376,22 @@ class DistributedCreateExpandCursor : public Cursor { return true; } + void PullMultiple(MultiFrame &multi_frame, ExecutionContext &context) override { + SCOPED_PROFILE_OP("CreateExpandMF"); + input_cursor_->PullMultiple(multi_frame, context); + auto request_vertices = ExpandCreationInfoToRequests(multi_frame, context); + { + SCOPED_REQUEST_WAIT_PROFILE; + auto &request_router = context.request_router; + auto results = request_router->CreateExpand(std::move(request_vertices)); + for (const auto &result : results) { + if (result.error) { + throw std::runtime_error("CreateExpand Request failed"); + } + } + } + } + void Shutdown() override { input_cursor_->Shutdown(); } void Reset() override { @@ -2450,6 +2466,64 @@ class DistributedCreateExpandCursor : public Cursor { return edge_requests; } + std::vector ExpandCreationInfoToRequests(MultiFrame &multi_frame, ExecutionContext &context) const { + std::vector edge_requests; + auto reader = multi_frame.GetValidFramesConsumer(); + + for (auto &frame : reader) { + const auto &edge_info = self_.edge_info_; + msgs::NewExpand request{.id = {context.edge_ids_alloc->AllocateId()}}; + ExpressionEvaluator evaluator(&frame, context.symbol_table, context.evaluation_context, nullptr, + storage::v3::View::NEW); + request.type = {edge_info.edge_type}; + if (const auto *edge_info_properties = std::get_if(&edge_info.properties)) { + for (const auto &[property, value_expression] : *edge_info_properties) { + TypedValue val = value_expression->Accept(evaluator); + request.properties.emplace_back(property, storage::v3::TypedValueToValue(val)); + } + } else { + // handle parameter + auto property_map = evaluator.Visit(*std::get(edge_info.properties)).ValueMap(); + for (const auto &[property, value] : property_map) { + const auto property_id = context.request_router->NameToProperty(std::string(property)); + request.properties.emplace_back(property_id, storage::v3::TypedValueToValue(value)); + } + } + // src, dest + TypedValue &v1_value = frame[self_.input_symbol_]; + const auto &v1 = v1_value.ValueVertex(); + const auto &v2 = OtherVertex(frame); + + // Set src and dest vertices + // TODO(jbajic) Currently we are only handling scenario where vertices + // are matched + const auto set_vertex = [](const auto &vertex, auto &vertex_id) { + vertex_id.first = vertex.PrimaryLabel(); + vertex_id.second = vertex.GetVertex().id.second; + }; + + std::invoke([&]() { + switch (edge_info.direction) { + case EdgeAtom::Direction::IN: { + set_vertex(v2, request.src_vertex); + set_vertex(v1, request.dest_vertex); + break; + } + case EdgeAtom::Direction::OUT: { + set_vertex(v1, request.src_vertex); + set_vertex(v2, request.dest_vertex); + break; + } + case EdgeAtom::Direction::BOTH: + LOG_FATAL("Must indicate exact expansion direction here"); + } + }); + + edge_requests.push_back(std::move(request)); + } + return edge_requests; + } + private: void ResetExecutionState() {} diff --git a/tests/unit/CMakeLists.txt b/tests/unit/CMakeLists.txt index 188c6c1b0..b1c5c9c6f 100644 --- a/tests/unit/CMakeLists.txt +++ b/tests/unit/CMakeLists.txt @@ -410,3 +410,7 @@ target_link_libraries(${test_prefix}high_density_shard_create_scan mg-io mg-coor # Tests for awesome_memgraph_functions add_unit_test(query_v2_expression_evaluator.cpp) target_link_libraries(${test_prefix}query_v2_expression_evaluator mg-query-v2) + +# Tests for multiframes +add_unit_test(query_v2_create_expand_multiframe.cpp) +target_link_libraries(${test_prefix}query_v2_create_expand_multiframe mg-query-v2) diff --git a/tests/unit/mock_helpers.hpp b/tests/unit/mock_helpers.hpp new file mode 100644 index 000000000..5201aa210 --- /dev/null +++ b/tests/unit/mock_helpers.hpp @@ -0,0 +1,63 @@ +// Copyright 2022 Memgraph Ltd. +// +// Use of this software is governed by the Business Source License +// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source +// License, and you may not use this file except in compliance with the Business Source License. +// +// As of the Change Date specified in that file, in accordance with +// the Business Source License, use of this software will be governed +// by the Apache License, Version 2.0, included in the file +// licenses/APL.txt. + +#pragma once + +#include +#include +#include "query/v2/plan/operator.hpp" +#include "query/v2/request_router.hpp" + +namespace memgraph { +class MockedRequestRouter : public query::v2::RequestRouterInterface { + public: + MOCK_METHOD1(ScanVertices, std::vector(std::optional label)); + MOCK_METHOD1(CreateVertices, std::vector(std::vector)); + MOCK_METHOD1(ExpandOne, std::vector(msgs::ExpandOneRequest)); + MOCK_METHOD1(CreateExpand, std::vector(std::vector)); + MOCK_METHOD1(GetProperties, std::vector(msgs::GetPropertiesRequest)); + MOCK_METHOD0(StartTransaction, void()); + MOCK_METHOD0(Commit, void()); + + MOCK_CONST_METHOD1(NameToEdgeType, storage::v3::EdgeTypeId(const std::string &)); + MOCK_CONST_METHOD1(NameToProperty, storage::v3::PropertyId(const std::string &)); + MOCK_CONST_METHOD1(NameToLabel, storage::v3::LabelId(const std::string &)); + MOCK_CONST_METHOD1(LabelToName, storage::v3::LabelId(const std::string &)); + MOCK_CONST_METHOD1(PropertyToName, const std::string &(storage::v3::PropertyId)); + MOCK_CONST_METHOD1(LabelToName, const std::string &(storage::v3::LabelId label)); + MOCK_CONST_METHOD1(EdgeTypeToName, const std::string &(storage::v3::EdgeTypeId type)); + MOCK_CONST_METHOD1(MaybeNameToProperty, std::optional(const std::string &)); + MOCK_CONST_METHOD1(MaybeNameToEdgeType, std::optional(const std::string &)); + MOCK_CONST_METHOD1(MaybeNameToLabel, std::optional(const std::string &)); + MOCK_CONST_METHOD1(IsPrimaryLabel, bool(storage::v3::LabelId)); + MOCK_CONST_METHOD2(IsPrimaryKey, bool(storage::v3::LabelId, storage::v3::PropertyId)); +}; + +class MockedLogicalOperator : query::v2::plan::LogicalOperator { + public: + MOCK_CONST_METHOD1(MakeCursor, query::v2::plan::UniqueCursorPtr(utils::MemoryResource *)); + MOCK_CONST_METHOD1(OutputSymbols, std::vector(const expr::SymbolTable &)); + MOCK_CONST_METHOD1(ModifiedSymbols, std::vector(const expr::SymbolTable &)); + MOCK_CONST_METHOD0(HasSingleInput, bool()); + MOCK_CONST_METHOD0(input, std::shared_ptr()); + MOCK_METHOD1(set_input, void(std::shared_ptr)); + MOCK_CONST_METHOD1(Clone, std::unique_ptr(query::v2::AstStorage *storage)); +}; + +class MockedCursor : memgraph::query::v2::plan::Cursor { + public: + MOCK_METHOD2(Pull, bool(query::v2::Frame &, expr::ExecutionContext &)); + MOCK_METHOD2(PullMultiple, void(query::v2::MultiFrame &, expr::ExecutionContext &)); + MOCK_METHOD0(Reset, void()); + MOCK_METHOD0(Shutdown, void()); +}; + +} // namespace memgraph diff --git a/tests/unit/query_v2_create_expand_multiframe.cpp b/tests/unit/query_v2_create_expand_multiframe.cpp new file mode 100644 index 000000000..f6f6567db --- /dev/null +++ b/tests/unit/query_v2_create_expand_multiframe.cpp @@ -0,0 +1,88 @@ +// Copyright 2022 Memgraph Ltd. +// +// Use of this software is governed by the Business Source License +// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source +// License, and you may not use this file except in compliance with the Business Source License. +// +// As of the Change Date specified in that file, in accordance with +// the Business Source License, use of this software will be governed +// by the Apache License, Version 2.0, included in the file +// licenses/APL.txt. + +#include "mock_helpers.hpp" + +#include "query/v2/bindings/frame.hpp" +#include "query/v2/bindings/symbol_table.hpp" +#include "query/v2/common.hpp" +#include "query/v2/context.hpp" +#include "query/v2/plan/operator.hpp" +#include "query/v2/requests.hpp" +#include "storage/v3/property_value.hpp" +#include "storage/v3/shard.hpp" +#include "utils/logging.hpp" +#include "utils/memory.hpp" + +using namespace memgraph::query::v2; +using namespace memgraph::query::v2::plan; +namespace memgraph { +class TestTemplate : public testing::Test { + protected: + void SetUp() override {} +}; + +ExecutionContext MakeContext(const AstStorage &storage, const SymbolTable &symbol_table, RequestRouterInterface *router, + IdAllocator *id_alloc) { + ExecutionContext context; + context.symbol_table = symbol_table; + context.evaluation_context.properties = NamesToProperties(storage.properties_, router); + context.evaluation_context.labels = NamesToLabels(storage.labels_, router); + context.edge_ids_alloc = id_alloc; + context.request_router = router; + return context; +} + +MultiFrame CreateMultiFrame(const size_t max_pos, const Symbol &src, const Symbol &dst, MockedRequestRouter *router) { + static constexpr size_t frame_size = 100; + MultiFrame multi_frame(max_pos, frame_size, utils::NewDeleteResource()); + auto frames_populator = multi_frame.GetInvalidFramesPopulator(); + size_t i = 0; + for (auto &frame : frames_populator) { + frame.MakeValid(); + auto &src_acc = frame.at(src); + auto &dst_acc = frame.at(dst); + auto v1 = msgs::Vertex{.id = {{msgs::LabelId::FromUint(1)}, {msgs::Value(static_cast(i++))}}}; + auto v2 = msgs::Vertex{.id = {{msgs::LabelId::FromUint(1)}, {msgs::Value(static_cast(i++))}}}; + std::map mp; + src_acc = TypedValue(query::v2::accessors::VertexAccessor(v1, mp, router)); + dst_acc = TypedValue(query::v2::accessors::VertexAccessor(v2, mp, router)); + } + + return multi_frame; +} + +TEST_F(TestTemplate, CreateExpand) { + MockedRequestRouter router; + + AstStorage ast; + SymbolTable symbol_table; + + query::v2::plan::NodeCreationInfo node; + query::v2::plan::EdgeCreationInfo edge; + edge.edge_type = msgs::EdgeTypeId::FromUint(1); + edge.direction = EdgeAtom::Direction::IN; + auto id_alloc = IdAllocator(0, 100); + + const auto &src = symbol_table.CreateSymbol("n", true); + node.symbol = symbol_table.CreateSymbol("u", true); + + auto create_expand = query::v2::plan::CreateExpand(node, edge, nullptr, src, true); + auto cursor = create_expand.MakeCursor(utils::NewDeleteResource()); + + EXPECT_CALL(router, CreateExpand(testing::_)) + .Times(1) + .WillOnce(::testing::Return(std::vector{})); + auto context = MakeContext(ast, symbol_table, &router, &id_alloc); + auto multi_frame = CreateMultiFrame(context.symbol_table.max_position(), src, node.symbol, &router); + cursor->PullMultiple(multi_frame, context); +} +} // namespace memgraph