table.h

Go to the documentation of this file.

 
 
#ifndef PSTSDK_LTP_TABLE_H
#define PSTSDK_LTP_TABLE_H
 
#include <vector>
 
#include <unordered_map>
#include <boost/iterator/iterator_facade.hpp>
 
#include "pstsdk/util/primitives.h"
 
#include "pstsdk/ndb/node.h"
 
#include "pstsdk/ltp/object.h"
#include "pstsdk/ltp/heap.h"
 
namespace pstsdk
{
 
class table_impl;
 
typedef std::shared_ptr<table_impl> table_ptr;
typedef std::shared_ptr<const table_impl> const_table_ptr;
 
table_ptr open_table(const node& n);
table_ptr open_table(const node&, alias_tag);
 
class const_table_row : public const_property_object
{
public:
    const_table_row(const const_table_row& other)
        : m_position(other.m_position), m_table(other.m_table) { }
    const_table_row(ulong position, const const_table_ptr& table)
        : m_position(position), m_table(table) { }
 
    row_id get_row_id() const;
 
    // const_property_object
    std::vector<prop_id> get_prop_list() const;
    prop_type get_prop_type(prop_id id) const;
    bool prop_exists(prop_id id) const;
    size_t size(prop_id id) const;
    hnid_stream_device open_prop_stream(prop_id id);
 
private:
    // const_property_object
    byte get_value_1(prop_id id) const;
    ushort get_value_2(prop_id id) const;
    ulong get_value_4(prop_id id) const;
    ulonglong get_value_8(prop_id id) const;
    std::vector<byte> get_value_variable(prop_id id) const;
 
    ulong m_position;           
    const_table_ptr m_table;    
};
 
class const_table_row_iter : public boost::iterator_facade<const_table_row_iter, const_table_row, boost::random_access_traversal_tag, const_table_row>
{
public:
    const_table_row_iter()
        : m_position(0) { }
 
    const_table_row_iter(ulong pos, const const_table_ptr& table) 
        : m_position(pos), m_table(table)  { }
 
private:
    friend class boost::iterator_core_access;
 
    void increment() { ++m_position; }
    bool equal(const const_table_row_iter& other) const
        { return ((m_position == other.m_position) && (m_table == other.m_table)); }
    const_table_row dereference() const
        { return const_table_row(m_position, m_table); }
    void decrement() { --m_position; }
    void advance(int off) { m_position += off; }
    size_t distance_to(const const_table_row_iter& other) const
        { return (other.m_position - m_position); }
 
    ulong m_position;
    const_table_ptr m_table;
};
 
class table_impl : public std::enable_shared_from_this<table_impl>
{
public:
    virtual ~table_impl() { }
    virtual ulong lookup_row(row_id id) const = 0;
 
    const_table_row operator[](ulong row) const
        { return const_table_row(row, shared_from_this()); }
    const_table_row_iter begin() const
        { return const_table_row_iter(0, shared_from_this()); }
    const_table_row_iter end() const
        { return const_table_row_iter(size(), shared_from_this()); }
    
    virtual node& get_node() = 0;
    virtual const node& get_node() const = 0;
    virtual ulonglong get_cell_value(ulong row, prop_id id) const = 0;
    virtual std::vector<byte> read_cell(ulong row, prop_id id) const = 0;
    virtual hnid_stream_device open_cell_stream(ulong row, prop_id id) = 0;
    virtual std::vector<prop_id> get_prop_list() const = 0;
    virtual prop_type get_prop_type(prop_id id) const = 0;
    virtual row_id get_row_id(ulong row) const = 0;
    virtual size_t size() const = 0;
    virtual bool prop_exists(ulong row, prop_id id) const = 0;
    virtual size_t row_prop_size(ulong row, prop_id id) const = 0;
};
 
template<typename T>
class basic_table : public table_impl
{
public:
    node& get_node() 
        { return m_prows->get_node(); }
    const node& get_node() const
        { return m_prows->get_node(); }
    ulong lookup_row(row_id id) const;
    ulonglong get_cell_value(ulong row, prop_id id) const;
    std::vector<byte> read_cell(ulong row, prop_id id) const;
    hnid_stream_device open_cell_stream(ulong row, prop_id id);
    std::vector<prop_id> get_prop_list() const;
    prop_type get_prop_type(prop_id id) const;
    row_id get_row_id(ulong row) const;
    size_t size() const;
    bool prop_exists(ulong row, prop_id id) const;
    size_t row_prop_size(ulong row, prop_id id) const;
 
private:
    friend table_ptr open_table(const node& n);
    friend table_ptr open_table(const node& n, alias_tag);
    basic_table(const node& n);
    basic_table(const node& n, alias_tag);
 
    std::shared_ptr<bth_node<row_id, T> > m_prows;
 
    // only one of the following two items is valid
    std::vector<byte> m_vec_rowarray;
    std::shared_ptr<node> m_pnode_rowarray;
 
    std::unordered_map<prop_id, disk::column_description> m_columns;
    typedef std::unordered_map<prop_id, disk::column_description>::iterator column_iter;
    typedef std::unordered_map<prop_id, disk::column_description>::const_iterator const_column_iter;
 
    ushort m_offsets[disk::tc_offsets_max];
 
    // helper functions
    ulong cb_per_row() const { return m_offsets[disk::tc_offsets_bitmap]; }
    ulong exists_bitmap_start() const { return m_offsets[disk::tc_offsets_one]; }
    ulong rows_per_page() const { return (m_pnode_rowarray ? m_pnode_rowarray->get_page_size(0) / cb_per_row() : m_vec_rowarray.size() / cb_per_row()); }
    template<typename Val> Val read_raw_row(ulong row, ushort offset) const;
    std::vector<byte> read_exists_bitmap(ulong row) const;
};
 
typedef basic_table<ushort> small_table;
typedef basic_table<ulong> large_table;
 
class table
{
public:
    explicit table(const node& n);
    table(const node& n, alias_tag);
    table(const table& other);
    table(const table& other, alias_tag)
        : m_ptable(other.m_ptable) { }
 
    const_table_row operator[](ulong row) const
        { return (*m_ptable)[row]; }
    const_table_row_iter begin() const
        { return m_ptable->begin(); }
    const_table_row_iter end() const
        { return m_ptable->end(); }
 
    node& get_node() 
        { return m_ptable->get_node(); }
    const node& get_node() const
        { return m_ptable->get_node(); }
    ulonglong get_cell_value(ulong row, prop_id id) const
        { return m_ptable->get_cell_value(row, id); }
    std::vector<byte> read_cell(ulong row, prop_id id) const
        { return m_ptable->read_cell(row, id); }
    hnid_stream_device open_cell_stream(ulong row, prop_id id)
        { return m_ptable->open_cell_stream(row, id); }
    std::vector<prop_id> get_prop_list() const
        { return m_ptable->get_prop_list(); }
    prop_type get_prop_type(prop_id id) const
        { return m_ptable->get_prop_type(id); }
    row_id get_row_id(ulong row) const
        { return m_ptable->get_row_id(row); }
    ulong lookup_row(row_id id) const
        { return m_ptable->lookup_row(id); }
    size_t size() const
        { return m_ptable->size(); }
private:
    table();
 
    table_ptr m_ptable;
};
 
} // end pstsdk namespace
 
inline pstsdk::table_ptr pstsdk::open_table(const node& n)
{
    if(n.get_id() == nid_all_message_search_contents)
    {
        //return table_ptr(new gust(n));
        throw not_implemented("gust table");
    }
 
    heap h(n);
    std::vector<byte> table_info = h.read(h.get_root_id());
    disk::tc_header* pheader = (disk::tc_header*)&table_info[0];
 
    std::vector<byte> bth_info = h.read(pheader->row_btree_id);
    disk::bth_header* pbthheader = (disk::bth_header*)&bth_info[0];
 
    if(pbthheader->entry_size == 4)
       return table_ptr(new large_table(n));
    else
       return table_ptr(new small_table(n));
}
 
inline pstsdk::table_ptr pstsdk::open_table(const node& n, alias_tag)
{
    if(n.get_id() == nid_all_message_search_contents)
    {
        //return table_ptr(new gust(n));
        throw not_implemented("gust table");
    }
 
    heap h(n);
    std::vector<byte> table_info = h.read(h.get_root_id());
    disk::tc_header* pheader = (disk::tc_header*)&table_info[0];
 
    std::vector<byte> bth_info = h.read(pheader->row_btree_id);
    disk::bth_header* pbthheader = (disk::bth_header*)&bth_info[0];
 
    if(pbthheader->entry_size == 4)
       return table_ptr(new large_table(n, alias_tag()));
    else
       return table_ptr(new small_table(n, alias_tag()));
}
 
inline std::vector<pstsdk::prop_id> pstsdk::const_table_row::get_prop_list() const
{
    std::vector<prop_id> columns = m_table->get_prop_list();
    std::vector<prop_id> props;
 
    for(size_t i = 0; i < columns.size(); ++i)
    {
        if(prop_exists(columns[i]))
            props.push_back(columns[i]);
    }
 
    return props;
}
 
inline size_t pstsdk::const_table_row::size(prop_id id) const
{
    return m_table->row_prop_size(m_position, id);
}
 
inline pstsdk::prop_type pstsdk::const_table_row::get_prop_type(prop_id id) const
{
    return m_table->get_prop_type(id);
}
 
inline bool pstsdk::const_table_row::prop_exists(prop_id id) const
{
    return m_table->prop_exists(m_position, id);
}
 
inline pstsdk::row_id pstsdk::const_table_row::get_row_id() const
{
    return m_table->get_row_id(m_position);
}
 
inline pstsdk::byte pstsdk::const_table_row::get_value_1(prop_id id) const
{
    return (byte)m_table->get_cell_value(m_position, id); 
}
 
inline pstsdk::ushort pstsdk::const_table_row::get_value_2(prop_id id) const
{
    return (ushort)m_table->get_cell_value(m_position, id); 
}
 
inline pstsdk::ulong pstsdk::const_table_row::get_value_4(prop_id id) const
{
    return (ulong)m_table->get_cell_value(m_position, id); 
}
 
inline pstsdk::ulonglong pstsdk::const_table_row::get_value_8(prop_id id) const
{
    return m_table->get_cell_value(m_position, id); 
}
 
inline std::vector<pstsdk::byte> pstsdk::const_table_row::get_value_variable(prop_id id) const
{ 
    return m_table->read_cell(m_position, id); 
}
 
inline pstsdk::hnid_stream_device pstsdk::const_table_row::open_prop_stream(prop_id id)
{
    return (std::const_pointer_cast<table_impl>(m_table))->open_cell_stream(m_position, id);
}
 
template<typename T>
inline pstsdk::basic_table<T>::basic_table(const node& n)
{
    heap h(n, disk::heap_sig_tc);
 
    std::vector<byte> table_info = h.read(h.get_root_id());
    disk::tc_header* pheader = (disk::tc_header*)&table_info[0];
 
#ifdef PSTSDK_VALIDATION_LEVEL_WEAK
    if(pheader->signature != disk::heap_sig_tc)
        throw sig_mismatch("heap_sig_tc expected", 0, n.get_id(), pheader->signature, disk::heap_sig_tc);
#endif
 
    m_prows = h.open_bth<row_id, T>(pheader->row_btree_id);
 
    for(int i = 0; i < pheader->num_columns; ++i)
        m_columns[pheader->columns[i].id] = pheader->columns[i];
 
    for(int i = 0; i < disk::tc_offsets_max; ++i)
        m_offsets[i] = pheader->size_offsets[i];
 
    if(is_subnode_id(pheader->row_matrix_id))
    {
        m_pnode_rowarray.reset(new node(n.lookup(pheader->row_matrix_id)));
    }
    else if(pheader->row_matrix_id)
    {
        m_vec_rowarray = h.read(pheader->row_matrix_id);
    }
}
 
template<typename T>
inline pstsdk::basic_table<T>::basic_table(const node& n, alias_tag)
{
    heap h(n, disk::heap_sig_tc, alias_tag());
 
    std::vector<byte> table_info = h.read(h.get_root_id());
    disk::tc_header* pheader = (disk::tc_header*)&table_info[0];
 
#ifdef PSTSDK_VALIDATION_LEVEL_WEAK
    if(pheader->signature != disk::heap_sig_tc)
        throw sig_mismatch("heap_sig_tc expected", 0, n.get_id(), pheader->signature, disk::heap_sig_tc);
#endif
 
    m_prows = h.open_bth<row_id, T>(pheader->row_btree_id);
 
    for(int i = 0; i < pheader->num_columns; ++i)
        m_columns[pheader->columns[i].id] = pheader->columns[i];
 
    for(int i = 0; i < disk::tc_offsets_max; ++i)
        m_offsets[i] = pheader->size_offsets[i];
 
    if(is_subnode_id(pheader->row_matrix_id))
    {
        m_pnode_rowarray.reset(new node(n.lookup(pheader->row_matrix_id)));
    }
    else if(pheader->row_matrix_id)
    {
        m_vec_rowarray = h.read(pheader->row_matrix_id);
    }
}
 
template<typename T>
inline size_t pstsdk::basic_table<T>::size() const
{
    if(m_pnode_rowarray)
    {
        return (m_pnode_rowarray->get_page_count()-1) * rows_per_page() + m_pnode_rowarray->get_page_size(m_pnode_rowarray->get_page_count()-1) / cb_per_row();
    }
    else 
    {
        return m_vec_rowarray.size() / cb_per_row();
    }
}
 
template<typename T>
inline std::vector<pstsdk::prop_id> pstsdk::basic_table<T>::get_prop_list() const
{
    std::vector<prop_id> props;
 
    for(const_column_iter i = m_columns.begin(); i != m_columns.end(); ++i)
        props.push_back(i->first);
 
    return props;
}
 
template<typename T>
inline pstsdk::ulong pstsdk::basic_table<T>::lookup_row(row_id id) const
{ 
    try 
    { 
        return (ulong)m_prows->lookup(id); 
    } 
    catch(key_not_found<T>& e) 
    { 
        throw key_not_found<row_id>(e.which()); 
    } 
}
 
template<typename T>
inline pstsdk::ulonglong pstsdk::basic_table<T>::get_cell_value(ulong row, prop_id id) const
{
    if(!prop_exists(row, id))
        throw key_not_found<prop_id>(id);
 
    const_column_iter column = m_columns.find(id);
    ulonglong value;
 
    switch(column->second.size)
    {
        case 8:
            value = read_raw_row<ulonglong>(row, column->second.offset);
            break;
        case 4:
            value = read_raw_row<ulong>(row, column->second.offset);
            break;
        case 2:
            value = read_raw_row<ushort>(row, column->second.offset);
            break;
        case 1:
            value = read_raw_row<byte>(row, column->second.offset);
            break;
        default:
            throw database_corrupt("get_cell_value: invalid cell size");
    }
 
    return value;
}
 
template<typename T>
inline size_t pstsdk::basic_table<T>::row_prop_size(ulong row, prop_id id) const
{
    heapnode_id hid = static_cast<heapnode_id>(get_cell_value(row, id));
 
    if(is_subnode_id(hid))
        return get_node().lookup(hid).size();
    else
        return m_prows->get_heap_ptr()->size(hid);
}
 
template<typename T>
inline std::vector<pstsdk::byte> pstsdk::basic_table<T>::read_cell(ulong row, prop_id id) const
{
    heapnode_id hid = static_cast<heapnode_id>(get_cell_value(row, id));
    std::vector<byte> buffer;
 
    if(is_subnode_id(hid))
    {
        node sub(get_node().lookup(hid));
        buffer.resize(sub.size());
        sub.read(buffer, 0);
    }
    else
    {
        buffer = m_prows->get_heap_ptr()->read(hid);
    }
    return buffer;
}
 
template<typename T>
inline pstsdk::hnid_stream_device pstsdk::basic_table<T>::open_cell_stream(ulong row, prop_id id)
{
    heapnode_id hid = static_cast<heapnode_id>(get_cell_value(row, id));
 
    if(is_subnode_id(hid))
        return get_node().lookup(hid).open_as_stream();
    else
        return m_prows->get_heap_ptr()->open_stream(hid);
}
 
template<typename T>
inline pstsdk::prop_type pstsdk::basic_table<T>::get_prop_type(prop_id id) const
{
    const_column_iter iter = m_columns.find(id);
 
    if(iter == m_columns.end())
        throw key_not_found<prop_id>(id);
 
    return (prop_type)iter->second.type;
}
 
template<typename T>
inline pstsdk::row_id pstsdk::basic_table<T>::get_row_id(ulong row) const
{
    return read_raw_row<row_id>(row, 0);
}
 
template<typename T>
template<typename Val>
inline Val pstsdk::basic_table<T>::read_raw_row(ulong row, ushort offset) const
{
    if(row >= size())
        throw std::out_of_range("row >= size()");
 
    if(m_pnode_rowarray)
    {
        ulong page_num = row / rows_per_page();
        ulong page_offset = (row % rows_per_page()) * cb_per_row();
 
        return m_pnode_rowarray->read<Val>(page_num, page_offset+offset);
    }
    else
    {
        Val val;
        memcpy(&val, &m_vec_rowarray[ row * cb_per_row() + offset ], sizeof(Val));
        return val;
    }
}
 
template<typename T>
inline std::vector<pstsdk::byte> pstsdk::basic_table<T>::read_exists_bitmap(ulong row) const
{
    std::vector<byte> exists_bitmap(cb_per_row() - exists_bitmap_start());
 
    if(row >= size())
        throw std::out_of_range("row >= size()");
 
    if(m_pnode_rowarray)
    {
        ulong page_num = row / rows_per_page();
        ulong page_offset = (row % rows_per_page()) * cb_per_row();
 
        m_pnode_rowarray->read(exists_bitmap, page_num, page_offset + exists_bitmap_start());
    }
    else
    {
        memcpy(&exists_bitmap[0], &m_vec_rowarray[ row * cb_per_row() + exists_bitmap_start() ], exists_bitmap.size());
    }    
 
    return exists_bitmap;
}
 
template<typename T>
inline bool pstsdk::basic_table<T>::prop_exists(ulong row, prop_id id) const
{
    const_column_iter column = m_columns.find(id);
 
    if(column == m_columns.end())
        return false;
 
    std::vector<byte> exists_map = read_exists_bitmap(row);
 
    return test_bit(&exists_map[0], column->second.bit_offset);
}
 
inline pstsdk::table::table(const node& n)
{
    m_ptable = open_table(n);
}
 
inline pstsdk::table::table(const table& other)
{
    m_ptable = open_table(other.m_ptable->get_node());
}
 
#endif