Schema.hpp

Go to the documentation of this file.

#pragma once
 
#include <cmath>
#include <cstring>
#include <functional>
#include <iosfwd>
#include <iterator>
#include <list>
#include <memory>
#include <mutable/catalog/CardinalityEstimator.hpp>
#include <mutable/catalog/Type.hpp>
#include <mutable/mutable-config.hpp>
#include <mutable/storage/DataLayout.hpp>
#include <mutable/storage/Index.hpp>
#include <mutable/storage/Store.hpp>
#include <mutable/util/ADT.hpp>
#include <mutable/util/enum_ops.hpp>
#include <mutable/util/exception.hpp>
#include <mutable/util/fn.hpp>
#include <mutable/util/macro.hpp>
#include <type_traits>
#include <unordered_map>
#include <utility>
#include <vector>
 
 
namespace m {
 
namespace storage {
 
// forward declarations
struct DataLayoutFactory;
 
}
 
struct M_EXPORT Schema
{
    struct Identifier
    {
        private:
        static Identifier CONST_ID_;
 
        public:
        ThreadSafePooledOptionalString prefix; 
        ThreadSafePooledString name;           
 
        Identifier(ThreadSafePooledString name) : name(std::move(name)) { }
        Identifier(ThreadSafePooledOptionalString prefix, ThreadSafePooledString name)
            : prefix(std::move(prefix)), name(std::move(name))
        {
            if (this->prefix.has_value() and strlen(*this->prefix) == 0)
                throw invalid_argument("prefix must not be the empty string");
        }
        explicit Identifier(const ast::Expr&);
 
        static Identifier GetConstant();
        bool is_constant() const { return operator==(GetConstant()); }
 
        bool operator==(const Identifier &other) const {
            return this->prefix == other.prefix and this->name == other.name;
        }
        bool operator!=(const Identifier &other) const { return not operator==(other); }
 
M_LCOV_EXCL_START
        friend std::ostream & operator<<(std::ostream &out, const Identifier &id) {
            if (id.prefix.has_value())
                out << id.prefix << '.';
            return out << id.name;
        }
M_LCOV_EXCL_STOP
    };
 
    struct entry_type
    {
        enum constraints_t : uint64_t
        {
            NOT_NULLABLE = 0b1U,           
            UNIQUE = 0b10U,                
            REFERENCES_UNIQUE = 0b100U,    
            IS_HIDDEN = 0b1000U,           
        };
 
        Identifier id;
        const Type *type;
        constraints_t constraints;
 
        private:
        entry_type();
 
        public:
        entry_type(Identifier id, const Type *type, constraints_t constraints = constraints_t{0})
            : id(std::move(id))
            , type(M_notnull(type))
            , constraints(constraints)
        { }
 
        static entry_type CreateArtificial() { return entry_type(); }
 
        bool nullable() const { return not (NOT_NULLABLE & constraints); }
        bool unique() const { return bool(UNIQUE & constraints); }
        bool references_unique() const { return bool(REFERENCES_UNIQUE & constraints); }
    };
 
    private:
    std::vector<entry_type> entries_;
 
    public:
    using iterator = decltype(entries_)::iterator;
    using const_iterator = decltype(entries_)::const_iterator;
 
    const std::vector<entry_type> & entries() const { return entries_; }
 
    iterator begin() { return entries_.begin(); }
    iterator end()   { return entries_.end(); }
    const_iterator begin() const { return entries_.cbegin(); }
    const_iterator end()   const { return entries_.cend(); }
    const_iterator cbegin() const { return entries_.cbegin(); }
    const_iterator cend()   const { return entries_.cend(); }
 
    std::size_t num_entries() const { return entries_.size(); }
 
    bool empty() const { return entries_.empty(); }
 
    iterator find(const Identifier &id) {
        auto pred = [&id](const entry_type &e) -> bool { return e.id == id; }; // match qualified
        auto it = std::find_if(begin(), end(), pred);
        if (it != end() and std::find_if(std::next(it), end(), pred) != end())
            throw invalid_argument("duplicate identifier, lookup ambiguous");
        return it;
    }
    const_iterator find(const Identifier &id) const { return const_cast<Schema*>(this)->find(id); }
 
    bool has(const Identifier &id) const { return find(id) != end(); }
 
    entry_type & at(std::size_t idx) {
        if (idx >= entries_.size())
            throw out_of_range("index out of bounds");
        return entries_[idx];
    }
    const entry_type & at(std::size_t idx) const { return const_cast<Schema*>(this)->at(idx); }
    entry_type & operator[](std::size_t idx) {
        M_insist(idx < entries_.size(), "index out of bounds");
        return entries_[idx];
    }
    const entry_type & operator[](std::size_t idx) const { return const_cast<Schema*>(this)->operator[](idx); }
 
    std::pair<std::size_t, entry_type&> at(const Identifier &id) {
        auto pos = find(id);
        if (pos == end())
            throw out_of_range("identifier not found");
        return { std::distance(begin(), pos), *pos };
    }
    std::pair<std::size_t, const entry_type&> at(const Identifier &id) const { return const_cast<Schema*>(this)->at(id); }
    std::pair<std::size_t, entry_type&> operator[](const Identifier &id) {
        auto pos = find(id);
        M_insist(pos != end(), "identifier not found");
        return { std::distance(begin(), pos), *pos };
    }
    std::pair<std::size_t, const entry_type&> operator[](const Identifier &id) const {
        return const_cast<Schema*>(this)->operator[](id);
    }
 
    void add(entry_type e) { entries_.emplace_back(std::move(e)); }
    void add(Identifier id, const Type *type) { entries_.emplace_back(std::move(id), type); }
    void add(Identifier id, const Type *type, entry_type::constraints_t constraints) {
        entries_.emplace_back(std::move(id), type, constraints);
    }
 
    Schema deduplicate() const {
        Schema res;
        for (auto &e : *this) {
            if (not res.has(e.id))
                res.add(e.id, e.type, e.constraints);
        }
        return res;
    }
 
    Schema drop_constants() const {
        Schema res;
        for (auto &e : *this) {
            if (not e.id.is_constant())
                res.add(e.id, e.type, e.constraints);
        }
        return res;
    }
 
    Schema & operator+=(const Schema &other) {
        for (auto &e : other)
            entries_.emplace_back(e);
        return *this;
    }
 
    Schema & operator|=(const Schema &other) {
        for (auto &e : other) {
            if (not has(e.id))
                entries_.emplace_back(e);
        }
        return *this;
    }
 
    bool operator==(const Schema &other) const {
        return std::all_of(this->begin(), this->end(), [&](const entry_type &p) { return other.has(p.id); }) and
               std::all_of(other.begin(), other.end(), [&](const entry_type &p) { return this->has(p.id); });
    }
    bool operator!=(const Schema &other) const { return not operator==(other); }
 
M_LCOV_EXCL_START
    friend std::ostream & operator<<(std::ostream &out, const Schema &schema) {
        out << "{[";
        for (auto it = schema.begin(), end = schema.end(); it != end; ++it) {
            if (it != schema.begin()) out << ',';
            out << ' ' << it->id << " :" << *it->type;
        }
        return out << " ]}";
    }
M_LCOV_EXCL_STOP
 
    void dump(std::ostream &out) const;
    void dump() const;
};
 
inline Schema operator+(const Schema &left, const Schema &right)
{
    Schema S(left);
    S += right;
    return S;
}
 
inline Schema operator&(const Schema &left, const Schema &right)
{
    Schema res;
    for (auto &e : left) {
        auto it = right.find(e.id);
        if (it != right.end()) {
            if (e.type != it->type)
                throw invalid_argument("type mismatch");
            res.add(e.id, e.type, e.constraints | it->constraints); // merge constraints from both
        }
    }
    return res;
}
 
inline Schema operator|(const Schema &left, const Schema &right)
{
    Schema res(left);
    res |= right;
    return res;
}
 
 
/*======================================================================================================================
 * Attribute, Table, Function, Database
 *====================================================================================================================*/
 
struct ConcreteTable;
struct TableDecorator;
 
struct M_EXPORT Attribute
{
    friend struct ConcreteTable;
 
    std::size_t id; 
    const Table &table; 
    const PrimitiveType *type; 
    ThreadSafePooledString name; 
    bool not_nullable = false; 
    bool unique = false;
    bool is_hidden = false; 
    const Attribute *reference = nullptr; 
 
    private:
    explicit Attribute(std::size_t id, const Table &table, const PrimitiveType *type, ThreadSafePooledString name)
        : id(id)
        , table(table)
        , type(M_notnull(type))
        , name(std::move(name))
    {
        if (not type->is_vectorial())
            throw invalid_argument("attributes must be of vectorial type");
    }
 
    public:
    Attribute(const Attribute&) = delete;
    Attribute(Attribute&&) = default;
 
    bool is_unique() const;
 
    bool operator==(const Attribute &other) const { return &this->table == &other.table and this->id == other.id; }
    bool operator!=(const Attribute &other) const { return not operator==(other); }
 
M_LCOV_EXCL_START
    friend std::ostream & operator<<(std::ostream &out, const Attribute &attr) {
        return out << '`' << attr.name << "` " << *attr.type;
    }
M_LCOV_EXCL_STOP
 
    void dump(std::ostream &out) const;
    void dump() const;
};
 
template<typename T>
bool type_check(const Attribute &attr)
{
    auto ty = attr.type;
 
    /* Boolean */
    if constexpr (std::is_same_v<T, bool>) {
        if (is<const Boolean>(ty))
            return true;
    }
 
    /* CharacterSequence */
    if constexpr (std::is_same_v<T, std::string>) {
        if (auto s = cast<const CharacterSequence>(ty)) {
            if (not s->is_varying)
                return true;
        }
    }
    if constexpr (std::is_same_v<T, const char*>) {
        if (auto s = cast<const CharacterSequence>(ty)) {
            if (not s->is_varying)
                return true;
        }
    }
 
    /* Numeric */
    if constexpr (std::is_arithmetic_v<T>) {
        if (auto n = cast<const Numeric>(ty)) {
            switch (n->kind) {
                case Numeric::N_Int:
                    if (std::is_integral_v<T> and sizeof(T) * 8 == ty->size())
                        return true;
                    break;
 
                case Numeric::N_Float:
                    if (std::is_floating_point_v<T> and sizeof(T) * 8 == ty->size())
                        return true;
                    break;
 
                case Numeric::N_Decimal:
                    if (std::is_integral_v<T> and ceil_to_pow_2(ty->size()) == 8 * sizeof(T))
                        return true;
                    break;
            }
        }
    }
 
    return false;
}
 
struct M_EXPORT Table
{
    protected:
    using table_type = std::vector<Attribute>;
 
    template<bool V, bool H>
    struct the_iterator
    {
        static constexpr bool Show_Visible = V;
        static constexpr bool Show_Hidden = H;
 
 
        using value_type = Attribute;
        using it_type = table_type::const_iterator;
        using difference_type = std::ptrdiff_t;
        using pointer = const value_type*;
        using reference = const value_type&;
 
        private:
        it_type it_, start_, end_;
 
        public:
        the_iterator() = default;
        the_iterator(it_type start, it_type end) : it_(start), start_(start), end_(end) {
            if constexpr (Show_Visible and not Show_Hidden)
                while (it_ != end_ and it_->is_hidden) {
                    ++it_;
                }
            else if constexpr (not Show_Visible and Show_Hidden)
                while (it_ != end_ and not it_->is_hidden) {
                    ++it_;
                }
            else if constexpr (not Show_Visible and not Show_Hidden)
                it_ = end_;
        }
        the_iterator(it_type it, it_type start, it_type end) : it_(it), start_(start), end_(end) { }
 
        the_iterator & operator++() {
            ++it_;
            if constexpr (Show_Visible and not Show_Hidden)
                while (it_ != end_ and it_->is_hidden) {
                    ++it_;
                }
            else if constexpr (not Show_Visible and Show_Hidden)
                while (it_ != end_ and not it_->is_hidden) {
                    ++it_;
                }
            return *this;
        }
 
        the_iterator operator++(int) { the_iterator clone = *this; operator++(); return clone; }
 
        the_iterator & operator--() {
            --it_;
            if constexpr (Show_Visible and not Show_Hidden)
                while (it_ != start_ and it_->is_hidden) {
                    --it_;
                }
            else if constexpr (not Show_Visible and Show_Hidden)
                while (it_ != start_ and not it_->is_hidden) {
                    --it_;
                }
            return *this;
        }
 
        the_iterator operator--(int) { the_iterator clone = *this; operator--(); return clone; }
 
        reference operator*() const { return *it_; }
        pointer operator->() const { return it_.operator->(); }
 
        bool operator==(const the_iterator &other) const {
            return it_ == other.it_ and start_ == other.start_ and end_ == other.end_;
        }
        bool operator!=(const the_iterator &other) const { return not operator==(other); }
 
        the_iterator & operator+=(int offset) {
            if constexpr (Show_Visible and Show_Hidden) {
                it_ += offset;
                return *this;
            }
 
            for (size_t i = 0 ; i < offset; ++i) {
                ++it_;
                if constexpr (Show_Visible and not Show_Hidden)
                    while (it_ != end_ and it_->is_hidden) {
                        ++it_;
                    }
                else if constexpr (not Show_Visible and Show_Hidden)
                    while (it_ != end_ and not it_->is_hidden) {
                        ++it_;
                    }
            }
            return *this;
        }
        the_iterator & operator-=(int offset) {
            if constexpr (Show_Visible and Show_Hidden) {
                it_ -= offset;
                return *this;
            }
 
            for (size_t i = 0 ; i < offset; ++i) {
                --it_;
                if constexpr (Show_Visible and not Show_Hidden)
                    while (it_ != start_ and it_->is_hidden) {
                        --it_;
                    }
                else if constexpr (not Show_Visible and Show_Hidden)
                    while (it_ != start_ and not it_->is_hidden) {
                        --it_;
                    }
            }
            return *this;
        }
 
        difference_type operator-(the_iterator other) const {
            if constexpr (Show_Visible and Show_Hidden) return this->it_ - other.it_;
            if (this->it_ - other.it_ == 0) return 0;
 
            auto smaller_it = (this->it_ < other.it_) ? this->it_ : other.it_;
            auto larger_it = (this->it_ < other.it_) ? other.it_ : this->it_;
            difference_type ignored = 0;
            for (auto i = smaller_it; i != larger_it; ++i) {
                if constexpr (Show_Visible and not Show_Hidden) {
                    if (i != end_ and i->is_hidden) ignored++;
                } else if constexpr (not Show_Visible and Show_Hidden) {
                    if (i != end_ and not i->is_hidden) ignored++;
                }
            }
            difference_type distance = this->it_ - other.it_;
            distance += (distance > 0) ? -ignored : ignored;
            return distance;
        }
    };
 
    public:
    using iterator = the_iterator<true, false>;
    using hidden_iterator = the_iterator<false, true>;
    using all_iterator = the_iterator<true, true>;
 
    virtual ~Table() = default;
 
    virtual std::size_t num_attrs() const = 0;
    virtual std::size_t num_hidden_attrs() const = 0;
    virtual std::size_t num_all_attrs() const = 0;
 
    virtual iterator begin()  const = 0;
    virtual iterator end()    const = 0;
    virtual iterator cbegin() const = 0;
    virtual iterator cend()   const = 0;
 
    virtual hidden_iterator begin_hidden()  const = 0;
    virtual hidden_iterator end_hidden()    const = 0;
    virtual hidden_iterator cbegin_hidden() const = 0;
    virtual hidden_iterator cend_hidden()   const = 0;
 
    virtual all_iterator begin_all()  const = 0;
    virtual all_iterator end_all()    const = 0;
    virtual all_iterator cbegin_all() const = 0;
    virtual all_iterator cend_all()   const = 0;
 
    virtual Attribute & at(std::size_t id) = 0;
    virtual const Attribute & at(std::size_t id) const = 0;
    virtual Attribute & operator[](std::size_t id) = 0;
    virtual const Attribute & operator[](std::size_t id) const = 0;
 
    virtual Attribute & at(const ThreadSafePooledString &name) = 0;
    virtual const Attribute & at(const ThreadSafePooledString &name) const = 0;
    virtual Attribute & operator[](const ThreadSafePooledString &name) = 0;
    virtual const Attribute & operator[](const ThreadSafePooledString &name) const = 0;
 
    virtual bool has_attribute(const ThreadSafePooledString &name) const = 0;
 
    virtual bool operator== (const Table &other) = 0;
    virtual bool operator== (const Table &other) const = 0;
 
    virtual const ThreadSafePooledString & name() const = 0;
 
    virtual Store & store() const = 0;
    virtual void store(std::unique_ptr<Store> new_store) = 0;
 
    virtual const storage::DataLayout & layout() const = 0;
    virtual void layout(storage::DataLayout &&new_layout) = 0;
    virtual void layout(const storage::DataLayoutFactory &factory) = 0;
 
    virtual std::vector<std::reference_wrapper<const Attribute>> primary_key() const = 0;
 
    virtual void add_primary_key(const ThreadSafePooledString &name) = 0;
 
    virtual void push_back(ThreadSafePooledString name, const PrimitiveType *type) = 0;
 
    virtual Schema schema(const ThreadSafePooledOptionalString &alias = {}) const = 0;
 
    virtual size_t convert_id(size_t id) = 0;
 
    virtual void dump(std::ostream &out) const = 0;
    virtual void dump() const = 0;
};
 
struct M_EXPORT ConcreteTable : Table
{
    private:
    ThreadSafePooledString name_; 
    table_type attrs_; 
    std::unordered_map<ThreadSafePooledString, table_type::size_type> name_to_attr_; 
    std::unique_ptr<Store> store_; 
    storage::DataLayout layout_; 
    SmallBitset primary_key_; 
 
    public:
    ConcreteTable(ThreadSafePooledString name) : name_(std::move(name)) { }
    virtual ~ConcreteTable() = default;
 
    std::size_t num_attrs() const override { return end() - begin(); }
 
    std::size_t num_hidden_attrs() const override { return end_hidden() - begin_hidden(); }
 
    std::size_t num_all_attrs() const override { return attrs_.size(); }
 
    virtual iterator begin()  const override { return iterator(attrs_.begin(), attrs_.end()); }
    virtual iterator end()    const override { return iterator(attrs_.end(), attrs_.begin(), attrs_.end()); }
    virtual iterator cbegin() const override { return iterator(attrs_.begin(), attrs_.end()); }
    virtual iterator cend()   const override { return iterator(attrs_.end(), attrs_.begin(), attrs_.end()); }
 
    virtual hidden_iterator begin_hidden()  const override { return hidden_iterator(attrs_.begin(), attrs_.end()); }
    virtual hidden_iterator end_hidden()    const override { return hidden_iterator(attrs_.end(), attrs_.begin(), attrs_.end()); }
    virtual hidden_iterator cbegin_hidden() const override { return hidden_iterator(attrs_.begin(), attrs_.end()); }
    virtual hidden_iterator cend_hidden()   const override { return hidden_iterator(attrs_.end(), attrs_.begin(), attrs_.end()); }
 
    virtual all_iterator begin_all()  const override { return all_iterator(attrs_.begin(), attrs_.end()); }
    virtual all_iterator end_all()    const override { return all_iterator(attrs_.end(), attrs_.begin(), attrs_.end()); }
    virtual all_iterator cbegin_all() const override { return all_iterator(attrs_.begin(), attrs_.end()); }
    virtual all_iterator cend_all()   const override { return all_iterator(attrs_.end(), attrs_.begin(), attrs_.end()); }
 
    Attribute & at(std::size_t id) override {
        if (id >= attrs_.size())
            throw std::out_of_range("id out of bounds");
        auto &attr = attrs_[id];
        M_insist(attr.id == id, "attribute ID mismatch");
        return attr;
    }
    const Attribute & at(std::size_t id) const override { return const_cast<ConcreteTable*>(this)->at(id); }
    Attribute & operator[](std::size_t id) override {
        M_insist(id < attrs_.size());
        auto &attr = attrs_[id];
        M_insist(attr.id == id, "attribute ID mismatch");
        return attr;
    }
    const Attribute & operator[](std::size_t id) const override { return const_cast<ConcreteTable*>(this)->operator[](id); }
 
    Attribute & at(const ThreadSafePooledString &name) override {
        if (auto it = name_to_attr_.find(name); it != name_to_attr_.end()) {
            M_insist(it->second < attrs_.size());
            return operator[](it->second);
        }
        throw std::out_of_range("name does not exists");
    }
    const Attribute & at(const ThreadSafePooledString &name) const override { return const_cast<ConcreteTable*>(this)->at(name); }
    Attribute & operator[](const ThreadSafePooledString &name) override { return operator[](name_to_attr_.find(name)->second); }
    const Attribute & operator[](const ThreadSafePooledString &name) const override { return const_cast<ConcreteTable*>(this)->operator[](name); }
 
    bool has_attribute(const ThreadSafePooledString &name) const override { return name_to_attr_.contains(name); }
 
    bool operator== (const Table &other) override {
        if (is<const ConcreteTable>(other))
            return this == &other; // check for referential equality.
        if (is<const TableDecorator>(other))
            return other.operator==(*this);
        M_unreachable("unknown table type");
    }
    bool operator== (const Table &other) const override { return const_cast<ConcreteTable*>(this)->operator==(other); }
 
    const ThreadSafePooledString & name() const override { return name_; }
 
    Store & store() const override { return *store_; }
    void store(std::unique_ptr<Store> new_store) override { using std::swap; swap(store_, new_store); }
 
    const storage::DataLayout & layout() const override { M_insist(bool(layout_)); return layout_; }
    void layout(storage::DataLayout &&new_layout) override { layout_ = std::move(new_layout); }
    virtual void layout(const storage::DataLayoutFactory &factory) override;
 
    std::vector<std::reference_wrapper<const Attribute>> primary_key() const override {
        std::vector<std::reference_wrapper<const Attribute>> res;
        for (auto id : primary_key_)
            res.emplace_back(operator[](id));
        return res;
    }
    void add_primary_key(const ThreadSafePooledString &name) override {
        auto &attr = at(name);
        primary_key_(attr.id) = true;
    }
 
    void push_back(ThreadSafePooledString name, const PrimitiveType *type) override {
        auto res = name_to_attr_.emplace(name, attrs_.size());
        if (not res.second)
            throw std::invalid_argument("attribute name already in use");
        attrs_.emplace_back(Attribute(attrs_.size(), *this, type, std::move(name)));
    }
 
    Schema schema(const ThreadSafePooledOptionalString &alias = {}) const override;
 
    size_t convert_id(size_t id) override {
        for (size_t i = 0; i <= id; ++i)
            if (attrs_[i].is_hidden) ++id;
        return id;
    }
 
    virtual void dump(std::ostream &out) const override;
    virtual void dump() const override;
};
 
struct TableDecorator : Table
{
    protected:
    std::unique_ptr<Table> table_;
 
    public:
    TableDecorator(std::unique_ptr<Table> table) : table_(std::move(table)) { }
    virtual ~TableDecorator() = default;
 
    virtual size_t num_attrs() const override { return table_->num_attrs(); }
    virtual size_t num_hidden_attrs() const override { return table_->num_hidden_attrs(); }
    virtual size_t num_all_attrs() const override { return table_->num_all_attrs(); }
 
    virtual iterator begin()  const override { return table_->begin(); }
    virtual iterator end()    const override { return table_->end(); }
    virtual iterator cbegin() const override { return table_->cbegin(); }
    virtual iterator cend()   const override { return table_->cend(); }
 
    virtual hidden_iterator begin_hidden()  const override { return table_->begin_hidden(); }
    virtual hidden_iterator end_hidden()    const override { return table_->end_hidden(); }
    virtual hidden_iterator cbegin_hidden() const override { return table_->cbegin_hidden(); }
    virtual hidden_iterator cend_hidden()   const override { return table_->cend_hidden(); }
 
    virtual all_iterator begin_all()  const override { return table_->begin_all(); }
    virtual all_iterator end_all()    const override { return table_->end_all(); }
    virtual all_iterator cbegin_all() const override { return table_->cbegin_all(); }
    virtual all_iterator cend_all()   const override { return table_->cend_all(); }
 
    virtual Attribute & at(std::size_t id) override { return table_->at(id); }
    virtual const Attribute & at(std::size_t id) const override { return table_->at(id); }
 
    virtual Attribute & operator[](std::size_t id) override { return table_->operator[](id); }
    virtual const Attribute & operator[](std::size_t id) const override { return table_->operator[](id); }
 
    virtual Attribute & at(const ThreadSafePooledString &name) override { return table_->at(name); }
    virtual const Attribute & at(const ThreadSafePooledString &name) const override { return table_->at(name); }
 
    virtual Attribute & operator[](const ThreadSafePooledString &name) override { return table_->at(name); }
    virtual const Attribute & operator[](const ThreadSafePooledString &name) const override { return table_->at(name); }
 
    virtual bool has_attribute(const ThreadSafePooledString &name) const override { return table_->has_attribute(name); }
 
    virtual bool operator== (const Table &other) override { return other == *table_; }
    virtual bool operator== (const Table &other) const override { return const_cast<TableDecorator*>(this)->operator==(other); }
 
    virtual const ThreadSafePooledString & name() const override { return table_->name(); }
 
    virtual Store & store() const override { return table_->store(); }
    virtual void store(std::unique_ptr<Store> new_store) override { table_->store(std::move(new_store)); }
 
    virtual const storage::DataLayout & layout() const override { return table_->layout(); }
    virtual void layout(storage::DataLayout &&new_layout) override { table_->layout(std::move(new_layout)); }
    virtual void layout(const storage::DataLayoutFactory &factory) override { table_->layout(factory); }
 
    virtual std::vector<std::reference_wrapper<const Attribute>> primary_key() const override { return table_->primary_key(); }
    virtual void add_primary_key(const ThreadSafePooledString &name) override { table_->add_primary_key(name); }
 
    virtual void push_back(ThreadSafePooledString name, const PrimitiveType * type) override { table_->push_back(std::move(name), type); }
 
    virtual Schema schema(const ThreadSafePooledOptionalString &alias) const override { return table_->schema(alias); }
 
    virtual size_t convert_id(size_t id) override { return table_->convert_id(id); }
 
    virtual void dump(std::ostream & out) const override { table_->dump(out); }
    virtual void dump() const override { table_->dump(); }
};
 
struct M_EXPORT MultiVersioningTable : TableDecorator
{
    public:
    MultiVersioningTable(std::unique_ptr<Table> table);
 
    ~MultiVersioningTable() { }
 
    void dump(std::ostream &out) const override;
    void dump() const override;
};
 
struct M_EXPORT Function
{
#define kind_t(X) \
    X(FN_Scalar) \
    X(FN_Aggregate)
 
    enum fnid_t {
#define M_FUNCTION(NAME, KIND) FN_ ## NAME,
#include <mutable/tables/Functions.tbl>
#undef M_FUNCTION
        FN_UDF, // for all user-defined functions
    };
 
    ThreadSafePooledString name; 
    fnid_t fnid; 
    M_DECLARE_ENUM(kind_t) kind; 
 
    Function(ThreadSafePooledString name, fnid_t fnid, kind_t kind) : name(std::move(name)), fnid(fnid), kind(kind) { }
 
    bool is_UDF() const { return fnid == FN_UDF; }
 
    bool is_scalar() const { return kind == FN_Scalar; }
    bool is_aggregate() const { return kind == FN_Aggregate; }
 
    void dump(std::ostream &out) const;
    void dump() const;
 
    private:
    static constexpr const char *FNID_TO_STR_[] = {
#define M_FUNCTION(NAME, KIND) "FN_" #NAME,
#include <mutable/tables/Functions.tbl>
#undef M_FUNCTION
            "FN_UDF",
    };
    static constexpr const char *KIND_TO_STR_[] = { M_ENUM_TO_STR(kind_t) };
#undef kind_t
};
 
struct M_EXPORT Database
{
    friend struct Catalog;
 
    struct index_entry_type
    {
        ThreadSafePooledString name; 
        const Table &table; 
        const Attribute &attribute; 
        std::unique_ptr<idx::IndexBase> index; 
        bool is_valid; 
 
        index_entry_type(ThreadSafePooledString name, const Table &table, const Attribute &attribute,
                         std::unique_ptr<idx::IndexBase> index)
            : name(std::move(name))
            , table(table)
            , attribute(attribute)
            , index(std::move(index))
            , is_valid(true)
        { }
    };
 
    public:
    ThreadSafePooledString name; 
    private:
    std::unordered_map<ThreadSafePooledString, std::unique_ptr<Table>> tables_; 
    std::unordered_map<ThreadSafePooledString, Function*> functions_; 
    std::unique_ptr<CardinalityEstimator> cardinality_estimator_; 
    std::list<index_entry_type> indexes_; 
 
    private:
    Database(ThreadSafePooledString name);
 
    public:
    ~Database();
 
    std::size_t size() const { return tables_.size(); }
    auto begin_tables() const { return tables_.cbegin(); }
    auto end_tables() const { return tables_.cend(); }
 
    /*===== Tables ===================================================================================================*/
    Table & get_table(const ThreadSafePooledString &name) const { return *tables_.at(name); }
    Table & add_table(ThreadSafePooledString name);
    Table & add(std::unique_ptr<Table> table) {
        auto it = tables_.find(table->name());
        if (it != tables_.end()) throw std::invalid_argument("table with that name already exists");
        it = tables_.emplace_hint(it, table->name(), std::move(table));
        return *it->second;
    }
    bool has_table(const ThreadSafePooledString &name) const { return tables_.contains(name); }
    void drop_table(const ThreadSafePooledString &name) {
        auto it = tables_.find(name);
        if (it == tables_.end())
            throw std::invalid_argument("Table of that name does not exist.");
        drop_indexes(name);
        tables_.erase(it);
    };
 
    /*===== Functions ================================================================================================*/
    const Function * get_function(const ThreadSafePooledString &name) const;
 
    /*===== Statistics ===============================================================================================*/
    std::unique_ptr<CardinalityEstimator> cardinality_estimator(std::unique_ptr<CardinalityEstimator> CE) {
        auto old = std::move(cardinality_estimator_); cardinality_estimator_ = std::move(CE); return old;
    }
    const CardinalityEstimator & cardinality_estimator() const { return *cardinality_estimator_; }
 
    /*===== Indexes ==================================================================================================*/
    void add_index(std::unique_ptr<idx::IndexBase> index, const ThreadSafePooledString &table_name,
                   const ThreadSafePooledString &attribute_name, ThreadSafePooledString index_name)
    {
        if (has_index(index_name))
            throw invalid_argument("Index with that name already exists.");
        auto &table = get_table(table_name);
        auto &attribute = table.at(attribute_name);
        indexes_.emplace_back(std::move(index_name), table, attribute, std::move(index));
    }
    void drop_index(const ThreadSafePooledString &index_name) {
        for (auto it = indexes_.cbegin(); it != indexes_.cend(); ++it) {
            if (it->name == index_name) {
                indexes_.erase(it);
                return;
            }
        }
        throw invalid_argument("Index of that name does not exist.");
    }
    void drop_indexes(const ThreadSafePooledString &table_name) {
        if (not has_table(table_name))
            throw invalid_argument("Table with that name does not exist.");
        for (auto it = indexes_.cbegin(); it != indexes_.end();) {
            if (it->table.name() == table_name)
                it = indexes_.erase(it);
            else
                ++it;
        }
    }
    bool has_index(const ThreadSafePooledString &index_name) const {
        for (auto it = indexes_.cbegin(); it != indexes_.cend(); ++it)
            if (it->name == index_name) return true;
        return false;
    }
    bool has_index(const ThreadSafePooledString &table_name, const ThreadSafePooledString &attribute_name,
                   idx::IndexMethod method) const {
        auto it = tables_.find(table_name);
        if (it == tables_.end())
            throw m::invalid_argument("Table with that name does not exist.");
        auto &table = it->second;
        if (not table->has_attribute(attribute_name))
            throw m::invalid_argument("Attribute with that name does not exist.");
        for (auto &entry : indexes_) {
            if (entry.is_valid and entry.table.name() == table_name and entry.attribute.name == attribute_name and
                entry.index->method() == method)
            {
                return true;
            }
        }
        return false;
    }
    const idx::IndexBase & get_index(const ThreadSafePooledString &index_name) const {
        for (auto &entry : indexes_) {
            if (entry.name == index_name)
                return *entry.index;
        }
        throw m::invalid_argument("Index of that name does not exist.");
    }
    const idx::IndexBase & get_index(const ThreadSafePooledString &table_name,
                                     const ThreadSafePooledString &attribute_name, idx::IndexMethod method) const {
        auto it = tables_.find(table_name);
        if (it == tables_.end())
            throw m::invalid_argument("Table with that name does not exist.");
        auto &table = it->second;
        if (not table->has_attribute(attribute_name))
            throw m::invalid_argument("Attribute with that name does not exist.");
        for (auto &entry : indexes_) {
            if (entry.is_valid and entry.table.name() == table_name and entry.attribute.name == attribute_name and
                entry.index->method() == method)
            {
                return *entry.index;
            }
        }
        throw m::invalid_argument("Index of that method on that attribute of that table does not exist.");
    }
    void invalidate_indexes(const ThreadSafePooledString &table_name) {
        if (not has_table(table_name))
            throw m::invalid_argument("Table with that name does not exist.");
        for (auto &entry : indexes_) {
            if (entry.table.name() == table_name)
                entry.is_valid = false;
        }
    }
};
 
}
 
namespace std {
 
template<>
struct hash<m::Schema::Identifier>
{
    uint64_t operator()(const m::Schema::Identifier &id) const {
        using std::hash;
        uint64_t h = hash<m::ThreadSafePooledString>{}(id.name);
        if (id.prefix.has_value())
            h *= hash<m::ThreadSafePooledOptionalString>{}(id.prefix);
        return h;
    }
};
 
template<>
struct hash<m::Attribute>
{
    uint64_t operator()(const m::Attribute &attr) const {
        using std::hash;
        auto h = hash<m::ThreadSafePooledString>{}(attr.table.name());
        return h * (attr.id + 1);
    }
};
 
}