Operator.hpp

Go to the documentation of this file.

#pragma once
 
#include <mutable/mutable-config.hpp>
#include <mutable/catalog/Schema.hpp>
#include <mutable/IR/CNF.hpp>
#include <mutable/IR/QueryGraph.hpp>
#include <mutable/storage/Store.hpp>
#include <mutable/util/enum_ops.hpp>
#include <mutable/util/macro.hpp>
#include <functional>
#include <iostream>
#include <unordered_map>
#include <utility>
#include <variant>
#include <vector>
 
 
namespace m {
 
// forward declarations
struct OperatorVisitor;
struct ConstOperatorVisitor;
struct Tuple;
 
struct M_EXPORT OperatorInformation
{
    QueryGraph::Subproblem subproblem;
 
    double estimated_cardinality;
};
 
struct M_EXPORT OperatorData
{
    virtual ~OperatorData() = 0;
};
 
struct M_EXPORT Operator
{
    friend void swap(Operator &first, Operator &second) {
        using std::swap;
        swap(first.schema_, second.schema_);
        swap(first.info_,   second.info_);
        swap(first.data_,   second.data_);
        swap(first.id_,     second.id_);
    }
 
    private:
    Schema schema_; 
    std::unique_ptr<OperatorInformation> info_; 
    mutable OperatorData *data_ = nullptr; 
    protected:
    mutable std::size_t id_ = -1UL; 
 
    public:
    Operator() = default;
    Operator(Operator &&other) : Operator() { swap(*this, other); }
    virtual ~Operator() { delete data_; }
 
    Schema & schema() { return schema_; }
    const Schema & schema() const { return schema_; }
 
    bool has_info() const { return bool(info_); }
    OperatorInformation & info() { M_insist(bool(info_)); return *info_; }
    const OperatorInformation & info() const { return const_cast<Operator*>(this)->info(); }
    std::unique_ptr<OperatorInformation> info(std::unique_ptr<OperatorInformation> new_info) {
        using std::swap;
        swap(new_info, info_);
        return new_info;
    }
 
    virtual void assign_post_order_ids(std::size_t start_id = 0UL) const { id_ = start_id; }
    virtual void reset_ids() const { id_ = -1UL; }
    std::size_t id() const {
        M_insist(id_ != -1UL, "id must be first set by calling `assign_post_order_ids()`");
        return id_;
    }
    private:
    void id(std::size_t id) const { id_ = id; }
 
    public:
    OperatorData * data(OperatorData *data) const { std::swap(data, data_); return data; }
    OperatorData * data() const { return data_; }
 
    virtual void accept(OperatorVisitor &v) = 0;
    virtual void accept(ConstOperatorVisitor &v) const = 0;
 
    friend M_EXPORT std::ostream & operator<<(std::ostream &out, const Operator &op);
 
    void minimize_schema();
 
    void dot(std::ostream &out) const;
 
    void dump(std::ostream &out) const;
    void dump() const;
};
 
struct Consumer;
 
struct M_EXPORT Producer : virtual Operator
{
    private:
    Consumer *parent_; 
 
    public:
    Consumer * parent() const { return parent_; }
    Consumer * parent(Consumer *c) { std::swap(parent_, c); return c; }
};
 
struct M_EXPORT Consumer : virtual Operator
{
    private:
    std::vector<Producer*> children_; 
 
    public:
    Consumer() = default;
    Consumer(Consumer&&) = default;
    virtual ~Consumer() {
        for (auto c : children_)
            delete c;
    }
 
    virtual void add_child(Producer *child) {
        if (not child)
            throw invalid_argument("no child given");
        children_.push_back(child);
        child->parent(this);
        schema() += child->schema();
    }
    virtual Producer * set_child(Producer *child, std::size_t i) {
        if (not child)
            throw invalid_argument("no child given");
        if (i >= children_.size())
            throw out_of_range("index i out of bounds");
        auto old = children_[i];
        children_[i] = child;
        child->parent(this);
 
        /* Recompute operator schema. */
        auto &S = schema();
        S = Schema();
        for (auto c : children_)
            S += c->schema();
 
        return old;
    }
 
    const std::vector<Producer*> & children() const { return children_; }
 
    Producer * child(std::size_t i) const {
        if (i >= children_.size())
            throw out_of_range("index i out of bounds");
        return children_[i];
    }
 
    protected:
    std::vector<Producer*> & children() { return children_; }
 
    public:
    void assign_post_order_ids(std::size_t start_id = 0UL) const override {
        auto next_start_id = start_id;
        for (auto c : children_) {
            c->assign_post_order_ids(next_start_id);
            next_start_id = c->id() + 1;
        }
        id_ = next_start_id;
    }
    void reset_ids() const override {
        id_ = -1UL;
        for (auto c : children_)
            c->reset_ids();
    }
};
 
struct M_EXPORT CallbackOperator : Consumer
{
    using callback_type = std::function<void(const Schema &, const Tuple&)>;
 
    private:
    callback_type callback_;
 
    public:
    CallbackOperator(callback_type callback) : callback_(std::move(callback)) { }
 
    CallbackOperator clone_node() const { return CallbackOperator(callback_); }
 
    const auto & callback() const { return callback_; }
 
    void accept(OperatorVisitor &v) override;
    void accept(ConstOperatorVisitor &v) const override;
};
 
struct M_EXPORT PrintOperator : Consumer
{
    std::ostream &out;
 
    PrintOperator(std::ostream &out) : out(out) { }
 
    PrintOperator clone_node() const { return PrintOperator(out); }
 
    void accept(OperatorVisitor &v) override;
    void accept(ConstOperatorVisitor &v) const override;
};
 
struct M_EXPORT NoOpOperator : Consumer
{
    std::ostream &out;
 
    NoOpOperator(std::ostream &out) : out(out) { }
 
    NoOpOperator clone_node() const { return NoOpOperator(out); }
 
    void accept(OperatorVisitor &v) override;
    void accept(ConstOperatorVisitor &v) const override;
};
 
struct M_EXPORT ScanOperator : Producer
{
    private:
    const Store &store_;
    ThreadSafePooledString alias_;
 
    public:
    ScanOperator(const Store &store, ThreadSafePooledString alias)
        : store_(store)
        , alias_(std::move(alias))
    {
        auto &S = schema();
        for (auto &e : store.table().schema())
            S.add({alias_, e.id.name}, e.type, e.constraints);
    }
 
    ScanOperator clone_node() const { return ScanOperator(store_, alias_); }
 
    const Store & store() const { return store_; }
    const ThreadSafePooledString & alias() const { return alias_; }
 
    void accept(OperatorVisitor &v) override;
    void accept(ConstOperatorVisitor &v) const override;
};
 
struct M_EXPORT FilterOperator : Producer, Consumer
{
    protected:
    cnf::CNF filter_;
 
    public:
    FilterOperator(cnf::CNF filter) : filter_(std::move(filter)) { }
 
    FilterOperator clone_node() const { return FilterOperator(filter_); }
 
    const cnf::CNF & filter() const { return filter_; }
    const cnf::CNF filter(cnf::CNF f) {
        auto old_filter = std::move(filter_);
        filter_ = std::move(f);
        return old_filter;
    }
 
    void accept(OperatorVisitor &v) override;
    void accept(ConstOperatorVisitor &v) const override;
};
 
struct M_EXPORT DisjunctiveFilterOperator : FilterOperator
{
    DisjunctiveFilterOperator(cnf::CNF filter)
        : FilterOperator(std::move(filter))
    {
        M_insist(this->filter().size() == 1, "a disjunctive filter must have exactly one clause");
        M_insist(this->filter()[0].size() >= 2, "a disjunctive filter must have at least two predicates ");
    }
 
    DisjunctiveFilterOperator clone_node() const { return DisjunctiveFilterOperator(filter_); }
 
    void accept(OperatorVisitor &v) override;
    void accept(ConstOperatorVisitor &v) const override;
};
 
struct M_EXPORT JoinOperator : Producer, Consumer
{
    static constexpr Schema::entry_type::constraints_t REMOVED_CONSTRAINTS =
        Schema::entry_type::UNIQUE | Schema::entry_type::REFERENCES_UNIQUE; // FIXME: still unique for 1:1 joins and depends on what was referenced
 
    private:
    cnf::CNF predicate_;
 
    public:
    JoinOperator(cnf::CNF predicate) : predicate_(std::move(predicate)) { }
 
    /*----- Override child setters to correctly adapt the constraints of the computed schema! ------------------------*/
    virtual void add_child(Producer *child) override {
        const auto old_num_entries = schema().num_entries();
        Consumer::add_child(child); // delegate to inherited method
 
        /* Remove uniqueness constraints (in added schema part) due to denormalization. */
        for (auto i = old_num_entries; i != schema().num_entries(); ++i)
            schema()[i].constraints &= ~REMOVED_CONSTRAINTS;
    }
    virtual Producer * set_child(Producer *child, std::size_t i) override {
        auto old = Consumer::set_child(child, i); // delegate to inherited method
 
        /* Remove uniqueness constraints (in entire schema because of recomputation) due to denormalization. */
        for (auto &e : schema())
            e.constraints &= ~REMOVED_CONSTRAINTS;
 
        return old;
    }
 
    JoinOperator clone_node() const { return JoinOperator(predicate_); }
 
    const cnf::CNF & predicate() const { return predicate_; }
 
    void accept(OperatorVisitor &v) override;
    void accept(ConstOperatorVisitor &v) const override;
};
 
struct M_EXPORT ProjectionOperator : Producer, Consumer
{
    using projection_type = QueryGraph::projection_type;
 
    private:
    std::vector<projection_type> projections_;
 
    public:
    ProjectionOperator(std::vector<projection_type> projections);
 
    ProjectionOperator clone_node() const { return ProjectionOperator(projections_); }
 
    /*----- Override child setters to *NOT* modify the computed schema! ----------------------------------------------*/
    virtual void add_child(Producer *child) override {
        if (not child)
            throw invalid_argument("no child given");
        children().push_back(child);
        child->parent(this);
 
        /* Add constraints from child to computed schema. */
        auto &S = schema();
        for (std::size_t i = 0; i < projections_.size(); ++i) {
            if (auto D = cast<const ast::Designator>(projections_[i].first)) {
                Schema::Identifier id(D->table_name.text, D->attr_name.text.assert_not_none());
                S[i].constraints |= child->schema()[id].second.constraints;
            }
        }
    }
    virtual Producer * set_child(Producer *child, std::size_t i) override {
        if (not child)
            throw invalid_argument("no child given");
        if (i >= children().size())
            throw out_of_range("index i out of bounds");
        auto old = children()[i];
        children()[i] = child;
        child->parent(this);
 
        /* Add constraints from child to computed schema. */
        auto &S = schema();
        for (std::size_t i = 0; i < projections_.size(); ++i) {
            if (auto D = cast<const ast::Designator>(projections_[i].first)) {
                Schema::Identifier id(D->table_name.text, D->attr_name.text.assert_not_none());
                S[i].constraints |= child->schema()[id].second.constraints;
            }
        }
 
        return old;
    }
 
    const std::vector<projection_type> & projections() const { return projections_; }
    std::vector<projection_type> & projections() { return projections_; }
 
    void accept(OperatorVisitor &v) override;
    void accept(ConstOperatorVisitor &v) const override;
};
 
struct M_EXPORT LimitOperator : Producer, Consumer
{
    /* This class is used to unwind the stack when the limit of produced tuples is reached. */
    struct stack_unwind : std::exception { };
 
    private:
    std::size_t limit_;
    std::size_t offset_;
 
    public:
    LimitOperator(std::size_t limit, std::size_t offset)
        : limit_(limit)
        , offset_(offset)
    { }
 
    LimitOperator clone_node() const { return LimitOperator(limit_, offset_); }
 
    std::size_t limit() const { return limit_; }
    std::size_t offset() const { return offset_; }
 
    void accept(OperatorVisitor &v) override;
    void accept(ConstOperatorVisitor &v) const override;
};
 
struct M_EXPORT GroupingOperator : Producer, Consumer
{
    using group_type = QueryGraph::group_type;
 
    private:
    std::vector<group_type> group_by_; 
    std::vector<std::reference_wrapper<const ast::FnApplicationExpr>> aggregates_; 
 
    public:
    GroupingOperator(std::vector<group_type> group_by,
                     std::vector<std::reference_wrapper<const ast::FnApplicationExpr>> aggregates);
 
    GroupingOperator clone_node() const { return GroupingOperator(group_by_, aggregates_); }
 
    /*----- Override child setters to *NOT* modify the computed schema! ----------------------------------------------*/
    virtual void add_child(Producer *child) override {
        if (not child)
            throw invalid_argument("no child given");
        children().push_back(child);
        child->parent(this);
 
        /* Add constraints from child to computed schema. */
        auto &S = schema();
        for (std::size_t i = 0; i < group_by_.size(); ++i) {
            if (auto D = cast<const ast::Designator>(group_by_[i].first)) {
                Schema::Identifier id(D->table_name.text, D->attr_name.text.assert_not_none());
                S[i].constraints |= child->schema()[id].second.constraints;
            }
        }
    }
    virtual Producer * set_child(Producer *child, std::size_t i) override {
        if (not child)
            throw invalid_argument("no child given");
        if (i >= children().size())
            throw out_of_range("index i out of bounds");
        auto old = children()[i];
        children()[i] = child;
        child->parent(this);
 
        /* Add constraints from child to computed schema. */
        auto &S = schema();
        for (std::size_t i = 0; i < group_by_.size(); ++i) {
            if (auto D = cast<const ast::Designator>(group_by_[i].first)) {
                Schema::Identifier id(D->table_name.text, D->attr_name.text.assert_not_none());
                S[i].constraints |= child->schema()[id].second.constraints;
            }
        }
 
        return old;
    }
 
    const auto & group_by() const { return group_by_; }
    const auto & aggregates() const { return aggregates_; }
    auto & aggregates() { return aggregates_; }
 
    void accept(OperatorVisitor &v) override;
    void accept(ConstOperatorVisitor &v) const override;
};
 
struct M_EXPORT AggregationOperator : Producer, Consumer
{
    private:
    std::vector<std::reference_wrapper<const ast::FnApplicationExpr>> aggregates_; 
 
    public:
    AggregationOperator(std::vector<std::reference_wrapper<const ast::FnApplicationExpr>> aggregates);
 
    AggregationOperator clone_node() const { return AggregationOperator(aggregates_); }
 
    /*----- Override child setters to *NOT* modify the computed schema! ----------------------------------------------*/
    virtual void add_child(Producer *child) override {
        if (not child)
            throw invalid_argument("no child given");
        children().push_back(child);
        child->parent(this);
    }
    virtual Producer * set_child(Producer *child, std::size_t i) override {
        if (not child)
            throw invalid_argument("no child given");
        if (i >= children().size())
            throw out_of_range("index i out of bounds");
        auto old = children()[i];
        children()[i] = child;
        child->parent(this);
        return old;
    }
 
    const auto & aggregates() const { return aggregates_; }
    auto & aggregates() { return aggregates_; }
 
    void accept(OperatorVisitor &v) override;
    void accept(ConstOperatorVisitor &v) const override;
};
 
struct M_EXPORT SortingOperator : Producer, Consumer
{
    using order_type = QueryGraph::order_type;
 
    private:
    std::vector<order_type> order_by_; 
 
    public:
    SortingOperator(std::vector<order_type> order_by) : order_by_(std::move(order_by)) { }
 
    SortingOperator clone_node() const { return SortingOperator(order_by_); }
 
    const auto & order_by() const { return order_by_; }
 
    void accept(OperatorVisitor &v) override;
    void accept(ConstOperatorVisitor &v) const override;
};
 
#define M_OPERATOR_LIST(X) \
    X(ScanOperator) \
    X(CallbackOperator) \
    X(PrintOperator) \
    X(NoOpOperator) \
    X(FilterOperator) \
    X(DisjunctiveFilterOperator) \
    X(JoinOperator) \
    X(ProjectionOperator) \
    X(LimitOperator) \
    X(GroupingOperator) \
    X(AggregationOperator) \
    X(SortingOperator)
 
M_DECLARE_VISITOR(OperatorVisitor, Operator, M_OPERATOR_LIST)
M_DECLARE_VISITOR(ConstOperatorVisitor, const Operator, M_OPERATOR_LIST)
 
namespace {
 
template<bool C>
struct ThePreOrderOperatorVisitor : std::conditional_t<C, ConstOperatorVisitor, OperatorVisitor>
{
    using super = std::conditional_t<C, ConstOperatorVisitor, OperatorVisitor>;
    template<typename T> using Const = typename super::template Const<T>;
    void operator()(Const<Operator> &op) override {
        try { op.accept(*this); } catch (visit_skip_subtree) { return; }
        if (auto c = cast<Const<Consumer>>(&op)) {
            for (auto child : c->children())
                (*this)(*child);
        }
    }
};
 
template<bool C>
struct ThePostOrderOperatorVisitor : std::conditional_t<C, ConstOperatorVisitor, OperatorVisitor>
{
    using super = std::conditional_t<C, ConstOperatorVisitor, OperatorVisitor>;
    template<typename T> using Const = typename super::template Const<T>;
    void operator()(Const<Operator> &op) override {
        if (auto c = cast<Const<Consumer>>(&op)) {
            for (auto child : c->children())
                (*this)(*child);
        }
        op.accept(*this);
    }
};
 
}
using PreOrderOperatorVisitor = ThePreOrderOperatorVisitor<false>;
using ConstPreOrderOperatorVisitor = ThePreOrderOperatorVisitor<true>;
using PostOrderOperatorVisitor = ThePostOrderOperatorVisitor<false>;
using ConstPostOrderOperatorVisitor = ThePostOrderOperatorVisitor<true>;
 
M_MAKE_STL_VISITABLE(PreOrderOperatorVisitor, Operator, M_OPERATOR_LIST)
M_MAKE_STL_VISITABLE(ConstPreOrderOperatorVisitor, const Operator, M_OPERATOR_LIST)
M_MAKE_STL_VISITABLE(PostOrderOperatorVisitor, Operator, M_OPERATOR_LIST)
M_MAKE_STL_VISITABLE(ConstPostOrderOperatorVisitor, const Operator, M_OPERATOR_LIST)
 
 
enum class OperatorKind
{
#define X(Kind) Kind,
    M_OPERATOR_LIST(X)
#undef X
};
 
}