WasmAlgo.hpp

Go to the documentation of this file.

#pragma once
 
#include "backend/WasmUtil.hpp"
#include <mutable/parse/AST.hpp>
#include <optional>
 
// must be included after Binaryen due to conflicts, e.g. with `::wasm::Throw`
#include "backend/WasmMacro.hpp"
 
 
namespace m {
 
namespace wasm {
 
// forward declarations
template<bool IsGlobal> struct ChainedHashTable;
template<bool IsGlobal, bool ValueInPlace> struct OpenAddressingHashTable;
struct ProbingStrategy;
 
/*======================================================================================================================
 * sorting
 *====================================================================================================================*/
 
template<bool CmpPredicated, bool IsGlobal>
void quicksort(Buffer<IsGlobal> &buffer, const std::vector<SortingOperator::order_type> &order);
 
 
/*======================================================================================================================
 * hashing
 *====================================================================================================================*/
 
/*----- bit mix functions --------------------------------------------------------------------------------------------*/
 
U64x1 murmur3_bit_mix(U64x1 bits);
 
 
/*----- hash functions -----------------------------------------------------------------------------------------------*/
 
U64x1 fnv_1a(Ptr<U8x1> bytes, U32x1 num_bytes);
U64x1 str_hash(NChar str);
U64x1 murmur3_64a_hash(std::vector<std::pair<const Type*, SQL_t>> values);
 
 
/*----- hash tables --------------------------------------------------------------------------------------------------*/
 
struct HashTable
{
    using index_t = std::size_t;
    using offset_t = int32_t;
    using size_t = uint32_t;
 
    // forward declaration
    template<bool IsConst> struct the_entry;
 
    template<sql_type T, bool IsConst>
    struct the_reference;
 
    template<sql_type T, bool IsConst>
    requires (not std::same_as<T, NChar>) and (T::num_simd_lanes == 1)
    struct the_reference<T, IsConst>
    {
        friend struct the_entry<false>;
        friend struct the_entry<true>;
 
        using value_t = PrimitiveExpr<typename T::type>;
 
        private:
        Ptr<value_t> value_;
        std::optional<Ptr<U8x1>> is_null_byte_;
        std::optional<U8x1> is_null_mask_;
 
        explicit the_reference(Ptr<value_t> value, std::optional<Ptr<U8x1>> is_null_byte, std::optional<U8x1> is_null_mask)
            : value_(value)
            , is_null_byte_(std::move(is_null_byte))
            , is_null_mask_(std::move(is_null_mask))
        {
            M_insist(bool(is_null_byte_) == bool(is_null_mask_),
                     "either both or none of NULL byte and NULL mask must be specified");
        }
        explicit the_reference(Ptr<value_t> value, Ptr<U8x1> is_null_byte, U8x1 is_null_mask)
            : value_(value)
            , is_null_byte_(is_null_byte)
            , is_null_mask_(is_null_mask)
        { }
 
        public:
        explicit the_reference(Ptr<value_t> value) : value_(value) { }
        explicit the_reference(Ptr<value_t> value, Ptr<void> null_bitmap, uint8_t null_bit_offset)
            : value_(value)
            , is_null_byte_((null_bitmap + (null_bit_offset / 8)).to<uint8_t*>())
            , is_null_mask_(1U << (null_bit_offset % 8))
        { }
 
        void discard() {
            value_.discard();
            if (is_null_byte_) {
                is_null_byte_->discard();
                is_null_mask_->discard();
            }
        }
        the_reference clone() const {
            if (is_null_byte_)
                return the_reference(value_.clone(), is_null_byte_->clone(), is_null_mask_->clone());
            else
                return the_reference(value_.clone());
        }
 
        void operator=(T _value) requires (not IsConst) {
            M_insist(bool(is_null_byte_) == _value.can_be_null(), "value of non-nullable entry must not be nullable");
            if (is_null_byte_) {
                auto [value, is_null] = _value.split();
                *value_ = value;
                setbit(*is_null_byte_, is_null, *is_null_mask_);
            } else {
                *value_ = _value.insist_not_null();
            }
        }
        void set_value(value_t value) requires (not IsConst) {
            *value_ = value;
            if (is_null_byte_) {
                is_null_byte_->discard();
                is_null_mask_->discard();
            }
        }
 
        void set_null_bit(Boolx1 is_null) requires (not IsConst) {
            value_.discard();
            M_insist(bool(is_null_byte_));
            M_insist(bool(is_null_mask_));
            setbit(*is_null_byte_, is_null, *is_null_mask_);
        }
        void set_null() requires (not IsConst) {
            value_.discard();
            M_insist(bool(is_null_byte_));
            M_insist(bool(is_null_mask_));
            **is_null_byte_ = **is_null_byte_ bitor *is_null_mask_;
        }
        void set_not_null() requires (not IsConst) {
            value_.discard();
            M_insist(bool(is_null_byte_));
            M_insist(bool(is_null_mask_));
            **is_null_byte_ = **is_null_byte_ bitand ~*is_null_mask_;
        }
 
        Boolx1 operator==(T _value) {
            auto [value, is_null] = _value.split();
            if (is_null_byte_) {
                auto is_null_ = (**is_null_byte_ bitand *is_null_mask_).template to<bool>();
                auto equal_nulls = is_null_.clone() == is_null;
                return equal_nulls and (is_null_ or *value_ == value);
            } else {
                return not is_null and *value_ == value;
            }
        }
 
        operator T() {
            if (is_null_byte_)
                return T(*value_, (**is_null_byte_ bitand *is_null_mask_).template to<bool>());
            else
                return T(*value_);
        }
 
        friend the_reference Select(Boolx1 cond, the_reference tru, the_reference fals) {
            M_insist(bool(tru.is_null_byte_) == bool(fals.is_null_byte_), "null byte mismatch");
            M_insist(bool(tru.is_null_mask_) == bool(fals.is_null_mask_), "null mask mismatch");
            if (tru.is_null_byte_) {
                the_reference r(
                    /* value_=        */ Select(cond.clone(), tru.value_, fals.value_),
                    /* is_null_byte_= */ Select(cond.clone(), *tru.is_null_byte_, *fals.is_null_byte_),
                    /* is_null_mask_= */ Select(cond.clone(), *tru.is_null_mask_, *fals.is_null_mask_)
                );
                cond.discard();
                return r;
            } else {
                return the_reference(Select(cond, tru.value_, fals.value_));
            }
        }
    };
 
    template<bool IsConst>
    struct the_reference<NChar, IsConst>
    {
        friend struct the_entry<false>;
        friend struct the_entry<true>;
 
        private:
        NChar addr_;
        std::optional<Ptr<U8x1>> is_null_byte_;
        std::optional<U8x1> is_null_mask_;
 
        explicit the_reference(NChar addr, std::optional<Ptr<U8x1>> is_null_byte, std::optional<U8x1> is_null_mask)
            : addr_(addr)
            , is_null_byte_(std::move(is_null_byte))
            , is_null_mask_(std::move(is_null_mask))
        {
            M_insist(addr.can_be_null() == bool(is_null_byte_), "nullable entry must have a NULL bit");
            M_insist(bool(is_null_byte_) == bool(is_null_mask_),
                     "either both or none of NULL byte and NULL mask must be specified");
        }
        explicit the_reference(NChar addr, Ptr<U8x1> is_null_byte, U8x1 is_null_mask)
            : addr_(addr)
            , is_null_byte_(is_null_byte)
            , is_null_mask_(is_null_mask)
        {
            M_insist(addr.can_be_null(), "entry with NULL bit must be nullable");
        }
 
        public:
        explicit the_reference(NChar addr)
            : addr_(addr)
        {
            M_insist(not addr.can_be_null(), "entry without NULL bit must be nullable"); \
        }
        explicit the_reference(NChar addr, Ptr<void> null_bitmap, uint8_t null_bit_offset)
            : addr_(addr)
            , is_null_byte_((null_bitmap + (null_bit_offset / 8)).to<uint8_t*>())
            , is_null_mask_(1U << (null_bit_offset % 8))
        {
            M_insist(addr.can_be_null(), "entry with NULL bit must be nullable");
        }
 
        void discard() {
            addr_.discard();
            if (addr_.can_be_null()) {
                is_null_byte_->discard();
                is_null_mask_->discard();
            }
        }
        the_reference clone() const {
            if (addr_.can_be_null())
                return the_reference(addr_.clone(), is_null_byte_->clone(), is_null_mask_->clone());
            else
                return the_reference(addr_.clone());
        }
 
        void operator=(NChar addr) requires (not IsConst) {
            M_insist(addr_.length() == addr.length() and
                     addr_.guarantees_terminating_nul() == addr.guarantees_terminating_nul(),
                     "type mismatch");
            if (addr_.can_be_null()) {
                IF (not addr.clone().is_null()) {
                    strncpy(addr_, addr.clone(), U32x1(addr.size_in_bytes())).discard();
                };
                setbit(*is_null_byte_, addr.is_null(), *is_null_mask_);
            } else {
                Wasm_insist(addr.clone().not_null(), "value of non-nullable entry must not be nullable");
                strncpy(addr_, addr, U32x1(addr.size_in_bytes())).discard();
            }
        }
        void set_value(NChar addr) requires (not IsConst) {
            M_insist(addr_.length() == addr.length() and
                     addr_.guarantees_terminating_nul() == addr.guarantees_terminating_nul(),
                     "type mismatch");
            strncpy(addr_, addr, U32x1(addr.size_in_bytes())).discard();
            if (addr_.can_be_null()) {
                is_null_byte_->discard();
                is_null_mask_->discard();
            }
        }
        void set_null_bit(Boolx1 is_null) requires (not IsConst) {
            addr_.discard();
            M_insist(bool(is_null_byte_));
            M_insist(bool(is_null_mask_));
            setbit(*is_null_byte_, is_null, *is_null_mask_);
        }
        void set_null() requires (not IsConst) {
            addr_.discard();
            M_insist(bool(is_null_byte_));
            M_insist(bool(is_null_mask_));
            **is_null_byte_ = **is_null_byte_ bitor *is_null_mask_;
        }
        void set_not_null() requires (not IsConst) {
            addr_.discard();
            M_insist(bool(is_null_byte_));
            M_insist(bool(is_null_mask_));
            **is_null_byte_ = **is_null_byte_ bitand ~*is_null_mask_;
        }
 
        Boolx1 operator==(NChar _addr) {
            M_insist(addr_.length() == _addr.length() and
                     addr_.guarantees_terminating_nul() == _addr.guarantees_terminating_nul(),
                     "type mismatch");
            auto [addr, is_nullptr] = _addr.split();
            _Boolx1 _equal_addrs = strncmp(addr_, NChar(addr, _addr.can_be_null(), addr_.length(),
                                                      addr_.guarantees_terminating_nul()),
                                         U32x1(addr_.length()), EQ);
            auto [equal_addrs_val, equal_addrs_is_null] = _equal_addrs.split();
            equal_addrs_is_null.discard(); // use potentially-null value but it is overruled if it is invalid, i.e. NULL
            if (addr_.can_be_null()) {
                auto is_nullptr_ = (**is_null_byte_ bitand *is_null_mask_).template to<bool>();
                auto equal_nulls = is_nullptr_.clone() == is_nullptr;
                return equal_nulls and (is_nullptr_ or equal_addrs_val);
            } else {
                return not is_nullptr and equal_addrs_val;
            }
        }
 
        operator NChar() {
            if (addr_.can_be_null()) {
                Boolx1 is_null = (**is_null_byte_ bitand *is_null_mask_).template to<bool>();
                return NChar(Select(is_null, Ptr<Charx1>::Nullptr(), addr_.val()), /* can_be_null= */ true,
                             addr_.length(), addr_.guarantees_terminating_nul());
            } else {
                return addr_;
            }
        }
 
        friend the_reference Select(Boolx1 cond, the_reference tru, the_reference fals) {
            M_insist(tru.addr_.can_be_null() == fals.addr_.can_be_null(), "nullable mismatch");
            M_insist(tru.addr_.length() == fals.addr_.length() and
                     tru.addr_.guarantees_terminating_nul() == fals.addr_.guarantees_terminating_nul(),
                     "type mismatch");
            M_insist(bool(tru.is_null_byte_) == bool(fals.is_null_byte_), "null byte mismatch");
            M_insist(bool(tru.is_null_mask_) == bool(fals.is_null_mask_), "null mask mismatch");
            if (tru.addr_.can_be_null()) {
                the_reference r(
                    /* addr_=         */ NChar(Select(cond.clone(), tru.addr_, fals.addr_), /* can_be_null= */ true,
                                               tru.addr_.length(), tru.addr_.guarantees_terminating_nul()),
                    /* is_null_byte_= */ Select(cond.clone(), *tru.is_null_byte_, *fals.is_null_byte_),
                    /* is_null_mask_= */ Select(cond.clone(), *tru.is_null_mask_, *fals.is_null_mask_)
                );
                cond.discard();
                return r;
            } else {
                return the_reference(NChar(Select(cond, tru.addr_, fals.addr_), /* can_be_null= */ false,
                                           tru.addr_.length(), tru.addr_.guarantees_terminating_nul()));
            }
        }
    };
 
    template<sql_type T>
    using reference_t = the_reference<T, false>;
    template<sql_type T>
    using const_reference_t = the_reference<T, true>;
 
    template<bool IsConst>
    struct the_entry
    {
        friend struct HashTable;
 
        using value_t = std::variant<
            std::monostate
#define ADD_TYPE(TYPE) , the_reference<TYPE, IsConst>
            SQL_TYPES_SCALAR(ADD_TYPE)
#undef ADD_TYPE
        >;
 
        private:
        static void discard(value_t &ref) {
            std::visit(overloaded {
                [](auto &r) { r.discard(); },
                [](std::monostate) { M_unreachable("invalid variant"); },
            }, ref);
        }
 
        private:
        std::unordered_multimap<Schema::Identifier, value_t> refs_;
 
        public:
        the_entry() = default;
        the_entry(const the_entry&) = delete;
        the_entry(the_entry&&) = default;
 
        private:
        the_entry(std::unordered_multimap<Schema::Identifier, typename the_entry<not IsConst>::value_t> refs) {
            refs_.reserve(refs.size());
            for (auto &p : refs) {
                std::visit(overloaded {
                    [&]<typename T>(the_reference<T, not IsConst> &r) {
                        the_reference<T, IsConst> ref(
                            r.value_, std::move(r.is_null_byte_), std::move(r.is_null_mask_)
                        );
                        refs_.emplace(std::move(p.first), std::move(ref));
                    },
                    [&](the_reference<NChar, not IsConst> &r) {
                        the_reference<NChar, IsConst> ref(
                            r.addr_, std::move(r.is_null_byte_), std::move(r.is_null_mask_)
                        );
                        refs_.emplace(std::move(p.first), std::move(ref));
                    },
                    [](std::monostate) { M_unreachable("invalid variant"); },
                }, p.second);
            }
        }
 
        public:
        ~the_entry() {
            for (auto &p : refs_)
                discard(p.second);
        }
 
        operator the_entry<true>() requires (not IsConst) { return the_entry<true>(std::move(refs_)); }
 
        bool empty() const { return refs_.empty(); }
 
        bool has(const Schema::Identifier &id) const { return refs_.contains(id); }
 
        template<sql_type T>
        void add(Schema::Identifier id, the_reference<T, IsConst> &&ref) {
            refs_.emplace(std::move(id), std::forward<the_reference<T, IsConst>>(ref));
        }
 
        value_t extract(const Schema::Identifier &id) {
            auto it = refs_.find(id);
            M_insist(it != refs_.end(), "identifier not found");
            auto nh = refs_.extract(it);
            return std::move(nh.mapped());
        }
        template<sql_type T>
        the_reference<T, IsConst> extract(const Schema::Identifier &id) {
            using type = the_reference<T, IsConst>;
            M_insist(refs_.count(id) <= 1, "duplicate identifier, lookup ambiguous");
            auto it = refs_.find(id);
            M_insist(it != refs_.end(), "identifier not found");
            auto nh = refs_.extract(it);
            M_insist(std::holds_alternative<type>(nh.mapped()));
            return std::move(*std::get_if<type>(&nh.mapped()));
        }
 
        value_t get(const Schema::Identifier &id) const {
            M_insist(refs_.count(id) <= 1, "duplicate identifier, lookup ambiguous");
            auto it = refs_.find(id);
            M_insist(it != refs_.end(), "identifier not found");
            return std::visit(overloaded {
                [](auto &r) -> value_t { return r.clone(); },
                [](std::monostate) -> value_t { M_unreachable("invalid reference"); },
            }, it->second);
        }
        template<sql_type T>
        the_reference<T, IsConst> get(const Schema::Identifier &id) const {
            using type = the_reference<T, IsConst>;
            M_insist(refs_.count(id) <= 1, "duplicate identifier, lookup ambiguous");
            auto it = refs_.find(id);
            M_insist(it != refs_.end(), "identifier not found");
            M_insist(std::holds_alternative<type>(it->second));
            return std::get_if<type>(&it->second)->clone();
        }
    };
 
    using entry_t = the_entry<false>;
    using const_entry_t = the_entry<true>;
 
    using callback_t = std::function<void(const_entry_t)>;
    using hint_t = std::optional<Ptr<void>>;
 
    protected:
    static std::vector<SQL_t> clone(const std::vector<SQL_t> &values) {
        std::vector<SQL_t> cpy;
        cpy.reserve(values.size());
        for (auto &value : values) {
            std::visit(overloaded {
                [&](auto &v) { cpy.push_back(v.clone()); },
                [](std::monostate) { M_unreachable("invalid variant"); },
            }, value);
        }
        return cpy;
    }
 
    protected:
    std::reference_wrapper<const Schema> schema_; 
    std::vector<index_t> key_indices_; 
    std::vector<index_t> value_indices_; 
 
    public:
    HashTable() = delete;
 
    HashTable(const Schema &schema, std::vector<index_t> key_indices)
        : schema_(std::cref(schema))
        , key_indices_(std::move(key_indices))
    {
        M_insist(schema.num_entries() != 0, "hash table schema must not be empty");
        M_insist(not key_indices_.empty(), "must specify a key");
 
        /*----- Compute `value_indices_` as complement of `key_indices_`. -----*/
        for (index_t i = 0; i != schema.num_entries(); ++i) {
            if (contains(key_indices_, i)) continue;
            value_indices_.push_back(i);
        }
    }
 
    HashTable(const HashTable&) = delete;
    HashTable(HashTable&&) = default;
 
    virtual ~HashTable() { }
 
    const Schema & schema() const { return schema_; }
 
    virtual void setup() = 0;
    virtual void teardown() = 0;
 
    virtual void set_high_watermark(double percentage) = 0;
 
    virtual void clear() = 0;
 
    virtual Ptr<void> compute_bucket(std::vector<SQL_t> key) const = 0;
 
    virtual entry_t emplace(std::vector<SQL_t> key) = 0;
    virtual std::pair<entry_t, Boolx1> try_emplace(std::vector<SQL_t> key) = 0;
 
    virtual std::pair<entry_t, Boolx1> find(std::vector<SQL_t> key, hint_t bucket_hint = hint_t()) = 0;
    std::pair<const_entry_t, Boolx1> find(std::vector<SQL_t> key, hint_t bucket_hint = hint_t()) const {
        return const_cast<HashTable*>(this)->find(std::move(key), std::move(bucket_hint));
    }
 
    virtual void for_each(callback_t Pipeline) const = 0;
    virtual void for_each_in_equal_range(std::vector<SQL_t> key, callback_t Pipeline, bool predicated = false,
                                         hint_t bucket_hint = hint_t()) const = 0;
 
    virtual entry_t dummy_entry() = 0;
 
    protected:
    std::pair<size_t, size_t> set_byte_offsets(std::vector<offset_t> &offsets_in_bytes,
                                               const std::vector<const Type*> &types,
                                               offset_t initial_offset_in_bytes = 0,
                                               offset_t initial_max_alignment_in_bytes = 1);
};
 
 
/*----- chained hash tables ------------------------------------------------------------------------------------------*/
 
template<bool IsGlobal>
class chained_hash_table_storage;
 
template<>
class chained_hash_table_storage<false> {};
 
template<>
class chained_hash_table_storage<true>
{
    friend struct ChainedHashTable<true>;
 
    Global<Ptr<void>> address_; 
    Global<U32x1> mask_; 
    Global<U32x1> num_entries_; 
    Global<U32x1> high_watermark_absolute_; 
};
 
template<bool IsGlobal>
struct ChainedHashTable : HashTable
{
    private:
    template<typename T>
    using var_t = std::conditional_t<IsGlobal, Global<T>, Var<T>>;
 
    HashTable::size_t entry_size_in_bytes_; 
    HashTable::size_t entry_max_alignment_in_bytes_; 
    std::vector<HashTable::offset_t> entry_offsets_in_bytes_; 
    HashTable::offset_t null_bitmap_offset_in_bytes_;
    HashTable::offset_t ptr_offset_in_bytes_; 
 
    std::optional<Var<Ptr<void>>> address_; 
    std::optional<Var<U32x1>> mask_;
    std::optional<Var<U32x1>> num_entries_; 
    double high_watermark_percentage_ = 1.0; 
    std::optional<Var<U32x1>> high_watermark_absolute_;
    chained_hash_table_storage<IsGlobal> storage_;
    std::optional<FunctionProxy<void(void)>> rehash_;
    std::vector<std::pair<Ptr<void>, U32x1>> dummy_allocations_; 
    std::optional<var_t<Ptr<void>>> predication_dummy_; 
 
    public:
    ChainedHashTable(const Schema &schema, std::vector<HashTable::index_t> key_indices, uint32_t initial_capacity);
 
    ChainedHashTable(ChainedHashTable&&) = default;
 
    ~ChainedHashTable();
 
    private:
    Ptr<void> begin() const { M_insist(bool(address_), "must call `setup()` before"); return *address_; }
    Ptr<void> end() const { return begin() + size_in_bytes().make_signed(); }
    U32x1 mask() const { M_insist(bool(mask_), "must call `setup()` before"); return *mask_; }
    U32x1 capacity() const { return mask() + 1U; }
    U32x1 size_in_bytes() const { return capacity() * uint32_t(sizeof(uint32_t)); }
 
    public:
    void setup() override;
    void teardown() override;
 
    void set_high_watermark(double percentage) override {
        if (percentage < 1.0) {
            std::cerr << "warning: using chained collisions the load factor must be in [1,∞), ignore invalid value "
                      << percentage << std::endl;
            return;
        }
        high_watermark_percentage_ = percentage;
        update_high_watermark();
    }
    private:
    void update_high_watermark() {
        M_insist(bool(high_watermark_absolute_), "must call `setup()` before");
        auto high_watermark_absolute_new = high_watermark_percentage_ * capacity().make_signed().template to<double>();
        auto high_watermark_absolute_floored = high_watermark_absolute_new.template to<int32_t>().make_unsigned();
        Wasm_insist(high_watermark_absolute_floored.clone() >= 1U,
                    "at least one entry must be allowed to insert before growing the table");
        *high_watermark_absolute_ = high_watermark_absolute_floored;
    }
 
    void create_predication_dummy() {
        M_insist(not predication_dummy_);
        predication_dummy_.emplace(); // since globals cannot be constructed with runtime values
        *predication_dummy_ = Module::Allocator().allocate(sizeof(uint32_t), sizeof(uint32_t));
        *predication_dummy_->template to<uint32_t*>() = 0U; // set to nullptr
    }
 
    public:
    void clear() override;
 
    Ptr<void> compute_bucket(std::vector<SQL_t> key) const override;
 
    entry_t emplace(std::vector<SQL_t> key) override;
    std::pair<entry_t, Boolx1> try_emplace(std::vector<SQL_t> key) override;
 
    std::pair<entry_t, Boolx1> find(std::vector<SQL_t> key, hint_t bucket_hint) override;
 
    void for_each(callback_t Pipeline) const override;
    void for_each_in_equal_range(std::vector<SQL_t> key, callback_t Pipeline, bool predicated,
                                 hint_t bucket_hint) const override;
 
    entry_t dummy_entry() override;
 
    private:
    Ptr<void> hash_to_bucket(std::vector<SQL_t> key) const;
 
    entry_t emplace_without_rehashing(std::vector<SQL_t> key);
 
    Boolx1 equal_key(Ptr<void> entry, std::vector<SQL_t> key) const;
 
    void insert_key(Ptr<void> entry, std::vector<SQL_t> key);
 
    entry_t value_entry(Ptr<void> entry) const;
    const_entry_t entry(Ptr<void> entry) const;
 
    void rehash();
};
 
using LocalChainedHashTable = ChainedHashTable<false>;
using GlobalChainedHashTable = ChainedHashTable<true>;
 
 
/*----- open addressing hash tables ----------------------------------------------------------------------------------*/
 
struct OpenAddressingHashTableBase : HashTable
{
    friend struct ProbingStrategy;
 
    using ref_t = uint32_t; 
 
    struct ProbingStrategy
    {
        protected:
        const OpenAddressingHashTableBase &ht_; 
 
        public:
        ProbingStrategy(const OpenAddressingHashTableBase &ht) : ht_(ht) { }
 
        virtual ~ProbingStrategy() { }
 
        virtual Ptr<void> skip_slots(Ptr<void> bucket, U32x1 skips) const = 0;
        virtual Ptr<void> advance_to_next_slot(Ptr<void> slot, U32x1 current_step) const = 0;
    };
 
    protected:
    std::unique_ptr<ProbingStrategy> probing_strategy_; 
    HashTable::offset_t refs_offset_in_bytes_; 
    HashTable::size_t entry_size_in_bytes_; 
    HashTable::size_t entry_max_alignment_in_bytes_; 
    double high_watermark_percentage_ = 1.0; 
 
    public:
    OpenAddressingHashTableBase(const Schema &schema, std::vector<index_t> key_indices)
        : HashTable(schema, std::move(key_indices))
    { }
 
    virtual Ptr<void> begin() const = 0;
    virtual Ptr<void> end() const = 0;
    virtual U32x1 mask() const = 0;
    U32x1 capacity() const { return mask() + 1U; }
    U32x1 size_in_bytes() const { return capacity() * entry_size_in_bytes_; }
    HashTable::size_t entry_size_in_bytes() const { return entry_size_in_bytes_; }
 
    template<typename T>
    void set_probing_strategy() { probing_strategy_ = std::make_unique<T>(*this); }
    protected:
    const ProbingStrategy & probing_strategy() const { M_insist(bool(probing_strategy_)); return *probing_strategy_; }
 
    Reference<ref_t> reference_count(Ptr<void> entry) const { return *(entry + refs_offset_in_bytes_).to<ref_t*>(); }
 
    public:
    void set_high_watermark(double percentage) override {
        if (percentage < 0.5 or percentage >= 1.0) {
            std::cerr << "warning: using open addressing the load factor must be in [0.5,1), ignore invalid value "
                      << percentage << std::endl;
            return;
        }
        high_watermark_percentage_ = percentage;
        update_high_watermark();
    }
    protected:
    virtual void update_high_watermark() { }
 
    public:
    void clear() override;
 
    protected:
    Ptr<void> hash_to_bucket(std::vector<SQL_t> key) const;
};
 
template<bool ValueInPlace>
class open_addressing_hash_table_layout;
 
template<>
class open_addressing_hash_table_layout<true>
{
    friend struct OpenAddressingHashTable<false, true>;
    friend struct OpenAddressingHashTable<true, true>;
 
    std::vector<HashTable::offset_t> entry_offsets_in_bytes_;
    HashTable::offset_t null_bitmap_offset_in_bytes_; 
};
 
template<>
class open_addressing_hash_table_layout<false>
{
    friend struct OpenAddressingHashTable<false, false>;
    friend struct OpenAddressingHashTable<true, false>;
 
    std::vector<HashTable::offset_t> key_offsets_in_bytes_;
    std::vector<HashTable::offset_t> value_offsets_in_bytes_;
    HashTable::offset_t ptr_offset_in_bytes_; 
    HashTable::size_t values_size_in_bytes_;
    HashTable::size_t values_max_alignment_in_bytes_;
    HashTable::offset_t keys_null_bitmap_offset_in_bytes_; 
    HashTable::offset_t values_null_bitmap_offset_in_bytes_; 
};
 
template<bool IsGlobal>
class open_addressing_hash_table_storage;
 
template<>
class open_addressing_hash_table_storage<false> {};
 
template<>
class open_addressing_hash_table_storage<true>
{
    friend struct OpenAddressingHashTable<true, false>;
    friend struct OpenAddressingHashTable<true, true>;
 
    Global<Ptr<void>> address_; 
    Global<U32x1> mask_; 
    Global<U32x1> num_entries_; 
    Global<U32x1> high_watermark_absolute_; 
};
 
template<bool IsGlobal, bool ValueInPlace>
struct OpenAddressingHashTable : OpenAddressingHashTableBase
{
    private:
    template<typename T>
    using var_t = std::conditional_t<IsGlobal, Global<T>, Var<T>>;
 
    open_addressing_hash_table_layout<ValueInPlace> layout_; 
    std::optional<Var<Ptr<void>>> address_; 
    std::optional<Var<U32x1>> mask_; 
    std::optional<Var<U32x1>> num_entries_; 
    std::optional<Var<U32x1>> high_watermark_absolute_; 
    open_addressing_hash_table_storage<IsGlobal> storage_;
    std::optional<FunctionProxy<void(void)>> rehash_;
    std::vector<std::pair<Ptr<void>, U32x1>> dummy_allocations_; 
    std::optional<var_t<Ptr<void>>> predication_dummy_; 
 
    public:
    OpenAddressingHashTable(const Schema &schema, std::vector<HashTable::index_t> key_indices,
                            uint32_t initial_capacity);
 
    OpenAddressingHashTable(OpenAddressingHashTable&&) = default;
 
    ~OpenAddressingHashTable();
 
    private:
    Ptr<void> begin() const override { M_insist(bool(address_), "must call `setup()` before"); return *address_; }
    Ptr<void> end() const override { return begin() + (capacity() * entry_size_in_bytes_).make_signed(); }
    U32x1 mask() const override { M_insist(bool(mask_), "must call `setup()` before"); return *mask_; }
 
    public:
    void setup() override;
    void teardown() override;
 
    private:
    void update_high_watermark() override {
        M_insist(bool(high_watermark_absolute_), "must call `setup()` before");
        auto _capacity = capacity().make_signed().template to<double>();
        const Var<U32x1> high_watermark_absolute_new(
            (high_watermark_percentage_ * _capacity).ceil().template to<int32_t>().make_unsigned()
        );
        *high_watermark_absolute_ =
            Select(capacity() <= high_watermark_absolute_new, capacity() - 1U, high_watermark_absolute_new);
        Wasm_insist(*high_watermark_absolute_ >= 1U,
                    "at least one entry must be allowed to insert before growing the table");
        Wasm_insist(*high_watermark_absolute_ < capacity(), "at least one entry must always be unoccupied for lookups");
    }
 
    void create_predication_dummy() {
        M_insist(not predication_dummy_);
        predication_dummy_.emplace(); // since globals cannot be constructed with runtime values
        *predication_dummy_ = Module::Allocator().allocate(entry_size_in_bytes_, entry_max_alignment_in_bytes_);
        dummy_allocations_.emplace_back(*predication_dummy_, entry_size_in_bytes_);
        reference_count(*predication_dummy_) = ref_t(0); // set unoccupied
    }
 
    public:
    Ptr<void> compute_bucket(std::vector<SQL_t> key) const override;
 
    entry_t emplace(std::vector<SQL_t> key) override;
    std::pair<entry_t, Boolx1> try_emplace(std::vector<SQL_t> key) override;
 
    std::pair<entry_t, Boolx1> find(std::vector<SQL_t> key, hint_t bucket_hint) override;
 
    void for_each(callback_t Pipeline) const override;
    void for_each_in_equal_range(std::vector<SQL_t> key, callback_t Pipeline, bool predicated,
                                 hint_t bucket_hint) const override;
 
    entry_t dummy_entry() override;
 
    private:
    Ptr<void> emplace_without_rehashing(std::vector<SQL_t> key);
 
    Boolx1 equal_key(Ptr<void> slot, std::vector<SQL_t> key) const;
 
    void insert_key(Ptr<void> slot, std::vector<SQL_t> key);
 
    entry_t value_entry(Ptr<void> ptr) const;
    const_entry_t entry(Ptr<void> slot) const;
 
    void rehash();
};
 
using LocalOpenAddressingOutOfPlaceHashTable = OpenAddressingHashTable<false, false>;
using GlobalOpenAddressingOutOfPlaceHashTable = OpenAddressingHashTable<true, false>;
using LocalOpenAddressingInPlaceHashTable = OpenAddressingHashTable<false, true>;
using GlobalOpenAddressingInPlaceHashTable = OpenAddressingHashTable<true, true>;
 
 
/*----- probing strategies for open addressing hash tables -----------------------------------------------------------*/
 
struct LinearProbing : OpenAddressingHashTableBase::ProbingStrategy
{
    LinearProbing(const OpenAddressingHashTableBase &ht) : OpenAddressingHashTableBase::ProbingStrategy(ht) { }
 
    Ptr<void> skip_slots(Ptr<void> bucket, U32x1 skips) const override;
    Ptr<void> advance_to_next_slot(Ptr<void> slot, U32x1 current_step) const override;
};
 
struct QuadraticProbing : OpenAddressingHashTableBase::ProbingStrategy
{
    QuadraticProbing(const OpenAddressingHashTableBase &ht) : OpenAddressingHashTableBase::ProbingStrategy(ht) { }
 
    Ptr<void> skip_slots(Ptr<void> bucket, U32x1 skips) const override;
    Ptr<void> advance_to_next_slot(Ptr<void> slot, U32x1 current_step) const override;
};
 
 
/*======================================================================================================================
 * explicit instantiation declarations
 *====================================================================================================================*/
 
extern template void quicksort<false>(GlobalBuffer&, const std::vector<SortingOperator::order_type>&);
extern template void quicksort<true>(GlobalBuffer&, const std::vector<SortingOperator::order_type>&);
extern template struct m::wasm::ChainedHashTable<false>;
extern template struct m::wasm::ChainedHashTable<true>;
extern template struct m::wasm::OpenAddressingHashTable<false, false>;
extern template struct m::wasm::OpenAddressingHashTable<false, true>;
extern template struct m::wasm::OpenAddressingHashTable<true, false>;
extern template struct m::wasm::OpenAddressingHashTable<true, true>;
 
}
 
}