diff --git a/src/query/plan/operator.cpp b/src/query/plan/operator.cpp index 50058a99c..53a3fbf85 100644 --- a/src/query/plan/operator.cpp +++ b/src/query/plan/operator.cpp @@ -1155,9 +1155,8 @@ void Aggregate::AggregateCursor::EnsureOkForAvgSum( } } -bool Aggregate::AggregateCursor::TypedValueListEqual::operator()( - const std::list &left, - const std::list &right) const { +bool TypedValueListEqual::operator()(const std::list &left, + const std::list &right) const { return std::equal(left.begin(), left.end(), right.begin(), TypedValue::BoolEqual{}); } @@ -1581,4 +1580,35 @@ void Unwind::UnwindCursor::Reset() { input_value_it_ = input_value_.end(); } +Distinct::Distinct(const std::shared_ptr &input, + const std::vector &value_symbols) + : input_(input ? input : std::make_shared()), + value_symbols_(value_symbols) {} + +ACCEPT_WITH_INPUT(Distinct) + +std::unique_ptr Distinct::MakeCursor(GraphDbAccessor &db) { + return std::make_unique(*this, db); +} + +Distinct::DistinctCursor::DistinctCursor(Distinct &self, GraphDbAccessor &db) + : self_(self), input_cursor_(self.input_->MakeCursor(db)) {} + +bool Distinct::DistinctCursor::Pull(Frame &frame, + const SymbolTable &symbol_table) { + while (true) { + if (!input_cursor_->Pull(frame, symbol_table)) return false; + + std::list row; + for (const auto &symbol : self_.value_symbols_) + row.emplace_back(frame[symbol]); + if (seen_rows_.insert(std::move(row)).second) return true; + } +} + +void Distinct::DistinctCursor::Reset() { + input_cursor_->Reset(); + seen_rows_.clear(); +} + } // namespace query::plan diff --git a/src/query/plan/operator.hpp b/src/query/plan/operator.hpp index eb9c1c39a..6e22aa43a 100644 --- a/src/query/plan/operator.hpp +++ b/src/query/plan/operator.hpp @@ -76,6 +76,7 @@ class OrderBy; class Merge; class Optional; class Unwind; +class Distinct; /** @brief Base class for visitors of @c LogicalOperator class hierarchy. */ using LogicalOperatorVisitor = ::utils::Visitor< @@ -83,7 +84,7 @@ using LogicalOperatorVisitor = ::utils::Visitor< Filter, Produce, Delete, SetProperty, SetProperties, SetLabels, RemoveProperty, RemoveLabels, ExpandUniquenessFilter, ExpandUniquenessFilter, Accumulate, AdvanceCommand, Aggregate, - Skip, Limit, OrderBy, Merge, Optional, Unwind>; + Skip, Limit, OrderBy, Merge, Optional, Unwind, Distinct>; /** @brief Base class for logical operators. * @@ -886,6 +887,15 @@ class Accumulate : public LogicalOperator { }; }; +/** + * Custom equality function for a list of typed values. + * Used in unordered_maps in Aggregate and Distinct operators. + */ +struct TypedValueListEqual { + bool operator()(const std::list &left, + const std::list &right) const; +}; + /** @brief Performs an arbitrary number of aggregations of data * from the given input grouped by the given criteria. * @@ -930,12 +940,6 @@ class Aggregate : public LogicalOperator { void Reset() override; private: - // custom equality function for the unordered map - struct TypedValueListEqual { - bool operator()(const std::list &left, - const std::list &right) const; - }; - // Data structure for a single aggregation cache. // does NOT include the group-by values since those // are a key in the aggregation map. @@ -1293,5 +1297,44 @@ class Unwind : public LogicalOperator { }; }; +/** + * Ensures that only distinct rows are yielded. + * This implementation accepts a vector of Symbols + * which define a row. Only those Symbols are valid + * for use in operators following Distinct. + * + * This implementation maintains input ordering. + */ +class Distinct : public LogicalOperator { + public: + Distinct(const std::shared_ptr &input, + const std::vector &value_symbols); + void Accept(LogicalOperatorVisitor &visitor) override; + std::unique_ptr MakeCursor(GraphDbAccessor &db) override; + + private: + const std::shared_ptr input_; + const std::vector value_symbols_; + + class DistinctCursor : public Cursor { + public: + DistinctCursor(Distinct &self, GraphDbAccessor &db); + + bool Pull(Frame &frame, const SymbolTable &symbol_table) override; + void Reset() override; + + private: + const Distinct &self_; + const std::unique_ptr input_cursor_; + // a set of already seen rows + std::unordered_set< + std::list, + // use FNV collection hashing specialized for a list of TypedValues + FnvCollection, TypedValue, TypedValue::Hash>, + TypedValueListEqual> + seen_rows_; + }; +}; + } // namespace plan } // namespace query diff --git a/tests/unit/query_plan_match_filter_return.cpp b/tests/unit/query_plan_match_filter_return.cpp index c0b1eb5ef..9f77c5336 100644 --- a/tests/unit/query_plan_match_filter_return.cpp +++ b/tests/unit/query_plan_match_filter_return.cpp @@ -285,10 +285,10 @@ TEST(QueryPlan, ExpandOptional) { // MATCH (n) OPTIONAL MATCH (n)-[r]->(m) auto n = MakeScanAll(storage, symbol_table, "n"); - auto r_m = MakeExpand(storage, symbol_table, nullptr, n.sym_, - "r", EdgeAtom::Direction::RIGHT, false, "m", false); - auto optional = std::make_shared(n.op_, r_m.op_, - std::vector{r_m.edge_sym_, r_m.node_sym_}); + auto r_m = MakeExpand(storage, symbol_table, nullptr, n.sym_, "r", + EdgeAtom::Direction::RIGHT, false, "m", false); + auto optional = std::make_shared( + n.op_, r_m.op_, std::vector{r_m.edge_sym_, r_m.node_sym_}); // RETURN n, r, m auto n_ne = NEXPR("n", IDENT("n")); @@ -336,7 +336,7 @@ TEST(QueryPlan, OptionalMatchEmptyDB) { symbol_table[*n_ne->expression_] = n.sym_; symbol_table[*n_ne] = symbol_table.CreateSymbol("n"); auto optional = std::make_shared(nullptr, n.op_, - std::vector{n.sym_}); + std::vector{n.sym_}); auto produce = MakeProduce(optional, n_ne); auto results = CollectProduce(produce, symbol_table, *dba).GetResults(); @@ -642,3 +642,49 @@ TEST(QueryPlan, ExpandUniquenessFilter) { EXPECT_EQ(0, check_expand_results(true, false)); EXPECT_EQ(1, check_expand_results(false, true)); } + +TEST(QueryPlan, Distinct) { + // test queries like + // UNWIND [1, 2, 3, 3] AS x RETURN DISTINCT x + + Dbms dbms; + auto dba = dbms.active(); + AstTreeStorage storage; + SymbolTable symbol_table; + + auto check_distinct = [&](const std::vector input, + const std::vector output, + bool assume_int_value) { + + auto input_expr = LITERAL(TypedValue(input)); + + auto x = symbol_table.CreateSymbol("x"); + auto unwind = std::make_shared(nullptr, input_expr, x); + auto x_expr = IDENT("x"); + symbol_table[*x_expr] = x; + + auto distinct = + std::make_shared(unwind, std::vector{x}); + + auto x_ne = NEXPR("x", x_expr); + symbol_table[*x_ne] = symbol_table.CreateSymbol("x_ne"); + auto produce = MakeProduce(distinct, x_ne); + + auto results = CollectProduce(produce, symbol_table, *dba).GetResults(); + ASSERT_EQ(output.size(), results.size()); + auto output_it = output.begin(); + for (const auto &row : results) { + ASSERT_EQ(1, row.size()); + ASSERT_EQ(row[0].type(), output_it->type()); + if (assume_int_value) + EXPECT_EQ(output_it->Value(), row[0].Value()); + output_it++; + } + }; + + check_distinct({1, 1, 2, 3, 3, 3}, {1, 2, 3}, true); + check_distinct({3, 2, 3, 5, 3, 5, 2, 1, 2}, {3, 2, 5, 1}, true); + check_distinct( + {3, "two", TypedValue::Null, 3, true, false, "TWO", TypedValue::Null}, + {3, "two", TypedValue::Null, true, false, "TWO"}, false); +}