btree_page_h.h

Go to the documentation of this file.

/*
 * (c) Copyright 2011-2014, Hewlett-Packard Development Company, LP
 */

#ifndef __BTREE_PAGE_H_H
#define __BTREE_PAGE_H_H

#include "btree_page.h"

#include "w_defines.h"
#include "fixable_page_h.h"
#include "w_key.h"
#include "w_endian.h"
#include "w_base.h"

typedef w_base_t::base_stat_t base_stat_t;

struct btree_lf_stats_t;
struct btree_int_stats_t;

/*
* When this page belongs to a foster chain, we need to store
* high-fence of right-most sibling in every sibling to do
* batch-verification with bitmaps.  @see
* btree_impl::_ux_verify_volume()
*/

class btree_page_h;

class btrec_t {
public:
    btrec_t() {};

    btrec_t(const btree_page_h& page, slotid_t slot);

    ~btrec_t() {};

    btrec_t& set(const btree_page_h& page, slotid_t slot);

    smsize_t elen() const {
        return _elem.size();
    }

    const w_keystr_t& key() const {
        return _key;
    }

    const cvec_t& elem() const {
        return _elem;
    }

    PageID child() const {
        return _child;
    }

    const lsn_t& child_emlsn() const {
        return _child_emlsn;
    }

    bool is_ghost_record() const {
        return _ghost_record;
    }

private:
    friend class btree_page_h;

    bool _ghost_record;

    w_keystr_t _key;

    PageID _child;  // opaque pointer
    lsn_t _child_emlsn;

    cvec_t _elem;

    // disabled
    btrec_t(const btrec_t&);

    btrec_t& operator=(const btrec_t&);
};

inline btrec_t::btrec_t(const btree_page_h& page, slotid_t slot) {
    set(page, slot);
}

class btree_impl;

template<class T> class btree_ghost_t;
class btree_ghost_mark_log;
class btree_ghost_reclaim_log;

class btree_page_h : public fixable_page_h {
    friend class btree_impl;

    template<class T> friend class btree_ghost_t;
    friend class btree_ghost_mark_log;
    friend class btree_ghost_reclaim_log;

    template<class T> friend class page_img_format_t;
    friend class btree_split_log;

    btree_page* page() const {
        return reinterpret_cast<btree_page*>(_pp);
    }

    enum {
        data_sz = btree_page::data_sz,
        hdr_sz = btree_page::hdr_sz,
    };

    // Used for page rebalance log record generation purpose
    union record_info_int16_convert {
        char c[sizeof(int16_t)];// int16 = 2 bytes, enough for max. record size in current implementation
        int16_t is_ghost : 1,        // 1 = ghost, 0 = non-ghost
        len : 15;            // record length
    };

    // Used for page rebalance log record generation purpose
    union record_info_shpid_convert {
        char c[sizeof(PageID)];// int32 = 4 bytes
        PageID i;
    };

public:
    // ======================================================================
    //   BEGIN: Struct/Enum/Constructor
    // ======================================================================

    btree_page_h() {}

    btree_page_h(const btree_page_h& p) : fixable_page_h(p) {}

    ~btree_page_h() {}

    btree_page_h& operator=(btree_page_h& p) {
        fixable_page_h::operator=(p);
        w_assert1(_pp->tag == t_btree_p);
        return *this;
    }


    // ======================================================================
    //   BEGIN: Header Get/Set functions
    // ======================================================================

    PageID btree_root() const {
        return page()->btree_root;
    }

    smsize_t used_space() const;

    // Total usable space on page
    smsize_t usable_space() const;

    int level() const;

    PageID pid0() const;

    PageID pid0_opaqueptr() const;

    PageID root() const;

    bool is_leaf() const;

    bool is_node() const;

    bool is_leaf_parent() const;

    PageID get_foster() const;

    PageID get_foster_opaqueptr() const;

    const char* get_prefix_key() const;

    int16_t get_prefix_length() const;

    const char* get_fence_low_key() const;

    int16_t get_fence_low_length() const;

    void copy_fence_low_key(w_keystr_t& buffer) const {
        buffer.construct_from_keystr(get_fence_low_key(), get_fence_low_length());
    }

    bool is_fence_low_infimum() const {
        return get_fence_low_key()[0] == SIGN_NEGINF;
    }

    const char* get_fence_high_key_noprefix() const;

    int16_t get_fence_high_length() const;

    int16_t get_fence_high_length_noprefix() const {
        return get_fence_high_length() - get_prefix_length();
    }

    void copy_fence_high_key(w_keystr_t& buffer) const {
        buffer.construct_from_keystr(get_prefix_key(), get_prefix_length(),
                                     get_fence_high_key_noprefix(),
                                     get_fence_high_length_noprefix());
    }

    bool is_fence_high_supremum() const {
        return get_prefix_length() == 0 && get_fence_high_key_noprefix()[0] == SIGN_POSINF;
    }

    const char* get_chain_fence_high_key() const;

    int16_t get_chain_fence_high_length() const;

    void copy_chain_fence_high_key(w_keystr_t& buffer) const {
        buffer.construct_from_keystr(get_chain_fence_high_key(), get_chain_fence_high_length());
    }

    bool fence_contains(const w_keystr_t& key) const;

    int compare_with_fence_low(const w_keystr_t& key) const;

    int compare_with_fence_low(const char* key, size_t key_len) const;

    int compare_with_fence_low_noprefix(const char* key, size_t key_len) const;

    int compare_with_fence_high(const w_keystr_t& key) const;

    int compare_with_fence_high(const char* key, size_t key_len) const;

    int compare_with_fence_high_noprefix(const char* key, size_t key_len) const;

    int compare_with_chain_fence_high(const w_keystr_t& key) const;

    int compare_with_chain_fence_high(const char* key, size_t key_len) const;
    // no 'noprefix' version because chain_fence_high might not share the prefix!

    void accept_empty_child(lsn_t new_lsn, PageID new_page_id, const bool f_redo);

    rc_t format_foster_child(btree_page_h& parent, const PageID& new_page_id,
                             const w_keystr_t& triggering_key,
                             w_keystr_t& split_key, // OUT: actual split key
                             int& move_count);

    bool set_foster_child(PageID foster_child_pid,
                          const w_keystr_t& new_fence_high, const w_keystr_t& child_fence_chain);

    void delete_range(int from, int to);

    rc_t format_steal(lsn_t new_lsn,
                      const PageID& pid,
                      StoreID store,
                      PageID root,
                      int level,
                      PageID pid0,
                      lsn_t pid0_emlsn,
                      PageID foster,
                      lsn_t foster_emlsn,
                      const w_keystr_t& fence_low,
                      const w_keystr_t& fence_high,
                      const w_keystr_t& chain_fence_high,
                      bool log_it = true,
                      btree_page_h* steal_src1 = nullptr,
                      int steal_from1 = 0,
                      int steal_to1 = 0,
                      btree_page_h* steal_src2 = nullptr,
                      int steal_from2 = 0,
                      int steal_to2 = 0,
                      bool steal_src2_pid0 = false,
                      const bool full_logging = false,  // True if doing full logging for record movement
                      const bool log_src_1 = false,     // Use only if full_logging = true
            // True if log movements from src1
            // False if log movements from src2
                      const bool ghost = false          // When _init the page, should the fence key record be a ghost
                     );

    void _steal_records(btree_page_h* steal_src,
                        int steal_from,
                        int steal_to,
                        const bool full_logging);

    // Set the node fence keys, no change in data and other page information except record count (if move out)
    // A special function used by Single-Page-Recovery REDO operation for page rebalance and page merge
    // when full logging is on (no minimal logging), set the page fence key before the
    // fully logged actual record movements
    rc_t init_fence_keys(const bool set_low, const w_keystr_t& low,
                         const bool set_high, const w_keystr_t& high,
                         const bool set_chain, const w_keystr_t& chain_fence_high,
                         const bool set_pid0, const PageID new_pid0,
                         const bool set_emlsn, const lsn_t new_pid0_emlsn,
                         const bool set_foster, const PageID foster_pid0,
                         const bool set_foster_emlsn, const lsn_t foster_emlsn,
                         const int remove_count = 0);

    rc_t norecord_split(PageID foster, lsn_t foster_emlsn,
                        const w_keystr_t& fence_high,
                        const w_keystr_t& chain_fence_high);

    bool check_chance_for_norecord_split(const w_keystr_t& key_to_insert) const;


    // ======================================================================
    //   BEGIN: Public record access functions
    // ======================================================================

    int nrecs() const;

    int nghosts() const;

    bool is_ghost(slotid_t slot) const;

    void get_key(slotid_t slot, w_keystr_t& key) const;

    const char* element(int slot, smsize_t& len, bool& ghost) const;

    bool copy_element(int slot, char* out_buffer, smsize_t& len, bool& ghost) const;

    PageID child(slotid_t slot) const;

    PageID child_opaqueptr(slotid_t slot) const;

    PageID* page_pointer_address(int offset);

    size_t get_rec_space(int slot) const;

    // Determine the prefix length for a page rebalance operation
    int16_t calculate_prefix_length(int16_t existing_prefix_len,  // Existing prefix length
                                    w_keystr_t low_key,           // New low fence
                                    w_keystr_t high_key);         // New high fence

    // Copy the specified number of records into provided buffer (no ghost)
    // each record must be reconstructed back to the original format (key+data)
    // Assumption - this function is for page split and page merge purpose,
    // therefore copy the last 'rec_count' records in the page, in page merge case
    // 'rec_count' == nrecs()
    rc_t copy_records(const int rec_count,          // In: number of records to copy
                      char* data_buffer,       // Out: caller provided data buffer
                      smsize_t& len);              // In: buffer size
    // out: data length

    // Function to insert records into a page, this is used for page rebalance purpose only
    // while the incoming buffer contains all the record data plus whatever extra information
    // per record
    rc_t insert_records_dest_redo(const int16_t prefix_len,      // In: source page prefix length
                                  const int32_t move_count,      //In: how many records to insert
                                  const int16_t record_data_len, // In: total length of data with extra info
                                  const char* record_data);      // In: Data buffer


    // ======================================================================
    //   BEGIN: Search functions
    // ======================================================================

    void search(const w_keystr_t& key,
                bool& found_key,
                slotid_t& return_slot) const {
        search((const char*)key.buffer_as_keystr(), key.get_length_as_keystr(), found_key, return_slot);
    }

    void search(const char* key_raw, size_t key_raw_len,
                bool& found_key, slotid_t& return_slot) const;

    void search_node(const w_keystr_t& key,
                     slotid_t& return_slot) const;


    // ======================================================================
    //   BEGIN: Insert/Update/Delete functions
    // ======================================================================

    rc_t insert_node(const w_keystr_t& key,
                     slotid_t slot,
                     PageID child,
                     const lsn_t& child_emlsn);

    void mark_ghost(slotid_t slot);

    void unmark_ghost(slotid_t slot);

    rc_t replace_ghost(const w_keystr_t& key, const cvec_t& elem, bool redo = false);

    rc_t replace_fence_rec_nolog_may_defrag(const w_keystr_t& low,
                                            const w_keystr_t& high, const w_keystr_t& chain,
                                            int new_prefix_length = -1);

    rc_t replace_fence_rec_nolog_no_defrag(const w_keystr_t& low,
                                           const w_keystr_t& high, const w_keystr_t& chain, int new_prefix_length);

    rc_t remove_shift_nolog(slotid_t slot);

    rc_t replace_el_nolog(slotid_t slot, const cvec_t& elem);

    void overwrite_el_nolog(slotid_t slot, smsize_t offset,
                            const char* new_el, smsize_t elen);

    void reserve_ghost(const w_keystr_t& key, size_t element_length) {
        reserve_ghost((const char*)key.buffer_as_keystr(), key.get_length_as_keystr(), element_length);
    }

    // to make it slightly faster.  not a neat kind of optimization
    void reserve_ghost(const char* key_raw, size_t key_raw_len, size_t element_length);

    void insert_nonghost(const w_keystr_t& key, const cvec_t& elem);

    bool _is_enough_spacious_ghost(
            const w_keystr_t& key, slotid_t slot,
            const cvec_t& el);

    bool check_space_for_insert_leaf(const w_keystr_t& trunc_key, const cvec_t& el);

    bool check_space_for_insert_leaf(size_t trunc_key_length, size_t element_length);

    bool check_space_for_insert_node(const w_keystr_t& key);

    void suggest_fence_for_split(
            w_keystr_t& mid, slotid_t& right_begins_from, const w_keystr_t& triggering_key) const;

    w_keystr_t recalculate_fence_for_split(slotid_t right_begins_from) const;

    bool is_insertion_extremely_skewed_right() const;

    bool is_insertion_skewed_right() const;

    bool is_insertion_skewed_left() const;


    // ======================================================================
    //   BEGIN: Statistics/Debug etc functions
    // ======================================================================

    rc_t defrag(const bool full_logging_redo = false);

    rc_t compress(const w_keystr_t& low, const w_keystr_t& high,
                  const w_keystr_t& chain, bool redo = false);

    rc_t leaf_stats(btree_lf_stats_t& btree_lf);

    rc_t int_stats(btree_int_stats_t& btree_int);

    void print(bool print_elem = false);

    bool is_consistent(bool check_keyorder = false, bool check_space = false) const;

    lsn_t* emlsn_address(general_recordid_t pos);

    const lsn_t& get_emlsn_general(general_recordid_t pos) const;

    void set_emlsn_general(general_recordid_t pos, const lsn_t& lsn);

    const lsn_t& get_foster_emlsn() const;

    const lsn_t& get_pid0_emlsn() const;

    /*
     * The combined sizes of the key (i.e., the number of actual data
     * bytes it contains) and value must be less than or equal to \ref
     * max_entry_size, which is a function of the page size, and is
     * such that two entries of this size fit on a page along with all
     * the page and entry metadata.  See sm_config_info_t and
     * ss_m::config_info.
     */
    static smsize_t max_entry_size;

private:
    // ======================================================================
    //   BEGIN: Private record data packers
    // ======================================================================

    typedef uint16_t key_length_t;

    enum _internal {
        max_key_length = (1 << (sizeof(key_length_t) * 8 - 1)) - 1,
    };

    void _pack_node_record(cvec_t& out, const cvec_t& trunc_key, const lsn_t& emlsn) const;

    typedef key_length_t pack_scratch_t;

    void _pack_leaf_record(cvec_t& out, pack_scratch_t& out_scratch,
                           const cvec_t& trunc_key,
                           const char* element, size_t element_len) const;

    void _pack_leaf_record_prefix(cvec_t& out, pack_scratch_t& out_scratch,
                                  const cvec_t& trunc_key) const;

    int _pack_fence_rec(cvec_t& out, const w_keystr_t& low,
                        const w_keystr_t& high,
                        const w_keystr_t& chain,
                        int new_prefix_len) const;

    size_t _predict_leaf_data_length(int trunc_key_length, int element_length) const;

    typedef uint16_t poor_man_key;

    poor_man_key _extract_poor_man_key(const void* trunc_key, size_t trunc_key_len) const;

    poor_man_key _extract_poor_man_key(const cvec_t& trunc_key) const;

    poor_man_key _extract_poor_man_key(const void* key_with_prefix,
                                       size_t key_len_with_prefix, size_t prefix_len) const;


    // ======================================================================
    //   BEGIN: Private record accessors
    // ======================================================================

    poor_man_key _poor(int slot) const;

    const char* _leaf_key_noprefix(slotid_t slot, size_t& len) const;

    const char* _node_key_noprefix(slotid_t slot, size_t& len) const;

    size_t _element_offset(int slot) const;

    int _compare_key_noprefix(slotid_t slot, const void* key_noprefix, size_t key_len) const;

    int _compare_slot_with_key(int slot, const void* key_noprefix, size_t key_len, poor_man_key poor) const;

    // ======================================================================
    //   BEGIN: Miscellaneous private members
    // ======================================================================

    bool _is_consistent_keyorder() const;

    bool _is_consistent_poormankey() const;

    void _update_btree_consecutive_skewed_insertions(slotid_t slot);

    bool _check_space_for_insert(size_t data_length);

    void _init(lsn_t lsn, PageID page_id, StoreID store,
               PageID root_pid, PageID pid0, lsn_t pid0_emlsn,
               PageID foster_pid, lsn_t foster_emlsn, int16_t btree_level,
               const w_keystr_t& low, const w_keystr_t& high,
               const w_keystr_t& chain_fence_high, const bool ghost);

public:
    friend std::ostream& operator<<(std::ostream& os, btree_page_h& b);
};

class borrowed_btree_page_h : public btree_page_h {
    fixable_page_h* _source;

public:
    borrowed_btree_page_h(fixable_page_h* source) :
            btree_page_h(),
            _source(source) {
        w_assert1(_source->_pp != nullptr);
        _pp = _source->get_generic_page();
        _bufferpool_managed = _source->_bufferpool_managed;
        _mode = _source->_mode;
        _Q_ticket = _source->_Q_ticket;
        _source->_mode = LATCH_NL;
    }

    ~borrowed_btree_page_h() {
        w_assert1(_source->_pp == _pp);
        w_assert1(_source->_mode == LATCH_NL);
        _source->_mode = _mode;
        _source->_Q_ticket = _Q_ticket;

        // prevent superclass destructor from trying to unfix us in the bufferpool:
        _bufferpool_managed = false;
        _mode = LATCH_NL;
    }
};

struct btree_lf_stats_t {
    base_stat_t hdr_bs;            /* page header (overhead) */
    base_stat_t key_bs;            /* space used for keys      */
    base_stat_t data_bs;           /* space for data associated to keys */
    base_stat_t entry_overhead_bs; /* e.g., slot header, padding for alignment */
    base_stat_t unused_bs;

    base_stat_t entry_cnt;

    base_stat_t unique_cnt;        /* number of unique entries */

    btree_lf_stats_t() {
        clear();
    }

    void add(const btree_lf_stats_t& stats);

    void clear();

    w_rc_t audit() const;

    base_stat_t total_bytes() const;

    void print(ostream&, const char*) const;/* pretty print */
    friend ostream& operator<<(ostream&, const btree_lf_stats_t& s);
};

struct btree_int_stats_t {
    base_stat_t used_bs;

    base_stat_t unused_bs;

    btree_int_stats_t() {
        clear();
    }

    void add(const btree_int_stats_t& stats);

    void clear();

    w_rc_t audit() const;

    base_stat_t total_bytes() const;

    void print(ostream&, const char*) const;/* pretty print */
    friend ostream& operator<<(ostream&, const btree_int_stats_t& s);
};

// returns by bulk load and
// btree_m::get_du_statistics
struct btree_stats_t {
    btree_lf_stats_t leaf_pg;  // byte counts for leaf pages
    btree_int_stats_t int_pg;   // byte counts for interior pages

    base_stat_t leaf_pg_cnt; // level-1 pages found by tree traversal
    base_stat_t int_pg_cnt;  // level>1 pages found by tree traversal
    base_stat_t unlink_pg_cnt; // pages allocated but not accounted-for
    // in a tree traversal; allocated count is
    // found by first_page/next_page traversal
    // of the store's extents
    // Given the way this is computed, it would be
    // pretty hard to fail an audit.
    // It would be nice if unlinked pages could be
    // accounted-for in another way for a meaningful audit.
    /* unlinked pages are empty and
   // will be freed the next
   // time they are encountered
   // during a traversal
   */
    base_stat_t unalloc_pg_cnt; // pages not-allocated by extent-traversal
    base_stat_t level_cnt;    /* number of levels in btree */

    btree_stats_t() {
        clear();
    }

    void add(const btree_stats_t& stats);

    void clear();

    w_rc_t audit() const;

    base_stat_t total_bytes() const;

    base_stat_t alloc_pg_cnt() const;

    void print(ostream&, const char*) const;/* pretty print */
    friend ostream& operator<<(ostream&, const btree_stats_t& s);
};

// ======================================================================
//   BEGIN: Inline function implementations
// ======================================================================

inline PageID btree_page_h::root() const {
    return page()->btree_root;
}

inline int btree_page_h::level() const {
    return page()->btree_level;
}

inline PageID btree_page_h::pid0_opaqueptr() const {
    return page()->btree_pid0;
}

inline bool btree_page_h::is_leaf() const {
    return level() == 1;
}

inline bool btree_page_h::is_leaf_parent() const {
    return level() == 2;
}

inline bool btree_page_h::is_node() const {
    return !is_leaf();
}

inline PageID btree_page_h::get_foster_opaqueptr() const {
    return page()->btree_foster;
}

inline int16_t btree_page_h::get_prefix_length() const {
    return page()->btree_prefix_length;
}

inline int16_t btree_page_h::get_fence_low_length() const {
    return page()->btree_fence_low_length;
}

inline int16_t btree_page_h::get_fence_high_length() const {
    return page()->btree_fence_high_length;
}

inline int16_t btree_page_h::get_chain_fence_high_length() const {
    return page()->btree_chain_fence_high_length;
}

inline const char* btree_page_h::get_fence_low_key() const {
    return page()->item_data(0);
}

inline const char* btree_page_h::get_fence_high_key_noprefix() const {
    return page()->item_data(0) + get_fence_low_length();
}

inline const char* btree_page_h::get_chain_fence_high_key() const {
    return page()->item_data(0) + get_fence_low_length() + get_fence_high_length_noprefix();
}

inline const char* btree_page_h::get_prefix_key() const {
    return get_fence_low_key(); // same thing.  only the length differs.
}

inline int btree_page_h::nrecs() const {
    return page()->number_of_items() - 1;
}

inline int btree_page_h::nghosts() const {
    return page()->number_of_ghosts();
}

inline int btree_page_h::compare_with_fence_low(const w_keystr_t& key) const {
    return key.compare_keystr(get_fence_low_key(), get_fence_low_length());
}

inline int btree_page_h::compare_with_fence_low(const char* key, size_t key_len) const {
    return w_keystr_t::compare_bin_str(key, key_len, get_fence_low_key(), get_fence_low_length());
}

inline int btree_page_h::compare_with_fence_low_noprefix(const char* key, size_t key_len) const {
    return w_keystr_t::compare_bin_str(key, key_len, get_fence_low_key() + get_prefix_length(),
                                       get_fence_low_length() - get_prefix_length());
}

inline int btree_page_h::compare_with_fence_high(const w_keystr_t& key) const {
    return compare_with_fence_high((const char*)key.buffer_as_keystr(), key.get_length_as_keystr());
}

inline int btree_page_h::compare_with_fence_high(const char* key, size_t key_len) const {
    size_t prefix_len = get_prefix_length();
    if (prefix_len > key_len) {
        return w_keystr_t::compare_bin_str(key, key_len, get_prefix_key(), key_len);
    } else {
        // first, compare with prefix part
        int ret = w_keystr_t::compare_bin_str(key, prefix_len, get_prefix_key(), prefix_len);
        if (ret != 0) {
            return ret;
        }
        // then, compare with suffix part
        return w_keystr_t::compare_bin_str(key + prefix_len, key_len - prefix_len,
                                           get_fence_high_key_noprefix(), get_fence_high_length_noprefix());
    }
}

inline int btree_page_h::compare_with_fence_high_noprefix(const char* key, size_t key_len) const {
    return w_keystr_t::compare_bin_str(key, key_len, get_fence_high_key_noprefix(), get_fence_high_length_noprefix());
}

inline int btree_page_h::compare_with_chain_fence_high(const w_keystr_t& key) const {
    return key.compare_keystr(get_chain_fence_high_key(), get_chain_fence_high_length());
}

inline int btree_page_h::compare_with_chain_fence_high(const char* key, size_t key_len) const {
    return w_keystr_t::compare_bin_str(key, key_len, get_chain_fence_high_key(), get_chain_fence_high_length());
}

inline bool btree_page_h::fence_contains(const w_keystr_t& key) const {
    // fence-low is inclusive
    if (compare_with_fence_low(key) < 0) {
        return false;
    }
    // fence-high is exclusive  (but if it's supremum, we allow it to handle to-the-end scan)
    if (!is_fence_high_supremum() && compare_with_fence_high(key) >= 0) {
        return false;
    }
    return true;
}

inline bool btree_page_h::is_insertion_extremely_skewed_right() const {
    // this means completely pre-sorted insertion like bulk loading.
    int ins = page()->btree_consecutive_skewed_insertions;
    return ins > 50
           || ins > nrecs() * 9 / 10
           || (ins > 1 && ins >= nrecs() - 1);
}

inline bool btree_page_h::is_insertion_skewed_right() const {
    return page()->btree_consecutive_skewed_insertions > 5;
}

inline bool btree_page_h::is_insertion_skewed_left() const {
    return page()->btree_consecutive_skewed_insertions < -5;
}

inline PageID btree_page_h::child_opaqueptr(slotid_t slot) const {
    w_assert1(is_node());
    w_assert1(slot >= 0);
    w_assert1(slot < nrecs());
    return page()->item_child(slot + 1);
}

inline PageID* btree_page_h::page_pointer_address(int offset) {
    if (offset == -2) {
        return &page()->btree_foster;
    }

    w_assert1(!is_leaf());
    w_assert1(-2 < offset && offset < nrecs());

    if (offset == -1) {
        return &page()->btree_pid0;
    }

    // CS TODO: why +1 here? what happens when offset == 0?
    return &page()->item_child(offset + 1);
}

inline size_t btree_page_h::get_rec_space(int slot) const {
    w_assert1(slot >= 0);
    return page()->item_space(slot + 1);
}

inline bool btree_page_h::is_ghost(slotid_t slot) const {
    return page()->is_ghost(slot + 1);
}

inline smsize_t
btree_page_h::used_space() const {
    return data_sz - page()->usable_space();
}

inline smsize_t
btree_page_h::usable_space() const {
    return page()->usable_space();
}


// ======================================================================
//   BEGIN: Private record data packers inline implementation
// ======================================================================

inline void btree_page_h::_pack_node_record(cvec_t& out, const cvec_t& trunc_key,
                                            const lsn_t& emlsn) const {
    w_assert1(!is_leaf());
    out.put(trunc_key);
    out.put(&emlsn, sizeof(lsn_t));
}

inline void btree_page_h::_pack_leaf_record_prefix(cvec_t& out, pack_scratch_t& out_scratch,
                                                   const cvec_t& trunc_key) const {
    w_assert1(is_leaf());
    out_scratch = trunc_key.size();
    out.put(&out_scratch, sizeof(out_scratch));
    out.put(trunc_key);
}

inline void btree_page_h::_pack_leaf_record(cvec_t& out, pack_scratch_t& out_scratch,
                                            const cvec_t& trunc_key,
                                            const char* element, size_t element_len) const {
    _pack_leaf_record_prefix(out, out_scratch, trunc_key);
    out.put(element, element_len);
}

inline int btree_page_h::_pack_fence_rec(cvec_t& out, const w_keystr_t& low,
                                         const w_keystr_t& high,
                                         const w_keystr_t& chain,
                                         int new_prefix_len) const {
    int prefix_len;
    if (new_prefix_len >= 0) {
        w_assert1(low.common_leading_bytes(high) >= (size_t)new_prefix_len);
        prefix_len = new_prefix_len;
    } else {
        prefix_len = low.common_leading_bytes(high);
    }

    out.put(low);
    // eliminate prefix part from high:
    out.put((const char*)high.buffer_as_keystr() + prefix_len,
            high.get_length_as_keystr() - prefix_len);
    if (!chain.is_constructed()) {
        // If chain-high is not set, we don't need to put it here. However,
        // we will at least need fench-high length of capacity when we split this page.
        // As running out of space even for splitting is a severe issue, we pre-allocate
        // a fence-high length of capacity here.
        out.put(high);
    } else {
        out.put(chain);
        if (chain.get_length_as_keystr() < high.get_length_as_keystr()) {
            // for the same purpose, put additional bytes
            out.put(high.buffer_as_keystr(),
                    high.get_length_as_keystr() - chain.get_length_as_keystr());
        }
    }
    return prefix_len;
}

inline size_t btree_page_h::_predict_leaf_data_length(int trunc_key_length,
                                                      int element_length) const {
    return sizeof(key_length_t) + trunc_key_length + element_length;
}

inline btree_page_h::poor_man_key btree_page_h::_extract_poor_man_key(const void* trunc_key,
                                                                      size_t trunc_key_len) const {
    if (trunc_key_len == 0) {
        return 0;
    } else if (trunc_key_len == 1) {
        return (*reinterpret_cast<const unsigned char*>(trunc_key)) << 8;
    } else {
        // convert big-endian array (usable with memcmp) into 16-bit integer (little-endian)
        return deserialize16_ho(trunc_key);
    }
}

inline btree_page_h::poor_man_key btree_page_h::_extract_poor_man_key(const cvec_t& trunc_key) const {
    char start[2];
    trunc_key.copy_to(start, 2);
    return _extract_poor_man_key(start, trunc_key.size());
}

inline btree_page_h::poor_man_key
btree_page_h::_extract_poor_man_key(const void* key_with_prefix, size_t key_len_with_prefix,
                                    size_t prefix_len) const {
    w_assert1(prefix_len <= key_len_with_prefix);
    return _extract_poor_man_key(((const char*)key_with_prefix) + prefix_len, key_len_with_prefix - prefix_len);
}


// ======================================================================
//   BEGIN: Private [robust] record accessors inline implementation
// ======================================================================

inline btree_page_h::poor_man_key btree_page_h::_poor(int slot) const {
    w_assert1(slot >= 0);
    return page()->item_poor(slot + 1);
}

inline const char* btree_page_h::_leaf_key_noprefix(slotid_t slot, size_t& len) const {
    w_assert1(is_leaf());
    w_assert1(slot >= 0);

    key_length_t* data = (key_length_t*)page()->item_data(slot + 1);
    len = *data++;
    return (const char*)data;
}

inline const char* btree_page_h::_node_key_noprefix(slotid_t slot, size_t& len) const {
    w_assert1(is_node());
    w_assert1(slot >= 0);

    len = page()->item_length(slot + 1) - sizeof(lsn_t);
    return page()->item_data(slot + 1);
}

inline size_t btree_page_h::_element_offset(int slot) const {
    w_assert1(is_leaf());
    w_assert1(slot >= 0);

    size_t key_noprefix_length;
    (void)_leaf_key_noprefix(slot, key_noprefix_length);

    return key_noprefix_length + sizeof(key_length_t);
}

inline int btree_page_h::_compare_key_noprefix(slotid_t slot, const void* key_noprefix,
                                               size_t key_len) const {
    size_t curkey_len;
    const char* curkey;
    if (is_leaf()) {
        curkey = _leaf_key_noprefix(slot, curkey_len);
    } else {
        curkey = _node_key_noprefix(slot, curkey_len);
    }

    return w_keystr_t::compare_bin_str(curkey, curkey_len, key_noprefix, key_len);
}

// ======================================================================
//   BEGIN: Single-Page-Recovery related EMLSN accessors implementation
// ======================================================================

inline lsn_t* btree_page_h::emlsn_address(general_recordid_t pos) {
    if (pos == GeneralRecordIds::PID0) {
        return &page()->btree_pid0_emlsn;
    } else if (pos == GeneralRecordIds::FOSTER_CHILD) {
        // this means foster child
        return &page()->btree_foster_emlsn;
    } else {
        // last 8 bytes after usual item_data is the child EMLSN.
        // Note: these are "pos", not "pos+1" because it's general_recordid_t, not slotid_t
        char* data = page()->item_data(pos);
        size_t len = page()->item_length(pos);
        w_assert1(len >= sizeof(lsn_t));
        len -= sizeof(lsn_t);
        return reinterpret_cast<lsn_t*>(data + len);
    }
}

inline const lsn_t& btree_page_h::get_emlsn_general(general_recordid_t pos) const {
    return *(const_cast<btree_page_h*>(this)->emlsn_address(pos));
}

inline void btree_page_h::set_emlsn_general(general_recordid_t pos, const lsn_t& lsn) {
    *emlsn_address(pos) = lsn;
}

inline const lsn_t& btree_page_h::get_foster_emlsn() const {
    return page()->btree_foster_emlsn;
}

inline const lsn_t& btree_page_h::get_pid0_emlsn() const {
    return page()->btree_pid0_emlsn;
}

#endif // __BTREE_PAGE_H_H