IndexKey.h

Go to the documentation of this file.

#ifndef INDEX_INDEXKEY_H
#define INDEX_INDEXKEY_H
 
#include "tdb.h"
 
#include "catalog/Schema.h"
 
namespace taco {
 
struct IndexKey;
 
typedef std::unique_ptr<IndexKey, AlignedAllocImpl::FreeMem> UniqueIndexKey;
 
struct IndexKey {
public:
    template<class SomeDatum>
    static UniqueIndexKey
    Create(const std::vector<SomeDatum> &data) {
        if (data.size() > (size_t) std::numeric_limits<FieldId>::max()) {
            LOG(kFatal, "too many fields in a key: %lu", data.size());
        }
        FieldId nkeys = (FieldId) data.size();
        size_t sz = ComputeStructSize(nkeys);
        void *new_key = aligned_alloc(8, sz);
        ConstructImpl(new_key, data.data(), nkeys, sz);
        return UniqueIndexKey((IndexKey*) new_key);
    }
 
    template<class SomeDatum>
    static UniqueIndexKey
    Create(SomeDatum *data, FieldId nkeys) {
        size_t sz = ComputeStructSize(nkeys);
        void *new_key = aligned_alloc(8, sz);
        ConstructImpl(new_key, data, nkeys, sz);
        return UniqueIndexKey((IndexKey*) new_key);
    }
 
    template<class SomeDatum>
    static void
    Construct(void *key, const std::vector<SomeDatum> &data) {
        if (data.size() > (size_t) std::numeric_limits<FieldId>::max()) {
            LOG(kFatal, "too many fields in a key: %lu", data.size());
        }
        FieldId nkeys = (FieldId) data.size();
        size_t sz = ComputeStructSize(nkeys);
        ConstructImpl(key, data.data(), nkeys, sz);
    }
 
    template<class SomeDatum>
    static void
    Construct(void *key, SomeDatum *data, FieldId nkeys) {
        size_t sz = ComputeStructSize(nkeys);
        ConstructImpl(key, data, nkeys, sz);
    }
 
    UniqueIndexKey
    Copy() const {
        size_t sz = ComputeStructSize(GetNumKeys());
        void *new_key = aligned_alloc(8, sz);
        ConstructImpl(new_key, &GetKey(0), GetNumKeys(), sz);
        return UniqueIndexKey((IndexKey*) new_key);
    }
 
    void
    CopyTo(IndexKey *key2) const {
        memcpy((void*) key2, (void*) this, GetStructSize());
    }
 
    void
    DeepCopy(const Schema *sch, std::vector<Datum> &data_buffer) {
        FieldId nkeys = GetNumKeys();
        ASSERT(nkeys <= sch->GetNumFields());
        for (FieldId i = 0; i < nkeys; ++i) {
            if (!IsNull(i) && sch->FieldPassByRef(i)) {
                Datum &old_datum = GetKey(i).GetDatum();
                if (data_buffer.size() == data_buffer.capacity() &&
                    !data_buffer.empty()) {
                    LOG(kFatal, "causing the data buffer to increase capacity "
                            "and this will make previously deep copied "
                            "IndexKey contain dangling pointers (need to "
                            "reserve space on the vector first!).");
                }
                data_buffer.emplace_back(old_datum.DeepCopy());
                GetKey(i) = data_buffer.back();
            }
        }
    }
 
    inline FieldId
    GetNumKeys() const {
        return *reinterpret_cast<const FieldId*>(this);
    }
 
    inline DatumRef&
    GetKey(FieldId keyid) const {
        return ((DatumRef*)(((uint8_t*) m_data)
            + MAXALIGN(2 + ((GetNumKeys() + 7) >> 3))))[keyid];
    }
 
    template<class SomeDatum>
    void
    SetKey(FieldId keyid, const SomeDatum &key) {
        if (key.isnull()) {
            SetNullBit(keyid, true);
            // The datum is left as is here because it is undefined to
            // call GetKey() if a field is declared as null.
        } else {
            SetNullBit(keyid, false);
            // relying on implicit conversions if SomeDatum is not DatumRef
            GetKey(keyid) = key;
        }
    }
 
    inline bool
    IsNull(FieldId keyid) const {
        if (keyid >= GetNumKeys()) {
            return true;
       }
        return !!(((uint8_t*) &m_data)[2 + (keyid >> 3)] & (1 << (keyid & 7)));
    }
 
    inline void
    SetNullBit(FieldId keyid, bool isnull) const {
        if (isnull)
            ((uint8_t*) &m_data)[2 + (keyid >> 3)] |= (1 << (keyid & 7));
        else
            ((uint8_t*) &m_data)[2 + (keyid >> 3)] &=
                ~(uint8_t)(1 << (keyid & 7));
    }
 
    inline bool
    HasAnyNull() const {
        uint8_t *bm = ((uint8_t *) &m_data) + 2;
        size_t n = GetNumKeys() >> 3;
        for (size_t i = 0; i < n; ++i) {
            // this is ok because we zero filled the null bitmap initially.
            if (bm[i] != 0)
                return true;
        }
        return false;
    }
 
    inline size_t
    GetStructSize() const {
        return ComputeStructSize(GetNumKeys());
    }
 
    inline static size_t
    ComputeStructSize(FieldId nkeys) {
        size_t sz = MAXALIGN(2 + ((nkeys + 7) >> 3)) + nkeys * sizeof(DatumRef);
        return sz;
    }
 
    inline NullableDatumRef
    GetNullableKey(FieldId keyid) const {
        if (IsNull(keyid)) {
            return NullableDatumRef(Datum::FromNull());
        }
        return NullableDatumRef(GetKey(keyid));
    }
 
    std::vector<NullableDatumRef>
    ToNullableDatumVector() const {
        std::vector<NullableDatumRef> vec;
        vec.reserve(GetNumKeys());
        for (FieldId i = 0; i < GetNumKeys(); ++i) {
            vec.emplace_back(GetNullableKey(i));
        }
        return vec;
    }
 
    IndexKey(const IndexKey&) = delete;
    IndexKey(IndexKey&&) = delete;
    IndexKey &operator=(const IndexKey&) = delete;
    IndexKey &operator=(IndexKey&&) = delete;
 
private:
    template<class SomeDatum>
    static void
    ConstructImpl(void *key, SomeDatum *data,
                  FieldId nkeys, size_t sz) {
        *(FieldId *)key = nkeys;
        uint8_t *bm = ((uint8_t *) key ) + 2;
        DatumRef *key_data =
            (DatumRef*)(((uint8_t *) key) + sz - nkeys * sizeof(DatumRef));
        memset(bm, 0, ((nkeys + 7) >> 3));
        for (FieldId i = 0; i < nkeys; ++i) {
            if (data[i].isnull()) {
                bm[i >> 3] |= 1 << (i & 7);
            } else {
                key_data[i] = DatumRef(data[i]);
            }
        }
    }
 
    // forces 8-byte alignment
    uint64_t m_data[1];
};
 
}   // namespace taco
 
#endif      // INDEX_INDEXKEY_H