Catalog.hpp

Go to the documentation of this file.

#pragma once
 
#include <concepts>
#include <memory>
#include <mutable/backend/Backend.hpp>
#include <mutable/backend/WebAssembly.hpp>
#include <mutable/catalog/CardinalityEstimator.hpp>
#include <mutable/catalog/CostFunction.hpp>
#include <mutable/catalog/DatabaseCommand.hpp>
#include <mutable/catalog/TableFactory.hpp>
#include <mutable/catalog/Scheduler.hpp>
#include <mutable/catalog/Schema.hpp>
#include <mutable/IR/PlanEnumerator.hpp>
#include <mutable/storage/DataLayoutFactory.hpp>
#include <mutable/util/ArgParser.hpp>
#include <mutable/util/macro.hpp>
#include <mutable/util/memory.hpp>
#include <mutable/util/Pool.hpp>
#include <mutable/util/Timer.hpp>
#include <type_traits>
 
 
namespace {
 
/*======================================================================================================================
 * Factory helpers
 *====================================================================================================================*/
 
struct StoreFactory
{
    virtual ~StoreFactory() { }
 
    virtual std::unique_ptr<m::Store> make(const m::Table&) const = 0;
};
 
template<typename T>
requires std::derived_from<T, m::Store>
struct ConcreteStoreFactory : StoreFactory
{
    std::unique_ptr<m::Store> make(const m::Table &tbl) const override { return std::make_unique<T>(tbl); }
};
 
struct CardinalityEstimatorFactory
{
    virtual ~CardinalityEstimatorFactory() { }
 
    virtual std::unique_ptr<m::CardinalityEstimator> make(m::ThreadSafePooledString database) const = 0;
};
 
template<typename T>
requires std::derived_from<T, m::CardinalityEstimator>
struct ConcreteCardinalityEstimatorFactory : CardinalityEstimatorFactory
{
    std::unique_ptr<m::CardinalityEstimator> make(m::ThreadSafePooledString database) const override {
        return std::make_unique<T>(std::move(database));
    }
};
 
struct WasmEngineFactory
{
    virtual ~WasmEngineFactory() { }
 
    virtual std::unique_ptr<m::WasmEngine> make() const = 0;
};
 
template<typename T>
requires std::derived_from<T, m::WasmEngine>
struct ConcreteWasmEngineFactory : WasmEngineFactory
{
    std::unique_ptr<m::WasmEngine> make() const override { return std::make_unique<T>(); }
};
 
struct BackendFactory
{
    virtual ~BackendFactory() { }
 
    virtual std::unique_ptr<m::Backend> make() const = 0;
};
 
template<typename T>
requires std::derived_from<T, m::Backend>
struct ConcreteBackendFactory : BackendFactory
{
    std::unique_ptr<m::Backend> make() const override { return std::make_unique<T>(); }
};
 
struct ConcreteWasmBackendFactory : BackendFactory
{
    private:
    std::unique_ptr<WasmEngineFactory> platform_factory_;
 
    public:
    ConcreteWasmBackendFactory(std::unique_ptr<WasmEngineFactory> platform_factory)
        : platform_factory_(std::move(platform_factory))
    { }
 
    std::unique_ptr<m::Backend> make() const override {
        return std::make_unique<m::WasmBackend>(platform_factory_->make());
    }
};
 
struct DatabaseInstructionFactory
{
    virtual ~DatabaseInstructionFactory() { }
 
    virtual std::unique_ptr<m::DatabaseInstruction> make(std::vector<std::string> args) const = 0;
};
 
template<typename T>
requires std::derived_from<T, m::DatabaseInstruction>
struct ConcreteDatabaseInstructionFactory : DatabaseInstructionFactory
{
    std::unique_ptr<m::DatabaseInstruction> make(std::vector<std::string> args) const override {
        return std::make_unique<T>(std::move(args));
    }
};
 
 
/*======================================================================================================================
 * Components
 *====================================================================================================================*/
 
template<typename T>
struct Component
{
    using type = T;
 
    private:
    const char *description_;
    std::unique_ptr<T> instance_;
 
    public:
    Component(const char *description, std::unique_ptr<T> instance)
        : description_(description), instance_(std::move(instance))
    { }
 
    Component(const char *description, T *instance)
        : description_(description), instance_(instance)
    { }
 
    const char * description() const { return description_; }
    T & operator*() const { return *instance_; }
};
 
template<typename T>
struct ComponentSet
{
    using type = T;
 
    private:
    using map_t = std::unordered_map<m::ThreadSafePooledString, Component<T>>;
    map_t components_;
    typename map_t::iterator default_;
 
    public:
    void add(m::ThreadSafePooledString name, Component<T> &&component) {
        auto it = components_.find(name);
        if (it != components_.end())
            throw std::invalid_argument("component with that name already exists");
        it = components_.emplace_hint(it, std::move(name), std::move(component));
        if (default_ == components_.end())
            default_ = it;
    }
 
    void set_default(const m::ThreadSafePooledString &name) {
        auto it = components_.find(name);
        if (it == components_.end())
            throw std::invalid_argument("component does not exist");
        default_ = it;
    }
 
    bool has_default() const { return default_ != components_.end(); }
 
    const m::ThreadSafePooledString & get_default_name() const { M_insist(has_default()); return default_->first; }
    const char * get_default_description() const { return default_->second.description; }
    T & get_default() const { M_insist(has_default()); return *default_->second; }
 
    const char * get_description(const m::ThreadSafePooledString &name) const {
        auto it = components_.find(name);
        if (it == components_.end())
            throw std::invalid_argument("component does not exist");
        return it->second.description;
    }
 
    T & get(const m::ThreadSafePooledString &name) const {
        auto it = components_.find(name);
        if (it == components_.end())
            throw std::invalid_argument("component does not exist");
        return *it->second;
    }
 
    typename map_t::iterator begin() { return components_.begin(); }
    typename map_t::iterator end() { return components_.end(); }
    typename map_t::const_iterator begin() const { return components_.begin(); }
    typename map_t::const_iterator end() const { return components_.end(); }
    typename map_t::const_iterator cbegin() const { return components_.begin(); }
    typename map_t::const_iterator cend() const { return components_.end(); }
};
 
}
 
namespace m {
 
/*======================================================================================================================
 * Catalog
 *====================================================================================================================*/
 
struct M_EXPORT Catalog
{
    private:
    static Catalog *the_catalog_;
 
    m::ArgParser arg_parser_;
 
    std::unique_ptr<memory::Allocator> allocator_; 
    mutable ThreadSafeStringPool pool_; 
    std::unordered_map<ThreadSafePooledString, Database*> databases_; 
    Database *database_in_use_ = nullptr; 
    std::unordered_map<ThreadSafePooledString, Function*> standard_functions_; 
    Timer timer_; 
 
    private:
    Catalog();
    Catalog(const Catalog&) = delete;
    Catalog & operator=(const Catalog&) = delete;
 
    public:
    ~Catalog();
 
    static Catalog & Get();
 
    static void Clear() {
        if (the_catalog_) {
            for (auto DB : the_catalog_->databases_)
                delete DB.second;
            the_catalog_->databases_.clear();
            the_catalog_->database_in_use_ = nullptr;
        }
    }
 
    static void Destroy();
 
    m::ArgParser & arg_parser() { return arg_parser_; }
 
    std::size_t num_databases() const { return databases_.size(); }
 
    ThreadSafeStringPool & get_pool() { return pool_; }
    const ThreadSafeStringPool & get_pool() const { return pool_; }
 
    Timer & timer() { return timer_; }
    const Timer & timer() const { return timer_; }
 
    memory::Allocator & allocator() { return *allocator_; }
    const memory::Allocator & allocator() const { return *allocator_; }
 
    ThreadSafePooledString pool(const char *str) const { return pool_(str); }
    ThreadSafePooledString pool(std::string_view str) const { return pool_(str); }
 
    /*===== Database =================================================================================================*/
    Database & add_database(ThreadSafePooledString name);
    Database & get_database(const ThreadSafePooledString &name) const { return *databases_.at(name); }
    bool has_database(const ThreadSafePooledString &name) const { return databases_.contains(name); }
    void drop_database(const ThreadSafePooledString &name);
    void drop_database(const Database &db) { return drop_database(db.name); }
 
    bool has_database_in_use() const { return database_in_use_ != nullptr; }
    Database & get_database_in_use() {
        if (not has_database_in_use())
            throw std::logic_error("no database currently in use");
        return *database_in_use_;
    }
    const Database & get_database_in_use() const { return const_cast<Catalog*>(this)->get_database_in_use(); }
    void set_database_in_use(Database &db) { database_in_use_ = &db; }
    void unset_database_in_use() { database_in_use_ = nullptr; }
 
    /*===== Functions ================================================================================================*/
    const Function * get_function(const ThreadSafePooledString &name) const { return standard_functions_.at(name); }
 
 
    /*------------------------------------------------------------------------------------------------------------------
     * Components
     *----------------------------------------------------------------------------------------------------------------*/
    private:
    ComponentSet<StoreFactory> stores_;
    ComponentSet<storage::DataLayoutFactory> data_layouts_;
    ComponentSet<CardinalityEstimatorFactory> cardinality_estimators_;
    ComponentSet<pe::PlanEnumerator> plan_enumerators_;
    ComponentSet<BackendFactory> backends_;
    ComponentSet<CostFunction> cost_functions_;
    ComponentSet<DatabaseInstructionFactory> instructions_;
    ComponentSet<Scheduler> schedulers_;
 
    using TableFactoryDecoratorCallback = std::function<std::unique_ptr<TableFactory>(std::unique_ptr<TableFactory>)>;
    ComponentSet<TableFactoryDecoratorCallback> table_properties_; // stores callback functions that decorate a table with the given decorator
 
    using PreOptimizationCallback = std::function<void(QueryGraph&)>;
    ComponentSet<PreOptimizationCallback> pre_optimizations_;
 
    using LogicalPostOptimizationCallback = std::function<std::unique_ptr<Producer>(std::unique_ptr<Producer>)>;
    ComponentSet<LogicalPostOptimizationCallback> logical_post_optimizations_;
 
    using PhysicalPostOptimizationCallback = std::function<std::unique_ptr<MatchBase>(std::unique_ptr<MatchBase>)>;
    ComponentSet<PhysicalPostOptimizationCallback> physical_post_optimizations_;
 
    public:
    /*===== Stores ===================================================================================================*/
    template<typename T>
    requires std::derived_from<T, m::Store>
    void register_store(ThreadSafePooledString name, const char *description = nullptr) {
        auto c = Component<StoreFactory>(description, std::make_unique<ConcreteStoreFactory<T>>());
        stores_.add(std::move(name), std::move(c));
    }
    void default_store(const ThreadSafePooledString &name) { stores_.set_default(name); }
    bool has_default_store() const { return stores_.has_default(); }
    std::unique_ptr<Store> create_store(const Table &tbl) const { return stores_.get_default().make(tbl); }
    std::unique_ptr<Store> create_store(const ThreadSafePooledString &name, const Table &tbl) const {
        return stores_.get(name).make(tbl);
    }
    const ThreadSafePooledString & default_store_name() const { return stores_.get_default_name(); }
 
    auto stores_begin()        { return stores_.begin(); }
    auto stores_end()          { return stores_.end(); }
    auto stores_begin()  const { return stores_.begin(); }
    auto stores_end()    const { return stores_.end(); }
    auto stores_cbegin() const { return stores_.begin(); }
    auto stores_cend()   const { return stores_.end(); }
 
    /*===== DataLayouts ==============================================================================================*/
    void register_data_layout(ThreadSafePooledString name, std::unique_ptr<storage::DataLayoutFactory> data_layout,
                              const char *description = nullptr)
    {
        data_layouts_.add(std::move(name), Component<storage::DataLayoutFactory>(description, std::move(data_layout)));
    }
    void default_data_layout(const ThreadSafePooledString &name) { data_layouts_.set_default(name); }
    bool has_default_data_layout() const { return data_layouts_.has_default(); }
    storage::DataLayoutFactory & data_layout() const { return data_layouts_.get_default(); }
    storage::DataLayoutFactory & data_layout(const ThreadSafePooledString &name) const { return data_layouts_.get(name); }
    const m::ThreadSafePooledString & default_data_layout_name() const { return data_layouts_.get_default_name(); }
 
    auto data_layouts_begin()        { return data_layouts_.begin(); }
    auto data_layouts_end()          { return data_layouts_.end(); }
    auto data_layouts_begin()  const { return data_layouts_.begin(); }
    auto data_layouts_end()    const { return data_layouts_.end(); }
    auto data_layouts_cbegin() const { return data_layouts_.begin(); }
    auto data_layouts_cend()   const { return data_layouts_.end(); }
 
    /*===== CardinalityEstimators ====================================================================================*/
    template<typename T>
    requires std::derived_from<T, m::CardinalityEstimator>
    void register_cardinality_estimator(ThreadSafePooledString name, const char *description = nullptr) {
        auto c = Component<CardinalityEstimatorFactory>(
            description,
            std::make_unique<ConcreteCardinalityEstimatorFactory<T>>()
        );
        cardinality_estimators_.add(std::move(name), std::move(c));
    }
 
    void default_cardinality_estimator(const ThreadSafePooledString &name) { cardinality_estimators_.set_default(name); }
    bool has_default_cardinality_estimator() const { return cardinality_estimators_.has_default(); }
    std::unique_ptr<CardinalityEstimator> create_cardinality_estimator(ThreadSafePooledString database) const {
        return cardinality_estimators_.get_default().make(std::move(database));
    }
    std::unique_ptr<CardinalityEstimator> create_cardinality_estimator(const ThreadSafePooledString &name,
                                                                       ThreadSafePooledString database) const {
        return cardinality_estimators_.get(name).make(std::move(database));
    }
    const ThreadSafePooledString & default_cardinality_estimator_name() const {
        return cardinality_estimators_.get_default_name();
    }
 
    auto cardinality_estimators_begin()        { return cardinality_estimators_.begin(); }
    auto cardinality_estimators_end()          { return cardinality_estimators_.end(); }
    auto cardinality_estimators_begin()  const { return cardinality_estimators_.begin(); }
    auto cardinality_estimators_end()    const { return cardinality_estimators_.end(); }
    auto cardinality_estimators_cbegin() const { return cardinality_estimators_.begin(); }
    auto cardinality_estimators_cend()   const { return cardinality_estimators_.end(); }
 
    /*===== Plan Enumerators =========================================================================================*/
    void register_plan_enumerator(ThreadSafePooledString name, std::unique_ptr<pe::PlanEnumerator> PE,
                                  const char *description = nullptr)
    {
        plan_enumerators_.add(std::move(name), Component<pe::PlanEnumerator>(description, std::move(PE)));
    }
    void default_plan_enumerator(const ThreadSafePooledString &name) { plan_enumerators_.set_default(name); }
    bool has_default_plan_enumerator() const { return plan_enumerators_.has_default(); }
    pe::PlanEnumerator & plan_enumerator() const { return plan_enumerators_.get_default(); }
    pe::PlanEnumerator & plan_enumerator(const ThreadSafePooledString &name) const { return plan_enumerators_.get(name); }
    const ThreadSafePooledString & default_plan_enumerator_name() const { return plan_enumerators_.get_default_name(); }
 
    auto plan_enumerators_begin() { return plan_enumerators_.begin(); }
    auto plan_enumerators_end() { return plan_enumerators_.end(); }
    auto plan_enumerators_begin() const { return plan_enumerators_.begin(); }
    auto plan_enumerators_end() const { return plan_enumerators_.end(); }
    auto plan_enumerators_cbegin() const { return plan_enumerators_.begin(); }
    auto plan_enumerators_cend() const { return plan_enumerators_.end(); }
 
    /*===== Backends =================================================================================================*/
    template<typename T>
    requires std::derived_from<T, m::Backend>
    void register_backend(ThreadSafePooledString name, const char *description = nullptr) {
        auto c = Component<BackendFactory>(description, std::make_unique<ConcreteBackendFactory<T>>());
        backends_.add(std::move(name), std::move(c));
    }
    template<typename T>
    requires std::derived_from<T, m::WasmEngine>
    void register_wasm_backend(ThreadSafePooledString name, const char *description = nullptr) {
        auto c = Component<BackendFactory>(
            description,
            std::make_unique<ConcreteWasmBackendFactory>(std::make_unique<ConcreteWasmEngineFactory<T>>())
        );
        backends_.add(std::move(name), std::move(c));
    }
    void default_backend(const ThreadSafePooledString &name) { backends_.set_default(name); }
    bool has_default_backend() const { return backends_.has_default(); }
    std::unique_ptr<Backend> create_backend() const { return backends_.get_default().make(); }
    std::unique_ptr<Backend> create_backend(const ThreadSafePooledString &name) const { return backends_.get(name).make(); }
    const ThreadSafePooledString & default_backend_name() const { return backends_.get_default_name(); }
 
    auto backends_begin()        { return backends_.begin(); }
    auto backends_end()          { return backends_.end(); }
    auto backends_begin()  const { return backends_.begin(); }
    auto backends_end()    const { return backends_.end(); }
    auto backends_cbegin() const { return backends_begin(); }
    auto backends_cend()   const { return backends_end(); }
 
    /*===== CostFunction =============================================================================================*/
    void register_cost_function(ThreadSafePooledString name, std::unique_ptr<CostFunction> CF,
                                const char *description = nullptr)
    {
        cost_functions_.add(std::move(name), Component<CostFunction>(description, std::move(CF)));
    }
    void default_cost_function(const ThreadSafePooledString &name) { cost_functions_.set_default(name); }
    bool has_default_cost_function() const { return cost_functions_.has_default(); }
    CostFunction & cost_function() const { return cost_functions_.get_default(); }
    CostFunction & cost_function(const ThreadSafePooledString &name) const { return cost_functions_.get(name); }
    const ThreadSafePooledString & default_cost_function_name() const { return cost_functions_.get_default_name(); }
 
    auto cost_functions_begin()        { return cost_functions_.begin(); }
    auto cost_functions_end()          { return cost_functions_.end(); }
    auto cost_functions_begin()  const { return cost_functions_.begin(); }
    auto cost_functions_end()    const { return cost_functions_.end(); }
    auto cost_functions_cbegin() const { return cost_functions_begin(); }
    auto cost_functions_cend()   const { return cost_functions_end(); }
 
    /*===== Instructions =============================================================================================*/
    template<typename T>
    requires std::derived_from<T, m::DatabaseInstruction>
    void register_instruction(ThreadSafePooledString name, const char *description = nullptr) {
        auto I = Component<DatabaseInstructionFactory>(
            description,
            std::make_unique<ConcreteDatabaseInstructionFactory<T>>()
        );
        instructions_.add(std::move(name), std::move(I));
    }
    std::unique_ptr<DatabaseInstruction> create_instruction(const ThreadSafePooledString &name,
                                                            const std::vector<std::string> &args) const
    {
        return instructions_.get(name).make(args);
    }
 
    auto instructions_begin() { return instructions_.begin(); }
    auto instructions_end() { return instructions_.end(); }
    auto instructions_begin() const { return instructions_.begin(); }
    auto instructions_end() const { return instructions_.end(); }
    auto instructions_cbegin() const { return instructions_.begin(); }
    auto instructions_cend() const { return instructions_.end(); }
 
    /*===== Schedulers =================================================================================================*/
    void register_scheduler(ThreadSafePooledString name, std::unique_ptr<Scheduler> scheduler,
                            const char *description = nullptr)
    {
        schedulers_.add(std::move(name), Component<Scheduler>(description, std::move(scheduler)));
    }
    void default_scheduler(const ThreadSafePooledString &name) { schedulers_.set_default(name); }
    bool has_default_scheduler() const { return schedulers_.has_default(); }
    Scheduler & scheduler() const { return schedulers_.get_default(); }
    Scheduler & scheduler(const ThreadSafePooledString &name) const { return schedulers_.get(name); }
    const ThreadSafePooledString & default_scheduler_name() const { return schedulers_.get_default_name(); }
 
    auto schedulers_begin()        { return schedulers_.begin(); }
    auto schedulers_end()          { return schedulers_.end(); }
    auto schedulers_begin()  const { return schedulers_.begin(); }
    auto schedulers_end()    const { return schedulers_.end(); }
    auto schedulers_cbegin() const { return schedulers_begin(); }
    auto schedulers_cend()   const { return schedulers_end(); }
 
    /*===== Table Factories ==========================================================================================*/
    private:
    std::unique_ptr<TableFactory> table_factory_; 
 
    public:
    std::unique_ptr<TableFactory> table_factory(std::unique_ptr<TableFactory> table_factory) {
        return std::exchange(table_factory_, std::move(table_factory));
    }
    TableFactory & table_factory() const { return *table_factory_; }
 
    template<class T>
    requires std::derived_from<T, TableFactoryDecorator>
    void register_table_property(ThreadSafePooledString name, const char *description = nullptr) {
        table_properties_.add(
            std::move(name),
            Component<TableFactoryDecoratorCallback>(
                description,
                std::make_unique<TableFactoryDecoratorCallback>([](std::unique_ptr<TableFactory> table_factory){
                    return std::make_unique<T>(std::move(table_factory));
                })
            )
        );
    }
    std::unique_ptr<TableFactory> apply_table_property(const ThreadSafePooledString &name,
                                                       std::unique_ptr<TableFactory> table_factory) const
    {
        return table_properties_.get(name).operator()(std::move(table_factory));
    }
 
    auto table_properties_begin()        { return table_properties_.begin(); }
    auto table_properties_end()          { return table_properties_.end(); }
    auto table_properties_begin()  const { return table_properties_.begin(); }
    auto table_properties_end()    const { return table_properties_.end(); }
    auto table_properties_cbegin() const { return table_properties_begin(); }
    auto table_properties_cend()   const { return table_properties_end(); }
 
    /*===== Pre-Optimizations ========================================================================================*/
    void register_pre_optimization(ThreadSafePooledString name, PreOptimizationCallback optimization,
                                   const char *description = nullptr)
    {
        pre_optimizations_.add(
            std::move(name),
            Component<PreOptimizationCallback>(
                description,
                std::make_unique<PreOptimizationCallback>(std::move(optimization))
            )
        );
    }
 
    auto pre_optimizations()              { return range(pre_optimizations_.begin(), pre_optimizations_.end()); }
    auto pre_optimizations_begin()        { return pre_optimizations_.begin(); }
    auto pre_optimizations_end()          { return pre_optimizations_.end(); }
    auto pre_optimizations_begin()  const { return pre_optimizations_.begin(); }
    auto pre_optimizations_end()    const { return pre_optimizations_.end(); }
    auto pre_optimizations_cbegin() const { return pre_optimizations_begin(); }
    auto pre_optimizations_cend()   const { return pre_optimizations_end(); }
 
    /*===== Logical Post-Optimizations ===============================================================================*/
    void register_logical_post_optimization(ThreadSafePooledString name, LogicalPostOptimizationCallback optimization,
                                            const char *description = nullptr)
    {
        logical_post_optimizations_.add(
            std::move(name),
            Component<LogicalPostOptimizationCallback>(
                description,
                std::make_unique<LogicalPostOptimizationCallback>(std::move(optimization))
            )
        );
    }
 
    auto logical_post_optimizations() {
        return range(logical_post_optimizations_.begin(), logical_post_optimizations_.end());
    }
    auto logical_post_optimizations_begin()        { return logical_post_optimizations_.begin(); }
    auto logical_post_optimizations_end()          { return logical_post_optimizations_.end(); }
    auto logical_post_optimizations_begin()  const { return logical_post_optimizations_.begin(); }
    auto logical_post_optimizations_end()    const { return logical_post_optimizations_.end(); }
    auto logical_post_optimizations_cbegin() const { return logical_post_optimizations_begin(); }
    auto logical_post_optimizations_cend()   const { return logical_post_optimizations_end(); }
 
    /*===== Physical Post-Optimizations ===============================================================================*/
    void register_physical_post_optimization(const char *name, PhysicalPostOptimizationCallback optimization,
                                             const char *description = nullptr)
    {
        physical_post_optimizations_.add(
            pool(name),
            Component<PhysicalPostOptimizationCallback>(
                description,
                std::make_unique<PhysicalPostOptimizationCallback>(std::move(optimization))
            )
        );
    }
 
    auto physical_post_optimizations() {
        return range(physical_post_optimizations_.begin(), physical_post_optimizations_.end());
    }
    auto physical_post_optimizations_begin()        { return physical_post_optimizations_.begin(); }
    auto physical_post_optimizations_end()          { return physical_post_optimizations_.end(); }
    auto physical_post_optimizations_begin()  const { return physical_post_optimizations_.begin(); }
    auto physical_post_optimizations_end()    const { return physical_post_optimizations_.end(); }
    auto physical_post_optimizations_cbegin() const { return physical_post_optimizations_begin(); }
    auto physical_post_optimizations_cend()   const { return physical_post_optimizations_end(); }
};
 
}