Compare commits
1 Commits
rewrite-se
...
separate_r
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
0efe68fdca |
@@ -2517,18 +2517,16 @@ class SetProperty : public memgraph::query::Clause {
|
||||
|
||||
bool Accept(HierarchicalTreeVisitor &visitor) override {
|
||||
if (visitor.PreVisit(*this)) {
|
||||
identifier_->Accept(visitor) && property_lookup_->Accept(visitor) && expression_->Accept(visitor);
|
||||
property_lookup_->Accept(visitor) && expression_->Accept(visitor);
|
||||
}
|
||||
return visitor.PostVisit(*this);
|
||||
}
|
||||
|
||||
memgraph::query::Identifier *identifier_{nullptr};
|
||||
memgraph::query::PropertyLookup *property_lookup_{nullptr};
|
||||
memgraph::query::Expression *expression_{nullptr};
|
||||
|
||||
SetProperty *Clone(AstStorage *storage) const override {
|
||||
SetProperty *object = storage->Create<SetProperty>();
|
||||
object->identifier_ = identifier_ ? identifier_->Clone(storage) : nullptr;
|
||||
object->property_lookup_ = property_lookup_ ? property_lookup_->Clone(storage) : nullptr;
|
||||
object->expression_ = expression_ ? expression_->Clone(storage) : nullptr;
|
||||
return object;
|
||||
|
||||
@@ -2550,10 +2550,6 @@ antlrcpp::Any CypherMainVisitor::visitSetItem(MemgraphCypher::SetItemContext *ct
|
||||
auto *set_property = storage_->Create<SetProperty>();
|
||||
set_property->property_lookup_ = std::any_cast<PropertyLookup *>(ctx->propertyExpression()->accept(this));
|
||||
set_property->expression_ = std::any_cast<Expression *>(ctx->expression()->accept(this));
|
||||
if (ctx->propertyExpression()->atom()->variable()) {
|
||||
set_property->identifier_ = storage_->Create<Identifier>(
|
||||
std::any_cast<std::string>(ctx->propertyExpression()->atom()->variable()->accept(this)));
|
||||
}
|
||||
return static_cast<Clause *>(set_property);
|
||||
}
|
||||
|
||||
|
||||
@@ -2605,9 +2605,9 @@ void Delete::DeleteCursor::Shutdown() { input_cursor_->Shutdown(); }
|
||||
|
||||
void Delete::DeleteCursor::Reset() { input_cursor_->Reset(); }
|
||||
|
||||
SetProperty::SetProperty(const std::shared_ptr<LogicalOperator> &input, Symbol input_symbol,
|
||||
storage::PropertyId property, PropertyLookup *lhs, Expression *rhs)
|
||||
: input_(input), input_symbol_(input_symbol), property_(property), lhs_(lhs), rhs_(rhs) {}
|
||||
SetProperty::SetProperty(const std::shared_ptr<LogicalOperator> &input, storage::PropertyId property,
|
||||
PropertyLookup *lhs, Expression *rhs)
|
||||
: input_(input), property_(property), lhs_(lhs), rhs_(rhs) {}
|
||||
|
||||
ACCEPT_WITH_INPUT(SetProperty)
|
||||
|
||||
|
||||
@@ -1170,8 +1170,8 @@ class SetProperty : public memgraph::query::plan::LogicalOperator {
|
||||
|
||||
SetProperty() {}
|
||||
|
||||
SetProperty(const std::shared_ptr<LogicalOperator> &input, Symbol input_symbol, storage::PropertyId property,
|
||||
PropertyLookup *lhs, Expression *rhs);
|
||||
SetProperty(const std::shared_ptr<LogicalOperator> &input, storage::PropertyId property, PropertyLookup *lhs,
|
||||
Expression *rhs);
|
||||
bool Accept(HierarchicalLogicalOperatorVisitor &visitor) override;
|
||||
UniqueCursorPtr MakeCursor(utils::MemoryResource *) const override;
|
||||
std::vector<Symbol> ModifiedSymbols(const SymbolTable &) const override;
|
||||
@@ -1181,7 +1181,6 @@ class SetProperty : public memgraph::query::plan::LogicalOperator {
|
||||
void set_input(std::shared_ptr<LogicalOperator> input) override { input_ = input; }
|
||||
|
||||
std::shared_ptr<memgraph::query::plan::LogicalOperator> input_;
|
||||
Symbol input_symbol_;
|
||||
storage::PropertyId property_;
|
||||
PropertyLookup *lhs_;
|
||||
Expression *rhs_;
|
||||
@@ -1189,7 +1188,6 @@ class SetProperty : public memgraph::query::plan::LogicalOperator {
|
||||
std::unique_ptr<LogicalOperator> Clone(AstStorage *storage) const override {
|
||||
auto object = std::make_unique<SetProperty>();
|
||||
object->input_ = input_ ? input_->Clone(storage) : nullptr;
|
||||
object->input_symbol_ = input_symbol_;
|
||||
object->property_ = property_;
|
||||
object->lhs_ = lhs_ ? lhs_->Clone(storage) : nullptr;
|
||||
object->rhs_ = rhs_ ? rhs_->Clone(storage) : nullptr;
|
||||
|
||||
@@ -22,7 +22,6 @@
|
||||
#include "query/plan/preprocess.hpp"
|
||||
#include "query/plan/pretty_print.hpp"
|
||||
#include "query/plan/rewrite/index_lookup.hpp"
|
||||
#include "query/plan/rewrite/set_property.hpp"
|
||||
#include "query/plan/rule_based_planner.hpp"
|
||||
#include "query/plan/variable_start_planner.hpp"
|
||||
#include "query/plan/vertex_count_cache.hpp"
|
||||
@@ -44,10 +43,7 @@ class PostProcessor final {
|
||||
|
||||
template <class TPlanningContext>
|
||||
std::unique_ptr<LogicalOperator> Rewrite(std::unique_ptr<LogicalOperator> plan, TPlanningContext *context) {
|
||||
auto rewritten_plan =
|
||||
RewriteWithIndexLookup(std::move(plan), context->symbol_table, context->ast_storage, context->db);
|
||||
return RewriteWithSetPropertyToSetProperties(std::move(rewritten_plan), context->symbol_table, context->ast_storage,
|
||||
context->db);
|
||||
return RewriteWithIndexLookup(std::move(plan), context->symbol_table, context->ast_storage, context->db);
|
||||
}
|
||||
|
||||
template <class TVertexCounts>
|
||||
|
||||
@@ -1,485 +0,0 @@
|
||||
// Copyright 2023 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.
|
||||
|
||||
/// @file
|
||||
/// This file provides a plan rewriter which replaces `Filter` and `ScanAll`
|
||||
/// operations with `ScanAllBy<Index>` if possible. The public entrypoint is
|
||||
/// `RewriteWithIndexLookup`.
|
||||
|
||||
#pragma once
|
||||
|
||||
#include <algorithm>
|
||||
#include <any>
|
||||
#include <memory>
|
||||
#include <optional>
|
||||
#include <unordered_map>
|
||||
#include <unordered_set>
|
||||
#include <vector>
|
||||
|
||||
#include "query/plan/operator.hpp"
|
||||
#include "query/plan/preprocess.hpp"
|
||||
|
||||
namespace memgraph::query::plan {
|
||||
|
||||
namespace impl {
|
||||
|
||||
template <class TDbAccessor>
|
||||
class SetPropertyRewritter final : public HierarchicalLogicalOperatorVisitor {
|
||||
public:
|
||||
SetPropertyRewritter(SymbolTable *symbol_table, AstStorage *ast_storage, TDbAccessor *db)
|
||||
: symbol_table_(symbol_table), ast_storage_(ast_storage), db_(db) {}
|
||||
|
||||
using HierarchicalLogicalOperatorVisitor::PostVisit;
|
||||
using HierarchicalLogicalOperatorVisitor::PreVisit;
|
||||
using HierarchicalLogicalOperatorVisitor::Visit;
|
||||
|
||||
bool Visit(Once &) override { return true; }
|
||||
|
||||
bool PreVisit(Filter &op) override {
|
||||
prev_ops_.push_back(&op);
|
||||
return true;
|
||||
}
|
||||
|
||||
bool PostVisit(Filter &op) override {
|
||||
prev_ops_.pop_back();
|
||||
return true;
|
||||
}
|
||||
|
||||
bool PreVisit(ScanAll &op) override {
|
||||
prev_ops_.push_back(&op);
|
||||
return true;
|
||||
}
|
||||
|
||||
bool PostVisit(ScanAll &scan) override {
|
||||
prev_ops_.pop_back();
|
||||
return true;
|
||||
}
|
||||
|
||||
bool PreVisit(Expand &op) override {
|
||||
prev_ops_.push_back(&op);
|
||||
return true;
|
||||
}
|
||||
|
||||
bool PostVisit(Expand &expand) override {
|
||||
prev_ops_.pop_back();
|
||||
return true;
|
||||
}
|
||||
|
||||
bool PreVisit(ExpandVariable &op) override {
|
||||
prev_ops_.push_back(&op);
|
||||
return true;
|
||||
}
|
||||
|
||||
bool PostVisit(ExpandVariable &expand) override {
|
||||
prev_ops_.pop_back();
|
||||
return true;
|
||||
}
|
||||
|
||||
// The following operators may only use index lookup in filters inside of
|
||||
// their own branches. So we handle them all the same.
|
||||
// * Input operator is visited with the current visitor.
|
||||
// * Custom operator branches are visited with a new visitor.
|
||||
|
||||
bool PreVisit(Merge &op) override {
|
||||
prev_ops_.push_back(&op);
|
||||
op.input()->Accept(*this);
|
||||
RewriteBranch(&op.merge_match_);
|
||||
return false;
|
||||
}
|
||||
|
||||
bool PostVisit(Merge &) override {
|
||||
prev_ops_.pop_back();
|
||||
return true;
|
||||
}
|
||||
|
||||
bool PreVisit(Optional &op) override {
|
||||
prev_ops_.push_back(&op);
|
||||
op.input()->Accept(*this);
|
||||
RewriteBranch(&op.optional_);
|
||||
return false;
|
||||
}
|
||||
|
||||
bool PostVisit(Optional &) override {
|
||||
prev_ops_.pop_back();
|
||||
return true;
|
||||
}
|
||||
|
||||
bool PreVisit(Cartesian &op) override {
|
||||
prev_ops_.push_back(&op);
|
||||
RewriteBranch(&op.left_op_);
|
||||
RewriteBranch(&op.right_op_);
|
||||
return false;
|
||||
}
|
||||
|
||||
bool PostVisit(Cartesian &) override {
|
||||
prev_ops_.pop_back();
|
||||
return true;
|
||||
}
|
||||
|
||||
bool PreVisit(Union &op) override {
|
||||
prev_ops_.push_back(&op);
|
||||
RewriteBranch(&op.left_op_);
|
||||
RewriteBranch(&op.right_op_);
|
||||
return false;
|
||||
}
|
||||
|
||||
bool PostVisit(Union &) override {
|
||||
prev_ops_.pop_back();
|
||||
return true;
|
||||
}
|
||||
|
||||
bool PreVisit(CreateNode &op) override {
|
||||
prev_ops_.push_back(&op);
|
||||
return true;
|
||||
}
|
||||
bool PostVisit(CreateNode &) override {
|
||||
prev_ops_.pop_back();
|
||||
return true;
|
||||
}
|
||||
|
||||
bool PreVisit(CreateExpand &op) override {
|
||||
prev_ops_.push_back(&op);
|
||||
return true;
|
||||
}
|
||||
bool PostVisit(CreateExpand &) override {
|
||||
prev_ops_.pop_back();
|
||||
return true;
|
||||
}
|
||||
|
||||
bool PreVisit(ScanAllByLabel &op) override {
|
||||
prev_ops_.push_back(&op);
|
||||
return true;
|
||||
}
|
||||
bool PostVisit(ScanAllByLabel &) override {
|
||||
prev_ops_.pop_back();
|
||||
return true;
|
||||
}
|
||||
|
||||
bool PreVisit(ScanAllByLabelPropertyRange &op) override {
|
||||
prev_ops_.push_back(&op);
|
||||
return true;
|
||||
}
|
||||
bool PostVisit(ScanAllByLabelPropertyRange &) override {
|
||||
prev_ops_.pop_back();
|
||||
return true;
|
||||
}
|
||||
|
||||
bool PreVisit(ScanAllByLabelPropertyValue &op) override {
|
||||
prev_ops_.push_back(&op);
|
||||
return true;
|
||||
}
|
||||
bool PostVisit(ScanAllByLabelPropertyValue &) override {
|
||||
prev_ops_.pop_back();
|
||||
return true;
|
||||
}
|
||||
|
||||
bool PreVisit(ScanAllByLabelProperty &op) override {
|
||||
prev_ops_.push_back(&op);
|
||||
return true;
|
||||
}
|
||||
bool PostVisit(ScanAllByLabelProperty &) override {
|
||||
prev_ops_.pop_back();
|
||||
return true;
|
||||
}
|
||||
|
||||
bool PreVisit(ScanAllById &op) override {
|
||||
prev_ops_.push_back(&op);
|
||||
return true;
|
||||
}
|
||||
bool PostVisit(ScanAllById &) override {
|
||||
prev_ops_.pop_back();
|
||||
return true;
|
||||
}
|
||||
|
||||
bool PreVisit(ConstructNamedPath &op) override {
|
||||
prev_ops_.push_back(&op);
|
||||
return true;
|
||||
}
|
||||
bool PostVisit(ConstructNamedPath &) override {
|
||||
prev_ops_.pop_back();
|
||||
return true;
|
||||
}
|
||||
|
||||
bool PreVisit(Produce &op) override {
|
||||
prev_ops_.push_back(&op);
|
||||
return true;
|
||||
}
|
||||
bool PostVisit(Produce &) override {
|
||||
prev_ops_.pop_back();
|
||||
return true;
|
||||
}
|
||||
|
||||
bool PreVisit(EmptyResult &op) override {
|
||||
prev_ops_.push_back(&op);
|
||||
return true;
|
||||
}
|
||||
bool PostVisit(EmptyResult &) override {
|
||||
prev_ops_.pop_back();
|
||||
return true;
|
||||
}
|
||||
|
||||
bool PreVisit(Delete &op) override {
|
||||
prev_ops_.push_back(&op);
|
||||
return true;
|
||||
}
|
||||
bool PostVisit(Delete &) override {
|
||||
prev_ops_.pop_back();
|
||||
return true;
|
||||
}
|
||||
|
||||
bool PreVisit(SetProperty &op) override {
|
||||
prev_ops_.push_back(&op);
|
||||
return true;
|
||||
}
|
||||
bool PostVisit(SetProperty &op) override {
|
||||
prev_ops_.pop_back();
|
||||
|
||||
if (op.input_->GetTypeInfo() == SetProperty::kType) {
|
||||
auto *set_prop_op = static_cast<SetProperty *>(op.input_.get());
|
||||
MergeSetPropertyWithSetProperty(op, *set_prop_op);
|
||||
} else if (op.input_->GetTypeInfo() == SetProperties::kType) {
|
||||
auto *set_props_op = static_cast<SetProperties *>(op.input_.get());
|
||||
MergeSetPropertyWithSetProperties(op, *set_props_op);
|
||||
}
|
||||
|
||||
return true;
|
||||
}
|
||||
|
||||
bool PreVisit(SetProperties &op) override {
|
||||
prev_ops_.push_back(&op);
|
||||
return true;
|
||||
}
|
||||
bool PostVisit(SetProperties &) override {
|
||||
prev_ops_.pop_back();
|
||||
return true;
|
||||
}
|
||||
|
||||
bool PreVisit(SetLabels &op) override {
|
||||
prev_ops_.push_back(&op);
|
||||
return true;
|
||||
}
|
||||
bool PostVisit(SetLabels &) override {
|
||||
prev_ops_.pop_back();
|
||||
return true;
|
||||
}
|
||||
|
||||
bool PreVisit(RemoveProperty &op) override {
|
||||
prev_ops_.push_back(&op);
|
||||
return true;
|
||||
}
|
||||
bool PostVisit(RemoveProperty &) override {
|
||||
prev_ops_.pop_back();
|
||||
return true;
|
||||
}
|
||||
|
||||
bool PreVisit(RemoveLabels &op) override {
|
||||
prev_ops_.push_back(&op);
|
||||
return true;
|
||||
}
|
||||
bool PostVisit(RemoveLabels &) override {
|
||||
prev_ops_.pop_back();
|
||||
return true;
|
||||
}
|
||||
|
||||
bool PreVisit(EdgeUniquenessFilter &op) override {
|
||||
prev_ops_.push_back(&op);
|
||||
return true;
|
||||
}
|
||||
bool PostVisit(EdgeUniquenessFilter &) override {
|
||||
prev_ops_.pop_back();
|
||||
return true;
|
||||
}
|
||||
|
||||
bool PreVisit(Accumulate &op) override {
|
||||
prev_ops_.push_back(&op);
|
||||
return true;
|
||||
}
|
||||
bool PostVisit(Accumulate &) override {
|
||||
prev_ops_.pop_back();
|
||||
return true;
|
||||
}
|
||||
|
||||
bool PreVisit(Aggregate &op) override {
|
||||
prev_ops_.push_back(&op);
|
||||
return true;
|
||||
}
|
||||
bool PostVisit(Aggregate &) override {
|
||||
prev_ops_.pop_back();
|
||||
return true;
|
||||
}
|
||||
|
||||
bool PreVisit(Skip &op) override {
|
||||
prev_ops_.push_back(&op);
|
||||
return true;
|
||||
}
|
||||
bool PostVisit(Skip &) override {
|
||||
prev_ops_.pop_back();
|
||||
return true;
|
||||
}
|
||||
|
||||
bool PreVisit(Limit &op) override {
|
||||
prev_ops_.push_back(&op);
|
||||
return true;
|
||||
}
|
||||
bool PostVisit(Limit &) override {
|
||||
prev_ops_.pop_back();
|
||||
return true;
|
||||
}
|
||||
|
||||
bool PreVisit(OrderBy &op) override {
|
||||
prev_ops_.push_back(&op);
|
||||
return true;
|
||||
}
|
||||
bool PostVisit(OrderBy &) override {
|
||||
prev_ops_.pop_back();
|
||||
return true;
|
||||
}
|
||||
|
||||
bool PreVisit(Unwind &op) override {
|
||||
prev_ops_.push_back(&op);
|
||||
return true;
|
||||
}
|
||||
bool PostVisit(Unwind &) override {
|
||||
prev_ops_.pop_back();
|
||||
return true;
|
||||
}
|
||||
|
||||
bool PreVisit(Distinct &op) override {
|
||||
prev_ops_.push_back(&op);
|
||||
return true;
|
||||
}
|
||||
bool PostVisit(Distinct &) override {
|
||||
prev_ops_.pop_back();
|
||||
return true;
|
||||
}
|
||||
|
||||
bool PreVisit(CallProcedure &op) override {
|
||||
prev_ops_.push_back(&op);
|
||||
return true;
|
||||
}
|
||||
bool PostVisit(CallProcedure &) override {
|
||||
prev_ops_.pop_back();
|
||||
return true;
|
||||
}
|
||||
|
||||
bool PreVisit(Foreach &op) override {
|
||||
prev_ops_.push_back(&op);
|
||||
op.input()->Accept(*this);
|
||||
RewriteBranch(&op.update_clauses_);
|
||||
return false;
|
||||
}
|
||||
|
||||
bool PostVisit(Foreach &) override {
|
||||
prev_ops_.pop_back();
|
||||
return true;
|
||||
}
|
||||
|
||||
bool PreVisit(EvaluatePatternFilter &op) override {
|
||||
prev_ops_.push_back(&op);
|
||||
return true;
|
||||
}
|
||||
|
||||
bool PostVisit(EvaluatePatternFilter & /*op*/) override {
|
||||
prev_ops_.pop_back();
|
||||
return true;
|
||||
}
|
||||
|
||||
bool PreVisit(Apply &op) override {
|
||||
prev_ops_.push_back(&op);
|
||||
op.input()->Accept(*this);
|
||||
RewriteBranch(&op.subquery_);
|
||||
return false;
|
||||
}
|
||||
|
||||
bool PostVisit(Apply & /*op*/) override {
|
||||
prev_ops_.pop_back();
|
||||
return true;
|
||||
}
|
||||
|
||||
bool PreVisit(LoadCsv &op) override {
|
||||
prev_ops_.push_back(&op);
|
||||
return true;
|
||||
}
|
||||
|
||||
bool PostVisit(LoadCsv & /*op*/) override {
|
||||
prev_ops_.pop_back();
|
||||
return true;
|
||||
}
|
||||
|
||||
std::shared_ptr<LogicalOperator> new_root_;
|
||||
|
||||
private:
|
||||
SymbolTable *symbol_table_;
|
||||
AstStorage *ast_storage_;
|
||||
TDbAccessor *db_;
|
||||
std::vector<LogicalOperator *> prev_ops_;
|
||||
|
||||
bool DefaultPreVisit() override { throw utils::NotYetImplemented("optimizing set property"); }
|
||||
|
||||
void MergeSetPropertyWithSetProperty(SetProperty &op, SetProperty &child) {
|
||||
if (child.input_symbol_ == op.input_symbol_) {
|
||||
std::unordered_map<PropertyIx, Expression *> elements;
|
||||
elements.insert({child.lhs_->property_, child.rhs_});
|
||||
elements.insert({op.lhs_->property_, op.rhs_});
|
||||
|
||||
auto *map_literal = ast_storage_->Create<MapLiteral>(elements);
|
||||
std::unique_ptr<LogicalOperator> set_properties =
|
||||
std::make_unique<SetProperties>(op.input_->input(), op.input_symbol_, map_literal, SetProperties::Op::UPDATE);
|
||||
SetOnParent(std::move(set_properties));
|
||||
}
|
||||
}
|
||||
|
||||
void MergeSetPropertyWithSetProperties(SetProperty &op, SetProperties &child) {
|
||||
if (child.input_symbol_ == op.input_symbol_) {
|
||||
auto *old_map = utils::Downcast<MapLiteral>(child.rhs_);
|
||||
old_map->elements_.insert({op.lhs_->property_, op.rhs_});
|
||||
|
||||
auto *map_literal = ast_storage_->Create<MapLiteral>(old_map->elements_);
|
||||
std::unique_ptr<LogicalOperator> set_properties =
|
||||
std::make_unique<SetProperties>(op.input_->input(), op.input_symbol_, map_literal, SetProperties::Op::UPDATE);
|
||||
SetOnParent(std::move(set_properties));
|
||||
}
|
||||
}
|
||||
|
||||
void SetOnParent(const std::shared_ptr<LogicalOperator> &input) {
|
||||
MG_ASSERT(input);
|
||||
if (prev_ops_.empty()) {
|
||||
MG_ASSERT(!new_root_);
|
||||
new_root_ = input;
|
||||
return;
|
||||
}
|
||||
prev_ops_.back()->set_input(input);
|
||||
}
|
||||
|
||||
void RewriteBranch(std::shared_ptr<LogicalOperator> *branch) {
|
||||
SetPropertyRewritter<TDbAccessor> rewriter(symbol_table_, ast_storage_, db_);
|
||||
(*branch)->Accept(rewriter);
|
||||
if (rewriter.new_root_) {
|
||||
*branch = rewriter.new_root_;
|
||||
}
|
||||
}
|
||||
};
|
||||
|
||||
} // namespace impl
|
||||
|
||||
template <class TDbAccessor>
|
||||
std::unique_ptr<LogicalOperator> RewriteWithSetPropertyToSetProperties(std::unique_ptr<LogicalOperator> root_op,
|
||||
SymbolTable *symbol_table,
|
||||
AstStorage *ast_storage, TDbAccessor *db) {
|
||||
impl::SetPropertyRewritter<TDbAccessor> rewriter(symbol_table, ast_storage, db);
|
||||
root_op->Accept(rewriter);
|
||||
if (rewriter.new_root_) {
|
||||
throw utils::NotYetImplemented("optimizing set property");
|
||||
}
|
||||
return root_op;
|
||||
}
|
||||
|
||||
} // namespace memgraph::query::plan
|
||||
@@ -392,10 +392,8 @@ class RuleBasedPlanner {
|
||||
} else if (auto *del = utils::Downcast<query::Delete>(clause)) {
|
||||
return std::make_unique<plan::Delete>(std::move(input_op), del->expressions_, del->detach_);
|
||||
} else if (auto *set = utils::Downcast<query::SetProperty>(clause)) {
|
||||
const auto &input_symbol = symbol_table.at(*set->identifier_);
|
||||
return std::make_unique<plan::SetProperty>(std::move(input_op), input_symbol,
|
||||
GetProperty(set->property_lookup_->property_), set->property_lookup_,
|
||||
set->expression_);
|
||||
return std::make_unique<plan::SetProperty>(std::move(input_op), GetProperty(set->property_lookup_->property_),
|
||||
set->property_lookup_, set->expression_);
|
||||
} else if (auto *set = utils::Downcast<query::SetProperties>(clause)) {
|
||||
auto op = set->update_ ? plan::SetProperties::Op::UPDATE : plan::SetProperties::Op::REPLACE;
|
||||
const auto &input_symbol = symbol_table.at(*set->identifier_);
|
||||
|
||||
@@ -69,6 +69,7 @@ struct Config {
|
||||
|
||||
struct DiskConfig {
|
||||
std::filesystem::path main_storage_directory{"storage/rocksdb_main_storage"};
|
||||
std::filesystem::path main_edge_directory{"storage/rocksdb_main_edge_storage"};
|
||||
std::filesystem::path label_index_directory{"storage/rocksdb_label_index"};
|
||||
std::filesystem::path label_property_index_directory{"storage/rocksdb_label_property_index"};
|
||||
std::filesystem::path unique_constraints_directory{"storage/rocksdb_unique_constraints"};
|
||||
@@ -76,6 +77,7 @@ struct Config {
|
||||
std::filesystem::path id_name_mapper_directory{"storage/rocksdb_id_name_mapper"};
|
||||
std::filesystem::path durability_directory{"storage/rocksdb_durability"};
|
||||
std::filesystem::path wal_directory{"storage/rocksdb_wal"};
|
||||
std::filesystem::path wal_edge_directory{"storage/rocksdb_wal_edge"};
|
||||
} disk;
|
||||
|
||||
std::string name;
|
||||
|
||||
@@ -45,21 +45,6 @@ struct RocksDBStorage {
|
||||
|
||||
rocksdb::Options options_;
|
||||
rocksdb::TransactionDB *db_;
|
||||
rocksdb::ColumnFamilyHandle *vertex_chandle = nullptr;
|
||||
rocksdb::ColumnFamilyHandle *edge_chandle = nullptr;
|
||||
rocksdb::ColumnFamilyHandle *default_chandle = nullptr;
|
||||
|
||||
uint64_t ApproximateVertexCount() const {
|
||||
uint64_t estimate_num_keys = 0;
|
||||
db_->GetIntProperty(vertex_chandle, "rocksdb.estimate-num-keys", &estimate_num_keys);
|
||||
return estimate_num_keys;
|
||||
}
|
||||
|
||||
uint64_t ApproximateEdgeCount() const {
|
||||
uint64_t estimate_num_keys = 0;
|
||||
db_->GetIntProperty(edge_chandle, "rocksdb.estimate-num-keys", &estimate_num_keys);
|
||||
return estimate_num_keys;
|
||||
}
|
||||
};
|
||||
|
||||
/// RocksDB comparator that compares keys with timestamps.
|
||||
|
||||
@@ -10,6 +10,7 @@
|
||||
// licenses/APL.txt.
|
||||
|
||||
#include <limits>
|
||||
#include <memory>
|
||||
#include <optional>
|
||||
#include <stdexcept>
|
||||
#include <vector>
|
||||
@@ -233,61 +234,52 @@ void DiskStorage::LoadUniqueConstraintInfoIfExists() const {
|
||||
}
|
||||
}
|
||||
|
||||
void DiskStorage::PrepareRocksDBOptions() {
|
||||
vertex_kvstore_->options_.create_if_missing = true;
|
||||
vertex_kvstore_->options_.comparator = new ComparatorWithU64TsImpl();
|
||||
vertex_kvstore_->options_.compression = rocksdb::kNoCompression;
|
||||
vertex_kvstore_->options_.wal_recovery_mode = rocksdb::WALRecoveryMode::kPointInTimeRecovery;
|
||||
vertex_kvstore_->options_.wal_dir = config_.disk.wal_directory;
|
||||
vertex_kvstore_->options_.wal_compression = rocksdb::kNoCompression;
|
||||
edge_kvstore_->options_.create_if_missing = true;
|
||||
edge_kvstore_->options_.comparator = new ComparatorWithU64TsImpl();
|
||||
edge_kvstore_->options_.compression = rocksdb::kNoCompression;
|
||||
edge_kvstore_->options_.wal_recovery_mode = rocksdb::WALRecoveryMode::kPointInTimeRecovery;
|
||||
edge_kvstore_->options_.wal_dir = config_.disk.wal_edge_directory;
|
||||
edge_kvstore_->options_.wal_compression = rocksdb::kNoCompression;
|
||||
}
|
||||
|
||||
DiskStorage::DiskStorage(Config config)
|
||||
: Storage(config, StorageMode::ON_DISK_TRANSACTIONAL),
|
||||
kvstore_(std::make_unique<RocksDBStorage>()),
|
||||
vertex_kvstore_(std::make_unique<RocksDBStorage>()),
|
||||
edge_kvstore_(std::make_unique<RocksDBStorage>()),
|
||||
durability_kvstore_(std::make_unique<kvstore::KVStore>(config.disk.durability_directory)) {
|
||||
LoadTimestampIfExists();
|
||||
LoadIndexInfoIfExists();
|
||||
LoadConstraintsInfoIfExists();
|
||||
kvstore_->options_.create_if_missing = true;
|
||||
kvstore_->options_.comparator = new ComparatorWithU64TsImpl();
|
||||
kvstore_->options_.compression = rocksdb::kNoCompression;
|
||||
kvstore_->options_.wal_recovery_mode = rocksdb::WALRecoveryMode::kPointInTimeRecovery;
|
||||
kvstore_->options_.wal_dir = config_.disk.wal_directory;
|
||||
kvstore_->options_.wal_compression = rocksdb::kNoCompression;
|
||||
std::vector<rocksdb::ColumnFamilyHandle *> column_handles;
|
||||
std::vector<rocksdb::ColumnFamilyDescriptor> column_families;
|
||||
if (utils::DirExists(config.disk.main_storage_directory)) {
|
||||
column_families.emplace_back(vertexHandle, kvstore_->options_);
|
||||
column_families.emplace_back(edgeHandle, kvstore_->options_);
|
||||
column_families.emplace_back(defaultHandle, kvstore_->options_);
|
||||
|
||||
logging::AssertRocksDBStatus(rocksdb::TransactionDB::Open(kvstore_->options_, rocksdb::TransactionDBOptions(),
|
||||
config.disk.main_storage_directory, column_families,
|
||||
&column_handles, &kvstore_->db_));
|
||||
kvstore_->vertex_chandle = column_handles[0];
|
||||
kvstore_->edge_chandle = column_handles[1];
|
||||
kvstore_->default_chandle = column_handles[2];
|
||||
} else {
|
||||
logging::AssertRocksDBStatus(rocksdb::TransactionDB::Open(kvstore_->options_, rocksdb::TransactionDBOptions(),
|
||||
config.disk.main_storage_directory, &kvstore_->db_));
|
||||
logging::AssertRocksDBStatus(
|
||||
kvstore_->db_->CreateColumnFamily(kvstore_->options_, vertexHandle, &kvstore_->vertex_chandle));
|
||||
logging::AssertRocksDBStatus(
|
||||
kvstore_->db_->CreateColumnFamily(kvstore_->options_, edgeHandle, &kvstore_->edge_chandle));
|
||||
}
|
||||
PrepareRocksDBOptions();
|
||||
logging::AssertRocksDBStatus(rocksdb::TransactionDB::Open(vertex_kvstore_->options_, rocksdb::TransactionDBOptions(),
|
||||
config.disk.main_storage_directory, &vertex_kvstore_->db_));
|
||||
logging::AssertRocksDBStatus(rocksdb::TransactionDB::Open(edge_kvstore_->options_, rocksdb::TransactionDBOptions(),
|
||||
config.disk.main_edge_directory, &edge_kvstore_->db_));
|
||||
}
|
||||
|
||||
DiskStorage::~DiskStorage() {
|
||||
durability_kvstore_->Put(lastTransactionStartTimeStamp, std::to_string(timestamp_));
|
||||
logging::AssertRocksDBStatus(kvstore_->db_->DestroyColumnFamilyHandle(kvstore_->vertex_chandle));
|
||||
logging::AssertRocksDBStatus(kvstore_->db_->DestroyColumnFamilyHandle(kvstore_->edge_chandle));
|
||||
if (kvstore_->default_chandle) {
|
||||
// We must destroy default column family handle only if it was read from existing database.
|
||||
// https://github.com/facebook/rocksdb/issues/5006#issuecomment-1003154821
|
||||
logging::AssertRocksDBStatus(kvstore_->db_->DestroyColumnFamilyHandle(kvstore_->default_chandle));
|
||||
}
|
||||
delete kvstore_->options_.comparator;
|
||||
kvstore_->options_.comparator = nullptr;
|
||||
delete vertex_kvstore_->options_.comparator;
|
||||
vertex_kvstore_->options_.comparator = nullptr;
|
||||
delete edge_kvstore_->options_.comparator;
|
||||
edge_kvstore_->options_.comparator = nullptr;
|
||||
}
|
||||
|
||||
DiskStorage::DiskAccessor::DiskAccessor(DiskStorage *storage, IsolationLevel isolation_level, StorageMode storage_mode)
|
||||
: Accessor(storage, isolation_level, storage_mode), config_(storage->config_.items) {
|
||||
rocksdb::WriteOptions write_options;
|
||||
auto txOptions = rocksdb::TransactionOptions{.set_snapshot = true};
|
||||
disk_transaction_ = storage->kvstore_->db_->BeginTransaction(write_options, txOptions);
|
||||
disk_transaction_->SetReadTimestampForValidation(transaction_.start_timestamp);
|
||||
vertex_disk_transaction_ = storage->vertex_kvstore_->db_->BeginTransaction(write_options, txOptions);
|
||||
vertex_disk_transaction_->SetReadTimestampForValidation(transaction_.start_timestamp);
|
||||
edge_disk_transaction_ = storage->edge_kvstore_->db_->BeginTransaction(write_options, txOptions);
|
||||
edge_disk_transaction_->SetReadTimestampForValidation(transaction_.start_timestamp);
|
||||
}
|
||||
|
||||
DiskStorage::DiskAccessor::DiskAccessor(DiskAccessor &&other) noexcept
|
||||
@@ -382,13 +374,11 @@ std::optional<EdgeAccessor> DiskStorage::DiskAccessor::DeserializeEdge(const roc
|
||||
}
|
||||
|
||||
VerticesIterable DiskStorage::DiskAccessor::Vertices(View view) {
|
||||
auto *disk_storage = static_cast<DiskStorage *>(storage_);
|
||||
rocksdb::ReadOptions ro;
|
||||
std::string strTs = utils::StringTimestamp(transaction_.start_timestamp);
|
||||
rocksdb::Slice ts(strTs);
|
||||
ro.timestamp = &ts;
|
||||
auto it =
|
||||
std::unique_ptr<rocksdb::Iterator>(disk_transaction_->GetIterator(ro, disk_storage->kvstore_->vertex_chandle));
|
||||
auto it = std::unique_ptr<rocksdb::Iterator>(vertex_disk_transaction_->GetIterator(ro));
|
||||
for (it->SeekToFirst(); it->Valid(); it->Next()) {
|
||||
LoadVertexToMainMemoryCache(it->key(), it->value());
|
||||
}
|
||||
@@ -596,10 +586,7 @@ VerticesIterable DiskStorage::DiskAccessor::Vertices(LabelId label, PropertyId p
|
||||
&storage_->constraints_, storage_->config_.items));
|
||||
}
|
||||
|
||||
uint64_t DiskStorage::DiskAccessor::ApproximateVertexCount() const {
|
||||
auto *disk_storage = static_cast<DiskStorage *>(storage_);
|
||||
return disk_storage->kvstore_->ApproximateVertexCount();
|
||||
}
|
||||
uint64_t DiskStorage::DiskAccessor::ApproximateVertexCount() const { return 0; }
|
||||
|
||||
bool DiskStorage::PersistLabelIndexCreation(LabelId label) const {
|
||||
if (auto label_index_store = durability_kvstore_->Get(label_index_str); label_index_store.has_value()) {
|
||||
@@ -693,8 +680,8 @@ uint64_t DiskStorage::GetDiskSpaceUsage() const {
|
||||
}
|
||||
|
||||
StorageInfo DiskStorage::GetInfo() const {
|
||||
auto vertex_count = kvstore_->ApproximateVertexCount();
|
||||
auto edge_count = kvstore_->ApproximateEdgeCount();
|
||||
auto vertex_count = 0U;
|
||||
auto edge_count = 0U;
|
||||
double average_degree = 0.0;
|
||||
if (vertex_count) {
|
||||
// NOLINTNEXTLINE(bugprone-narrowing-conversions, cppcoreguidelines-narrowing-conversions)
|
||||
@@ -761,9 +748,7 @@ std::optional<VertexAccessor> DiskStorage::DiskAccessor::FindVertex(storage::Gid
|
||||
auto strTs = utils::StringTimestamp(transaction_.start_timestamp);
|
||||
rocksdb::Slice ts(strTs);
|
||||
read_opts.timestamp = &ts;
|
||||
auto *disk_storage = static_cast<DiskStorage *>(storage_);
|
||||
auto it = std::unique_ptr<rocksdb::Iterator>(
|
||||
disk_transaction_->GetIterator(read_opts, disk_storage->kvstore_->vertex_chandle));
|
||||
auto it = std::unique_ptr<rocksdb::Iterator>(vertex_disk_transaction_->GetIterator(read_opts));
|
||||
for (it->SeekToFirst(); it->Valid(); it->Next()) {
|
||||
const auto &key = it->key();
|
||||
if (Gid::FromUint(std::stoull(utils::ExtractGidFromKey(key.ToString()))) == gid) {
|
||||
@@ -873,9 +858,7 @@ void DiskStorage::DiskAccessor::PrefetchEdges(const auto &prefetch_edge_filter)
|
||||
auto strTs = utils::StringTimestamp(transaction_.start_timestamp);
|
||||
rocksdb::Slice ts(strTs);
|
||||
read_opts.timestamp = &ts;
|
||||
auto *disk_storage = static_cast<DiskStorage *>(storage_);
|
||||
auto it = std::unique_ptr<rocksdb::Iterator>(
|
||||
disk_transaction_->GetIterator(read_opts, disk_storage->kvstore_->edge_chandle));
|
||||
auto it = std::unique_ptr<rocksdb::Iterator>(vertex_disk_transaction_->GetIterator(read_opts));
|
||||
for (it->SeekToFirst(); it->Valid(); it->Next()) {
|
||||
const rocksdb::Slice &key = it->key();
|
||||
const auto edge_parts = utils::Split(key.ToStringView(), "|");
|
||||
@@ -1134,9 +1117,8 @@ Result<std::optional<EdgeAccessor>> DiskStorage::DiskAccessor::DeleteEdge(EdgeAc
|
||||
/// TODO: at which storage naming
|
||||
/// TODO: this method should also delete the old key
|
||||
bool DiskStorage::DiskAccessor::WriteVertexToDisk(const Vertex &vertex) {
|
||||
auto *disk_storage = static_cast<DiskStorage *>(storage_);
|
||||
auto status = disk_transaction_->Put(disk_storage->kvstore_->vertex_chandle, utils::SerializeVertex(vertex),
|
||||
utils::SerializeProperties(vertex.properties));
|
||||
auto status =
|
||||
vertex_disk_transaction_->Put(utils::SerializeVertex(vertex), utils::SerializeProperties(vertex.properties));
|
||||
if (status.ok()) {
|
||||
spdlog::debug("rocksdb: Saved vertex with key {} and ts {}", utils::SerializeVertex(vertex), *commit_timestamp_);
|
||||
} else if (status.IsBusy()) {
|
||||
@@ -1153,13 +1135,11 @@ bool DiskStorage::DiskAccessor::WriteVertexToDisk(const Vertex &vertex) {
|
||||
|
||||
/// TODO: at which storage naming
|
||||
bool DiskStorage::DiskAccessor::WriteEdgeToDisk(const EdgeRef edge, const std::string &serializedEdgeKey) {
|
||||
auto *disk_storage = static_cast<DiskStorage *>(storage_);
|
||||
rocksdb::Status status;
|
||||
if (config_.properties_on_edges) {
|
||||
status = disk_transaction_->Put(disk_storage->kvstore_->edge_chandle, serializedEdgeKey,
|
||||
utils::SerializeProperties(edge.ptr->properties));
|
||||
status = edge_disk_transaction_->Put(serializedEdgeKey, utils::SerializeProperties(edge.ptr->properties));
|
||||
} else {
|
||||
status = disk_transaction_->Put(disk_storage->kvstore_->edge_chandle, serializedEdgeKey, "");
|
||||
status = edge_disk_transaction_->Put(serializedEdgeKey, "");
|
||||
}
|
||||
if (status.ok()) {
|
||||
spdlog::debug("rocksdb: Saved edge with key {} and ts {}", serializedEdgeKey, *commit_timestamp_);
|
||||
@@ -1175,8 +1155,7 @@ bool DiskStorage::DiskAccessor::WriteEdgeToDisk(const EdgeRef edge, const std::s
|
||||
}
|
||||
|
||||
bool DiskStorage::DiskAccessor::DeleteVertexFromDisk(const std::string &vertex) {
|
||||
auto *disk_storage = static_cast<DiskStorage *>(storage_);
|
||||
auto status = disk_transaction_->Delete(disk_storage->kvstore_->vertex_chandle, vertex);
|
||||
auto status = vertex_disk_transaction_->Delete(vertex);
|
||||
if (status.ok()) {
|
||||
spdlog::debug("rocksdb: Deleted vertex with key {}", vertex);
|
||||
} else if (status.IsBusy()) {
|
||||
@@ -1190,8 +1169,7 @@ bool DiskStorage::DiskAccessor::DeleteVertexFromDisk(const std::string &vertex)
|
||||
}
|
||||
|
||||
bool DiskStorage::DiskAccessor::DeleteEdgeFromDisk(const std::string &edge) {
|
||||
auto *disk_storage = static_cast<DiskStorage *>(storage_);
|
||||
auto status = disk_transaction_->Delete(disk_storage->kvstore_->edge_chandle, edge);
|
||||
auto status = edge_disk_transaction_->Delete(edge);
|
||||
if (status.ok()) {
|
||||
spdlog::debug("rocksdb: Deleted edge with key {}", edge);
|
||||
} else if (status.IsBusy()) {
|
||||
@@ -1366,7 +1344,7 @@ DiskStorage::DiskAccessor::CheckVertexConstraintsBeforeCommit(
|
||||
std::string strTs = utils::StringTimestamp(std::numeric_limits<uint64_t>::max());
|
||||
rocksdb::Slice ts(strTs);
|
||||
ro.timestamp = &ts;
|
||||
auto it = std::unique_ptr<rocksdb::Iterator>(kvstore_->db_->NewIterator(ro, kvstore_->vertex_chandle));
|
||||
auto it = std::unique_ptr<rocksdb::Iterator>(vertex_kvstore_->db_->NewIterator(ro));
|
||||
for (it->SeekToFirst(); it->Valid(); it->Next()) {
|
||||
std::vector<LabelId> labels = utils::DeserializeLabelsFromMainDiskStorage(it->key().ToString());
|
||||
PropertyStore properties = utils::DeserializePropertiesFromMainDiskStorage(it->value().ToStringView());
|
||||
@@ -1387,7 +1365,7 @@ DiskStorage::CheckExistingVerticesBeforeCreatingUniqueConstraint(LabelId label,
|
||||
std::string strTs = utils::StringTimestamp(std::numeric_limits<uint64_t>::max());
|
||||
rocksdb::Slice ts(strTs);
|
||||
ro.timestamp = &ts;
|
||||
auto it = std::unique_ptr<rocksdb::Iterator>(kvstore_->db_->NewIterator(ro, kvstore_->vertex_chandle));
|
||||
auto it = std::unique_ptr<rocksdb::Iterator>(vertex_kvstore_->db_->NewIterator(ro));
|
||||
for (it->SeekToFirst(); it->Valid(); it->Next()) {
|
||||
const std::string key_str = it->key().ToString();
|
||||
std::vector<LabelId> labels = utils::DeserializeLabelsFromMainDiskStorage(key_str);
|
||||
@@ -1435,13 +1413,21 @@ utils::BasicResult<StorageDataManipulationError, void> DiskStorage::DiskAccessor
|
||||
|
||||
if (commit_timestamp_) {
|
||||
// commit_timestamp_ is set only if the transaction has writes.
|
||||
logging::AssertRocksDBStatus(disk_transaction_->SetCommitTimestamp(*commit_timestamp_));
|
||||
logging::AssertRocksDBStatus(vertex_disk_transaction_->SetCommitTimestamp(*commit_timestamp_));
|
||||
logging::AssertRocksDBStatus(edge_disk_transaction_->SetCommitTimestamp(*commit_timestamp_));
|
||||
}
|
||||
auto commitStatus = disk_transaction_->Commit();
|
||||
delete disk_transaction_;
|
||||
disk_transaction_ = nullptr;
|
||||
if (!commitStatus.ok()) {
|
||||
spdlog::error("rocksdb: Commit failed with status {}", commitStatus.ToString());
|
||||
auto vertexCommitStatus = vertex_disk_transaction_->Commit();
|
||||
delete vertex_disk_transaction_;
|
||||
vertex_disk_transaction_ = nullptr;
|
||||
if (!vertexCommitStatus.ok()) {
|
||||
spdlog::error("rocksdb: Vertex commit failed with status {}", vertexCommitStatus.ToString());
|
||||
return StorageDataManipulationError{SerializationError{}};
|
||||
}
|
||||
auto edgeCommitStatus = edge_disk_transaction_->Commit();
|
||||
delete edge_disk_transaction_;
|
||||
edge_disk_transaction_ = nullptr;
|
||||
if (!edgeCommitStatus.ok()) {
|
||||
spdlog::error("rocksdb: Commit failed with status {}", edgeCommitStatus.ToString());
|
||||
return StorageDataManipulationError{SerializationError{}};
|
||||
}
|
||||
spdlog::debug("rocksdb: Commit successful");
|
||||
@@ -1458,7 +1444,7 @@ std::vector<std::pair<std::string, std::string>> DiskStorage::SerializeVerticesF
|
||||
auto strTs = utils::StringTimestamp(std::numeric_limits<uint64_t>::max());
|
||||
rocksdb::Slice ts(strTs);
|
||||
ro.timestamp = &ts;
|
||||
auto it = std::unique_ptr<rocksdb::Iterator>(kvstore_->db_->NewIterator(ro, kvstore_->vertex_chandle));
|
||||
auto it = std::unique_ptr<rocksdb::Iterator>(vertex_kvstore_->db_->NewIterator(ro));
|
||||
|
||||
const std::string serialized_label = utils::SerializeIdType(label);
|
||||
for (it->SeekToFirst(); it->Valid(); it->Next()) {
|
||||
@@ -1484,7 +1470,7 @@ std::vector<std::pair<std::string, std::string>> DiskStorage::SerializeVerticesF
|
||||
auto strTs = utils::StringTimestamp(std::numeric_limits<uint64_t>::max());
|
||||
rocksdb::Slice ts(strTs);
|
||||
ro.timestamp = &ts;
|
||||
auto it = std::unique_ptr<rocksdb::Iterator>(kvstore_->db_->NewIterator(ro, kvstore_->vertex_chandle));
|
||||
auto it = std::unique_ptr<rocksdb::Iterator>(vertex_kvstore_->db_->NewIterator(ro));
|
||||
|
||||
const std::string serialized_label = utils::SerializeIdType(label);
|
||||
for (it->SeekToFirst(); it->Valid(); it->Next()) {
|
||||
@@ -1510,10 +1496,15 @@ void DiskStorage::DiskAccessor::Abort() {
|
||||
// disk_transaction correctly in destructor.
|
||||
// This happens in tests when we create and remove storage in one test. For example, in
|
||||
// query_plan_accumulate_aggregate.cpp
|
||||
disk_transaction_->Rollback();
|
||||
disk_transaction_->ClearSnapshot();
|
||||
delete disk_transaction_;
|
||||
disk_transaction_ = nullptr;
|
||||
vertex_disk_transaction_->Rollback();
|
||||
vertex_disk_transaction_->ClearSnapshot();
|
||||
delete vertex_disk_transaction_;
|
||||
vertex_disk_transaction_ = nullptr;
|
||||
|
||||
edge_disk_transaction_->Rollback();
|
||||
edge_disk_transaction_->ClearSnapshot();
|
||||
delete edge_disk_transaction_;
|
||||
edge_disk_transaction_ = nullptr;
|
||||
|
||||
is_transaction_active_ = false;
|
||||
}
|
||||
|
||||
@@ -208,7 +208,8 @@ class DiskStorage final : public Storage {
|
||||
Config::Items config_;
|
||||
std::vector<std::string> edges_to_delete_;
|
||||
std::vector<std::pair<std::string, std::string>> vertices_to_delete_;
|
||||
rocksdb::Transaction *disk_transaction_;
|
||||
rocksdb::Transaction *vertex_disk_transaction_;
|
||||
rocksdb::Transaction *edge_disk_transaction_;
|
||||
};
|
||||
|
||||
std::unique_ptr<Storage::Accessor> Access(std::optional<IsolationLevel> override_isolation_level) override {
|
||||
@@ -219,7 +220,9 @@ class DiskStorage final : public Storage {
|
||||
return std::unique_ptr<DiskAccessor>(new DiskAccessor{this, isolation_level, storage_mode_});
|
||||
}
|
||||
|
||||
RocksDBStorage *GetRocksDBStorage() const { return kvstore_.get(); }
|
||||
void PrepareRocksDBOptions();
|
||||
|
||||
RocksDBStorage *GetRocksDBStorage() const { return vertex_kvstore_.get(); }
|
||||
|
||||
utils::BasicResult<StorageIndexDefinitionError, void> CreateIndex(
|
||||
LabelId label, std::optional<uint64_t> desired_commit_timestamp) override;
|
||||
@@ -296,7 +299,8 @@ class DiskStorage final : public Storage {
|
||||
|
||||
uint64_t CommitTimestamp(std::optional<uint64_t> desired_commit_timestamp = {});
|
||||
|
||||
std::unique_ptr<RocksDBStorage> kvstore_;
|
||||
std::unique_ptr<RocksDBStorage> vertex_kvstore_;
|
||||
std::unique_ptr<RocksDBStorage> edge_kvstore_;
|
||||
std::unique_ptr<kvstore::KVStore> durability_kvstore_;
|
||||
};
|
||||
|
||||
|
||||
@@ -20,17 +20,20 @@ namespace disk_test_utils {
|
||||
|
||||
memgraph::storage::Config GenerateOnDiskConfig(const std::string &testName) {
|
||||
return {.disk = {.main_storage_directory = "rocksdb_" + testName + "_db",
|
||||
.main_edge_directory = "rocksdb_" + testName + "_edge_db",
|
||||
.label_index_directory = "rocksdb_" + testName + "_label_index",
|
||||
.label_property_index_directory = "rocksdb_" + testName + "_label_property_index",
|
||||
.unique_constraints_directory = "rocksdb_" + testName + "_unique_constraints",
|
||||
.name_id_mapper_directory = "rocksdb_" + testName + "_name_id_mapper",
|
||||
.id_name_mapper_directory = "rocksdb_" + testName + "_id_name_mapper",
|
||||
.durability_directory = "rocksdb_" + testName + "_durability",
|
||||
.wal_directory = "rocksdb_" + testName + "_wal"}};
|
||||
.wal_directory = "rocksdb_" + testName + "_wal",
|
||||
.wal_edge_directory = "rocksdb_" + testName + "_edge_wal"}};
|
||||
}
|
||||
|
||||
void RemoveRocksDbDirs(const std::string &testName) {
|
||||
std::filesystem::remove_all("rocksdb_" + testName + "_db");
|
||||
std::filesystem::remove_all("rocksdb_" + testName + "_edge_db");
|
||||
std::filesystem::remove_all("rocksdb_" + testName + "_label_index");
|
||||
std::filesystem::remove_all("rocksdb_" + testName + "_label_property_index");
|
||||
std::filesystem::remove_all("rocksdb_" + testName + "_unique_constraints");
|
||||
@@ -38,6 +41,7 @@ void RemoveRocksDbDirs(const std::string &testName) {
|
||||
std::filesystem::remove_all("rocksdb_" + testName + "_id_name_mapper");
|
||||
std::filesystem::remove_all("rocksdb_" + testName + "_durability");
|
||||
std::filesystem::remove_all("rocksdb_" + testName + "_wal");
|
||||
std::filesystem::remove_all("rocksdb_" + testName + "_edge_wal");
|
||||
}
|
||||
|
||||
uint64_t GetRealNumberOfEntriesInRocksDB(rocksdb::TransactionDB *disk_storage) {
|
||||
|
||||
@@ -471,8 +471,7 @@ TYPED_TEST(PrintToJsonTest, SetProperty) {
|
||||
memgraph::storage::PropertyId prop = this->dba.NameToProperty("prop");
|
||||
|
||||
std::shared_ptr<LogicalOperator> last_op = std::make_shared<ScanAll>(nullptr, this->GetSymbol("node"));
|
||||
last_op = std::make_shared<plan::SetProperty>(last_op, this->GetSymbol("node"), prop,
|
||||
PROPERTY_LOOKUP(this->dba, "node", prop),
|
||||
last_op = std::make_shared<plan::SetProperty>(last_op, prop, PROPERTY_LOOKUP(this->dba, "node", prop),
|
||||
ADD(PROPERTY_LOOKUP(this->dba, "node", prop), LITERAL(1)));
|
||||
|
||||
this->Check(last_op.get(), R"sep(
|
||||
@@ -626,8 +625,7 @@ TYPED_TEST(PrintToJsonTest, Accumulate) {
|
||||
memgraph::storage::PropertyId prop = this->dba.NameToProperty("prop");
|
||||
auto node_sym = this->GetSymbol("node");
|
||||
std::shared_ptr<LogicalOperator> last_op = std::make_shared<ScanAll>(nullptr, node_sym);
|
||||
last_op = std::make_shared<plan::SetProperty>(last_op, this->GetSymbol("node"), prop,
|
||||
PROPERTY_LOOKUP(this->dba, "node", prop),
|
||||
last_op = std::make_shared<plan::SetProperty>(last_op, prop, PROPERTY_LOOKUP(this->dba, "node", prop),
|
||||
ADD(PROPERTY_LOOKUP(this->dba, "node", prop), LITERAL(1)));
|
||||
last_op = std::make_shared<plan::Accumulate>(last_op, std::vector<Symbol>{node_sym}, true);
|
||||
|
||||
|
||||
@@ -114,9 +114,9 @@ TYPED_TEST(QueryPlanTest, Accumulate) {
|
||||
|
||||
auto one = LITERAL(1);
|
||||
auto n_p = PROPERTY_LOOKUP(dba, IDENT("n")->MapTo(n.sym_), prop);
|
||||
auto set_n_p = std::make_shared<plan::SetProperty>(r_m.op_, n.sym_, prop, n_p, ADD(n_p, one));
|
||||
auto set_n_p = std::make_shared<plan::SetProperty>(r_m.op_, prop, n_p, ADD(n_p, one));
|
||||
auto m_p = PROPERTY_LOOKUP(dba, IDENT("m")->MapTo(r_m.node_sym_), prop);
|
||||
auto set_m_p = std::make_shared<plan::SetProperty>(set_n_p, r_m.node_sym_, prop, m_p, ADD(m_p, one));
|
||||
auto set_m_p = std::make_shared<plan::SetProperty>(set_n_p, prop, m_p, ADD(m_p, one));
|
||||
|
||||
std::shared_ptr<LogicalOperator> last_op = set_m_p;
|
||||
if (accumulate) {
|
||||
|
||||
@@ -1116,10 +1116,10 @@ TYPED_TEST(QueryPlanTest, SetProperty) {
|
||||
auto literal = LITERAL(42);
|
||||
|
||||
auto n_p = PROPERTY_LOOKUP(dba, IDENT("n")->MapTo(n.sym_), prop1);
|
||||
auto set_n_p = std::make_shared<plan::SetProperty>(r_m.op_, n.sym_, prop1, n_p, literal);
|
||||
auto set_n_p = std::make_shared<plan::SetProperty>(r_m.op_, prop1, n_p, literal);
|
||||
|
||||
auto r_p = PROPERTY_LOOKUP(dba, IDENT("r")->MapTo(r_m.edge_sym_), prop1);
|
||||
auto set_r_p = std::make_shared<plan::SetProperty>(set_n_p, r_m.edge_sym_, prop1, r_p, literal);
|
||||
auto set_r_p = std::make_shared<plan::SetProperty>(set_n_p, prop1, r_p, literal);
|
||||
auto context = MakeContext(this->storage, symbol_table, &dba);
|
||||
EXPECT_EQ(2, PullAll(*set_r_p, &context));
|
||||
dba.AdvanceCommand();
|
||||
@@ -1518,7 +1518,7 @@ TYPED_TEST(QueryPlanTest, NodeFilterSet) {
|
||||
// SET n.prop = n.prop + 1
|
||||
auto set_prop = PROPERTY_LOOKUP(dba, IDENT("n")->MapTo(scan_all.sym_), prop);
|
||||
auto add = ADD(set_prop, LITERAL(1));
|
||||
auto set = std::make_shared<plan::SetProperty>(node_filter, scan_all.sym_, prop.second, set_prop, add);
|
||||
auto set = std::make_shared<plan::SetProperty>(node_filter, prop.second, set_prop, add);
|
||||
auto context = MakeContext(this->storage, symbol_table, &dba);
|
||||
EXPECT_EQ(2, PullAll(*set, &context));
|
||||
dba.AdvanceCommand();
|
||||
@@ -1608,11 +1608,11 @@ TYPED_TEST(QueryPlanTest, Merge) {
|
||||
auto r_m = MakeExpand(this->storage, symbol_table, std::make_shared<Once>(), n.sym_, "r", EdgeAtom::Direction::BOTH,
|
||||
{}, "m", false, memgraph::storage::View::OLD);
|
||||
auto m_p = PROPERTY_LOOKUP(dba, IDENT("m")->MapTo(r_m.node_sym_), prop);
|
||||
auto m_set = std::make_shared<plan::SetProperty>(r_m.op_, r_m.node_sym_, prop.second, m_p, LITERAL(1));
|
||||
auto m_set = std::make_shared<plan::SetProperty>(r_m.op_, prop.second, m_p, LITERAL(1));
|
||||
|
||||
// merge_create branch
|
||||
auto n_p = PROPERTY_LOOKUP(dba, IDENT("n")->MapTo(n.sym_), prop);
|
||||
auto n_set = std::make_shared<plan::SetProperty>(std::make_shared<Once>(), n.sym_, prop.second, n_p, LITERAL(2));
|
||||
auto n_set = std::make_shared<plan::SetProperty>(std::make_shared<Once>(), prop.second, n_p, LITERAL(2));
|
||||
|
||||
auto merge = std::make_shared<plan::Merge>(n.op_, m_set, n_set);
|
||||
auto context = MakeContext(this->storage, symbol_table, &dba);
|
||||
@@ -1659,8 +1659,7 @@ TYPED_TEST(QueryPlanTest, SetPropertyOnNull) {
|
||||
auto literal = LITERAL(42);
|
||||
auto n_prop = PROPERTY_LOOKUP(dba, null, prop);
|
||||
auto once = std::make_shared<Once>();
|
||||
auto set_op =
|
||||
std::make_shared<plan::SetProperty>(once, symbol_table.CreateAnonymousSymbol(), prop.second, n_prop, literal);
|
||||
auto set_op = std::make_shared<plan::SetProperty>(once, prop.second, n_prop, literal);
|
||||
auto context = MakeContext(this->storage, symbol_table, &dba);
|
||||
EXPECT_EQ(1, PullAll(*set_op, &context));
|
||||
}
|
||||
@@ -1736,7 +1735,7 @@ TYPED_TEST(QueryPlanTest, DeleteSetProperty) {
|
||||
auto delete_op = std::make_shared<plan::Delete>(n.op_, std::vector<Expression *>{n_get}, false);
|
||||
auto prop = PROPERTY_PAIR(dba, "property");
|
||||
auto n_prop = PROPERTY_LOOKUP(dba, IDENT("n")->MapTo(n.sym_), prop);
|
||||
auto set_op = std::make_shared<plan::SetProperty>(delete_op, n.sym_, prop.second, n_prop, LITERAL(42));
|
||||
auto set_op = std::make_shared<plan::SetProperty>(delete_op, prop.second, n_prop, LITERAL(42));
|
||||
auto context = MakeContext(this->storage, symbol_table, &dba);
|
||||
EXPECT_THROW(PullAll(*set_op, &context), QueryRuntimeException);
|
||||
}
|
||||
@@ -1867,7 +1866,7 @@ class UpdatePropertiesWithAuthFixture : public QueryPlanTest<StorageType> {
|
||||
|
||||
auto literal = LITERAL(new_property_value);
|
||||
auto n_p = PROPERTY_LOOKUP(dba, IDENT("n")->MapTo(scan_all.sym_), entity_prop);
|
||||
auto set_property = std::make_shared<plan::SetProperty>(scan_all.op_, scan_all.sym_, entity_prop, n_p, literal);
|
||||
auto set_property = std::make_shared<plan::SetProperty>(scan_all.op_, entity_prop, n_p, literal);
|
||||
|
||||
// produce the node
|
||||
auto output =
|
||||
@@ -1889,7 +1888,7 @@ class UpdatePropertiesWithAuthFixture : public QueryPlanTest<StorageType> {
|
||||
// set property to 2 on n
|
||||
auto literal = LITERAL(new_property_value);
|
||||
auto n_p = PROPERTY_LOOKUP(dba, IDENT("r")->MapTo(expand.edge_sym_), entity_prop);
|
||||
auto set_property = std::make_shared<plan::SetProperty>(expand.op_, scan_all.sym_, entity_prop, n_p, literal);
|
||||
auto set_property = std::make_shared<plan::SetProperty>(expand.op_, entity_prop, n_p, literal);
|
||||
|
||||
memgraph::glue::FineGrainedAuthChecker auth_checker{user, &dba};
|
||||
auto context = MakeContextWithFineGrainedChecker(this->storage, symbol_table, &dba, &auth_checker);
|
||||
|
||||
@@ -171,8 +171,7 @@ TYPED_TEST(ReadWriteTypeCheckTest, SetRemovePropertiesLabels) {
|
||||
memgraph::storage::PropertyId prop = this->dba.NameToProperty("prop");
|
||||
|
||||
std::shared_ptr<LogicalOperator> last_op = std::make_shared<ScanAll>(nullptr, this->GetSymbol("node"));
|
||||
last_op = std::make_shared<plan::SetProperty>(last_op, this->GetSymbol("node"), prop,
|
||||
PROPERTY_LOOKUP(this->dba, "node", prop),
|
||||
last_op = std::make_shared<plan::SetProperty>(last_op, prop, PROPERTY_LOOKUP(this->dba, "node", prop),
|
||||
ADD(PROPERTY_LOOKUP(this->dba, "node", prop), LITERAL(1)));
|
||||
last_op = std::make_shared<plan::RemoveProperty>(
|
||||
last_op, this->dba.NameToProperty("prop"), PROPERTY_LOOKUP(this->dba, "node", this->dba.NameToProperty("prop")));
|
||||
|
||||
Reference in New Issue
Block a user