QueryGraph.hpp

Go to the documentation of this file.

#pragma once
 
#include <cstdint>
#include <cstring>
#include <deque>
#include <functional>
#include <memory>
#include <mutable/catalog/Scheduler.hpp>
#include <mutable/IR/CNF.hpp>
#include <mutable/mutable-config.hpp>
#include <mutable/util/AdjacencyMatrix.hpp>
#include <mutable/util/ADT.hpp>
#include <sstream>
#include <string>
#include <vector>
 
 
namespace m {
 
namespace ast {
 
struct Stmt;
 
}
 
struct DataSource;
struct Join;
struct QueryGraph;
 
struct M_EXPORT DataSource
{
    friend struct QueryGraph;
    friend struct GraphBuilder;
    friend struct Decorrelation;
 
    private:
    cnf::CNF filter_; 
    std::vector<std::reference_wrapper<Join>> joins_; 
    ThreadSafePooledOptionalString alias_; 
    std::size_t id_; 
 
    bool decorrelated_ = true; 
 
    protected:
    DataSource(std::size_t id, ThreadSafePooledOptionalString alias) : alias_(std::move(alias)), id_(id) {
        if (alias_.has_value() and strlen(*alias_) == 0)
            throw invalid_argument("if the data source has an alias, it must not be empty");
    }
 
    public:
    virtual ~DataSource() { }
 
    std::size_t id() const { return id_; }
    const ThreadSafePooledOptionalString & alias() const { return alias_; }
    virtual ThreadSafePooledOptionalString name() const = 0;
    const cnf::CNF & filter() const { return filter_; }
    void update_filter(cnf::CNF filter) { filter_ = filter_ and filter; }
    void add_join(Join &join) { joins_.emplace_back(join); }
    const auto & joins() const { return joins_; }
 
    virtual bool is_correlated() const = 0;
 
    private:
    void remove_join(Join &join) {
        auto it = std::find_if(joins_.begin(), joins_.end(), [&join](auto j) { return &j.get() == &join; });
        if (it == joins_.end())
            throw invalid_argument("given join not found");
        joins_.erase(it);
    }
 
    public:
    bool operator==(const DataSource &other) const { return this->id_ == other.id_; }
    bool operator!=(const DataSource &other) const { return not operator==(other); }
};
 
struct M_EXPORT BaseTable : DataSource
{
    friend struct QueryGraph;
    friend struct GetPrimaryKey;
 
    private:
    const Table &table_; 
    // std::vector<const ast::Designator*> expansion_;
 
    private:
    BaseTable(std::size_t id, ThreadSafePooledOptionalString alias, const Table &table)
        : DataSource(id, std::move(alias)), table_(table)
    { }
 
    public:
    const Table & table() const { return table_; }
 
    ThreadSafePooledOptionalString name() const override { return alias().has_value() ? alias() : table_.name(); }
 
    bool is_correlated() const override { return false; };
};
 
struct M_EXPORT Query : DataSource
{
    friend struct QueryGraph;
 
    private:
    std::unique_ptr<QueryGraph> query_graph_; 
 
    private:
    Query(std::size_t id, ThreadSafePooledOptionalString alias, std::unique_ptr<QueryGraph> query_graph)
        : DataSource(id, std::move(alias)), query_graph_(std::move(query_graph))
    { }
 
    public:
    QueryGraph & query_graph() const { return *query_graph_; }
 
    ThreadSafePooledOptionalString name() const override { return alias(); }
 
    bool is_correlated() const override;
};
 
struct M_EXPORT Join
{
    using sources_t = std::vector<std::reference_wrapper<DataSource>>;
 
    private:
    cnf::CNF condition_; 
    sources_t sources_; 
 
    public:
    Join(cnf::CNF condition, sources_t sources) : condition_(std::move(condition)) , sources_(std::move(sources)) { }
 
    const cnf::CNF & condition() const { return condition_; }
    void update_condition(cnf::CNF update) { condition_ = condition_ and update; }
    const sources_t & sources() const { return sources_; }
 
    bool operator==(const Join &other) const {
        if (this->condition_ != other.condition_) return false;
        if (this->sources().size() != other.sources().size()) return false;
        for (auto &this_src : sources_) {
            auto it = std::find_if(other.sources().begin(), other.sources().end(), [this_src](auto other_src) {
                return &this_src.get() == &other_src.get();
            });
            if (it == other.sources().end()) return false;
        }
        return true;
    }
    bool operator!=(const Join &other) const { return not operator==(other); }
};
 
struct M_EXPORT QueryGraph
{
    friend struct GraphBuilder;
    friend struct Decorrelation;
    friend struct GetPrimaryKey;
 
    using Subproblem = SmallBitset; 
    using projection_type = std::pair<std::reference_wrapper<const ast::Expr>, ThreadSafePooledOptionalString>;
    using order_type = std::pair<std::reference_wrapper<const ast::Expr>, bool>; 
    using group_type = std::pair<std::reference_wrapper<const ast::Expr>, ThreadSafePooledOptionalString>;
 
    private:
    std::vector<std::unique_ptr<DataSource>> sources_; 
    std::vector<std::unique_ptr<Join>> joins_; 
 
    std::vector<group_type> group_by_; 
    std::vector<std::reference_wrapper<const ast::FnApplicationExpr>> aggregates_; 
    std::vector<projection_type> projections_; 
    std::vector<order_type> order_by_; 
    struct { uint64_t limit = 0, offset = 0; } limit_; 
 
    mutable std::unique_ptr<AdjacencyMatrix> adjacency_matrix_;
 
    Scheduler::Transaction *t_ = nullptr; 
    std::vector<std::unique_ptr<ast::Expr>> custom_filter_exprs_;
 
    public:
    friend void swap(QueryGraph &first, QueryGraph &second) {
        using std::swap;
        swap(first.sources_,                second.sources_);
        swap(first.joins_,                  second.joins_);
        swap(first.group_by_,               second.group_by_);
        swap(first.aggregates_,             second.aggregates_);
        swap(first.projections_,            second.projections_);
        swap(first.order_by_,               second.order_by_);
        swap(first.limit_,                  second.limit_);
        swap(first.adjacency_matrix_,       second.adjacency_matrix_);
        swap(first.t_,                      second.t_);
        swap(first.custom_filter_exprs_,    second.custom_filter_exprs_);
    }
 
    QueryGraph();
    ~QueryGraph();
 
    QueryGraph(const QueryGraph&) = delete;
    QueryGraph(QueryGraph &&other) : QueryGraph() { swap(*this, other); }
 
    QueryGraph & operator=(QueryGraph &&other) { swap(*this, other); return *this; }
 
    static std::unique_ptr<QueryGraph> Build(const ast::Stmt &stmt);
 
    std::size_t num_sources() const { return sources_.size(); }
 
    std::size_t num_joins() const { return joins_.size(); }
 
    void add_source(std::unique_ptr<DataSource> source) {
        source->id_ = sources_.size();
        sources_.emplace_back(source.release());
    }
    BaseTable & add_source(ThreadSafePooledOptionalString alias, const Table &table) {
        std::unique_ptr<BaseTable> base(new BaseTable(sources_.size(), std::move(alias), table));
        auto &ref = sources_.emplace_back(std::move(base));
        return as<BaseTable>(*ref);
    }
    Query & add_source(ThreadSafePooledOptionalString alias, std::unique_ptr<QueryGraph> query_graph) {
        std::unique_ptr<Query> Q(new Query(sources_.size(), std::move(alias), std::move(query_graph)));
        auto &ref = sources_.emplace_back(std::move(Q));
        return as<Query>(*ref);
    }
 
    std::unique_ptr<DataSource> remove_source(std::size_t id) {
        auto it = std::next(sources_.begin(), id);
        auto ds = std::move(*it);
        M_insist(ds->id() == id, "IDs of sources must be sequential");
        sources_.erase(it);
        /* Decrement IDs of all sources after the deleted one to provide sequential IDs. */
        while (it != sources_.end()) {
            --(*it)->id_;
            ++it;
        }
        return ds;
    }
 
    const auto & sources() const { return sources_; }
    const auto & joins() const { return joins_; }
    const auto & group_by() const { return group_by_; }
    const auto & aggregates() const { return aggregates_; }
    const std::vector<projection_type> & projections() const { return projections_; }
    const auto & order_by() const { return order_by_; }
    auto limit() const { return limit_; }
 
    const DataSource & operator[](uint64_t id) const {
        auto &ds = sources_[id];
        M_insist(ds->id() == id, "given id and data source id must match");
        return *ds;
    }
 
    bool grouping() const { return not group_by_.empty() or not aggregates_.empty(); }
    bool is_correlated() const;
 
    AdjacencyMatrix & adjacency_matrix() {
        if (not adjacency_matrix_) [[unlikely]]
            compute_adjacency_matrix();
        return *adjacency_matrix_;
    }
 
    const AdjacencyMatrix & adjacency_matrix() const {
        if (not adjacency_matrix_) [[unlikely]]
            compute_adjacency_matrix();
        return *adjacency_matrix_;
    }
 
    void dot(std::ostream &out) const;
 
    void sql(std::ostream &out) const;
 
    void transaction(Scheduler::Transaction *t) {
        M_insist(not t_ or *t_ == *t, "QueryGraph is already linked to another transaction");
        t_ = t;
 
        for (auto &ds : sources_)
            if (auto bt = cast<const Query>(ds.get()))
                bt->query_graph_->transaction(t_);
    }
    Scheduler::Transaction * transaction() const { return t_; }
 
    void add_custom_filter(std::unique_ptr<ast::Expr> filter_expr, DataSource &ds) {
        M_insist(std::find_if(sources_.begin(), sources_.end(), [&](std::unique_ptr<DataSource> &source){
            return *source == ds;
        }) != sources_.end());
 
        auto filter = cnf::to_CNF(*filter_expr);
        ds.update_filter(filter);
        custom_filter_exprs_.push_back(std::move(filter_expr));
    }
 
 
    void dump(std::ostream &out) const;
    void dump() const;
 
    private:
    void compute_adjacency_matrix() const;
    void dot_recursive(std::ostream &out) const;
 
    void remove_join(Join &join) {
        auto it = std::find_if(joins_.begin(), joins_.end(), [&join](auto &j) { return j.get() == &join; });
        if (it == joins_.end())
            throw invalid_argument("given join not found");
        joins_.erase(it);
    }
};
 
}