glass_table.h

Go to the documentation of this file.

/* Copyright 1999,2000,2001 BrightStation PLC
 * Copyright 2002,2003,2004,2005,2006,2007,2008,2009,2010,2012,2013,2014,2015,2016 Olly Betts
 * Copyright 2008 Lemur Consulting Ltd
 *
 * This program is free software; you can redistribute it and/or
 * modify it under the terms of the GNU General Public License as
 * published by the Free Software Foundation; either version 2 of the
 * License, or (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with this program; if not, write to the Free Software
 * Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA  02110-1301
 * USA
 */

#ifndef OM_HGUARD_GLASS_TABLE_H
#define OM_HGUARD_GLASS_TABLE_H

#include <xapian/constants.h>
#include <xapian/error.h>

#include "glass_freelist.h"
#include "glass_cursor.h"
#include "glass_defs.h"

#include "io_utils.h"
#include "noreturn.h"
#include "omassert.h"
#include "str.h"
#include "stringutils.h"
#include "wordaccess.h"

#include "common/compression_stream.h"

#include <algorithm>
#include <string>

namespace Glass {

const size_t BLOCK_CAPACITY = 4;

#define GLASS_BTREE_MAX_KEY_LEN 255

// FIXME: This named constant probably isn't used everywhere it should be...
const int BYTES_PER_BLOCK_NUMBER = 4;

/*  The B-tree blocks have a number of internal lengths and offsets held in 1, 2
    or 4 bytes. To make the coding a little clearer,
       we use  for
       ------  ---
       K1      the 1 byte length of key
       I2      the 2 byte length of an item (key-tag pair)
       D2      the 2 byte offset to the item from the directory
       X2      the 2 byte component counter that ends each key
*/

const int K1 = 1;
const int I2 = 2;
const int D2 = 2;
const int X2 = 2;

/*  and when getting or setting them, we use these methods of the various
 *  *Item* classes: */

// getD(p, c)
// setD(p, c, x)
// getI()
// setI(x)
// getX(p, c)
// setX(p, c, x)

/* if you've been reading the comments from the top, the next four procedures
   will not cause any headaches.

   Recall that a leaf item has this form:

           i k     x
           | |     |
           I K key X tag
               ←K→
           <---SIZE---->
               <---I--->

   Except that X is omitted for the first component of a tag (there is a flag
   bit in the upper bits of I which indicates these).

   item_of(p, c) returns i, the address of the item at block address p,
   directory offset c,

   component_of(p, c) returns the number marked 'x' above,

   last_component(p, c) returns true if this is a final component.
*/

inline uint4 REVISION(const uint8_t * b) { return aligned_read4(b); }
inline int GET_LEVEL(const uint8_t * b) { return b[4]; }
inline int MAX_FREE(const uint8_t * b) { return unaligned_read2(b + 5); }
inline int TOTAL_FREE(const uint8_t * b) { return unaligned_read2(b + 7); }
inline int DIR_END(const uint8_t * b) { return unaligned_read2(b + 9); }
const int DIR_START = 11;

inline void SET_REVISION(uint8_t * b, uint4 rev) { aligned_write4(b, rev); }
inline void SET_LEVEL(uint8_t * b, int x) { AssertRel(x,<,256); b[4] = x; }
inline void SET_MAX_FREE(uint8_t * b, int x) { unaligned_write2(b + 5, x); }
inline void SET_TOTAL_FREE(uint8_t * b, int x) { unaligned_write2(b + 7, x); }
inline void SET_DIR_END(uint8_t * b, int x) { unaligned_write2(b + 9, x); }

// The item size is stored in 2 bytes, but the top bit is used to store a flag for
// "is the tag data compressed" and the next two bits are used to flag if this is the
// first and/or last item for this tag.
const int I_COMPRESSED_BIT = 0x80;
const int I_LAST_BIT = 0x40;
const int I_FIRST_BIT = 0x20;

const int I_MASK = (I_COMPRESSED_BIT|I_LAST_BIT|I_FIRST_BIT);

const int ITEM_SIZE_MASK = (0xffff &~ (I_MASK << 8));
const size_t MAX_ITEM_SIZE = (ITEM_SIZE_MASK + 3);

const int LEVEL_FREELIST = 254;

class RootInfo;

class Key {
    const uint8_t *p;
  public:
    explicit Key(const uint8_t * p_) : p(p_) { }
    const uint8_t * get_address() const { return p; }
    const uint8_t * data() const { return p + K1; }
    void read(std::string * key) const {
        key->assign(reinterpret_cast<const char *>(p + K1), length());
    }
    int length() const {
        return p[0];
    }
    char operator[](size_t i) const {
        AssertRel(i,<,size_t(length()));
        return p[i + K1];
    }
};

// LeafItem_wr wants to be "LeafItem with non-const p and more methods" - we can't
// achieve that nicely with inheritance, so we use a template base class.
template<class T> class LeafItem_base {
  protected:
    T p;
    int get_key_len() const { return p[I2]; }
    static int getD(const uint8_t * q, int c) {
        AssertRel(c, >=, DIR_START);
        AssertRel(c, <, 65535);
        Assert((c & 1) == 1);
        return unaligned_read2(q + c);
    }
    int getI() const { return unaligned_read2(p); }
    static int getX(const uint8_t * q, int c) { return unaligned_read2(q + c); }
  public:
    /* LeafItem from block address and offset to item pointer */
    LeafItem_base(T p_, int c) : p(p_ + getD(p_, c)) { }
    explicit LeafItem_base(T p_) : p(p_) { }
    T get_address() const { return p; }
    int size() const {
        return (getI() & ITEM_SIZE_MASK) + 3;
    }
    bool get_compressed() const { return *p & I_COMPRESSED_BIT; }
    bool first_component() const { return *p & I_FIRST_BIT; }
    bool last_component() const { return *p & I_LAST_BIT; }
    int component_of() const {
        if (first_component()) return 1;
        return getX(p, get_key_len() + I2 + K1);
    }
    Key key() const { return Key(p + I2); }
    void append_chunk(std::string * tag) const {
        // Offset to the start of the tag data.
        int cd = get_key_len() + I2 + K1;
        if (!first_component()) cd += X2;
        // Number of bytes to extract from current component.
        int l = size() - cd;
        const char * chunk = reinterpret_cast<const char *>(p + cd);
        tag->append(chunk, l);
    }
    bool decompress_chunk(CompressionStream& comp_stream, string& tag) const {
        // Offset to the start of the tag data.
        int cd = get_key_len() + I2 + K1;
        if (!first_component()) cd += X2;
        // Number of bytes to extract from current component.
        int l = size() - cd;
        const char * chunk = reinterpret_cast<const char *>(p + cd);
        return comp_stream.decompress_chunk(chunk, l, tag);
    }
};

class LeafItem : public LeafItem_base<const uint8_t *> {
  public:
    /* LeafItem from block address and offset to item pointer */
    LeafItem(const uint8_t * p_, int c)
        : LeafItem_base<const uint8_t *>(p_, c) { }
    explicit LeafItem(const uint8_t * p_)
        : LeafItem_base<const uint8_t *>(p_) { }
};

class LeafItem_wr : public LeafItem_base<uint8_t *> {
    void set_key_len(int x) {
        AssertRel(x, >=, 0);
        AssertRel(x, <=, GLASS_BTREE_MAX_KEY_LEN);
        p[I2] = x;
    }
    void setI(int x) { unaligned_write2(p, x); }
    static void setX(uint8_t * q, int c, int x) { unaligned_write2(q + c, x); }
  public:
    /* LeafItem_wr from block address and offset to item pointer */
    LeafItem_wr(uint8_t * p_, int c) : LeafItem_base<uint8_t *>(p_, c) { }
    explicit LeafItem_wr(uint8_t * p_) : LeafItem_base<uint8_t *>(p_) { }
    void set_component_of(int i) {
        AssertRel(i,>,1);
        *p &=~ I_FIRST_BIT;
        setX(p, get_key_len() + I2 + K1, i);
    }
    void set_size(int new_size) {
        AssertRel(new_size,>=,3);
        int I = new_size - 3;
        // We should never be able to pass too large a size here, but don't
        // corrupt the database if this somehow happens.
        if (rare(I &~ ITEM_SIZE_MASK)) throw Xapian::DatabaseError("item too large!");
        setI(I);
    }
    void form_key(const std::string & key_) {
        std::string::size_type key_len = key_.length();
        if (key_len > GLASS_BTREE_MAX_KEY_LEN) {
            // We check term length when a term is added to a document but
            // glass doubles zero bytes, so this can still happen for terms
            // which contain one or more zero bytes.
            std::string msg("Key too long: length was ");
            msg += str(key_len);
            msg += " bytes, maximum length of a key is "
                   STRINGIZE(GLASS_BTREE_MAX_KEY_LEN) " bytes";
            throw Xapian::InvalidArgumentError(msg);
        }

        set_key_len(key_len);
        std::memmove(p + I2 + K1, key_.data(), key_len);
        *p |= I_FIRST_BIT;
    }
    // FIXME passing cd here is icky
    void set_tag(int cd, const char *start, int len, bool compressed, int i, int m) {
        std::memmove(p + cd, start, len);
        set_size(cd + len);
        if (compressed) *p |= I_COMPRESSED_BIT;
        if (i == m) *p |= I_LAST_BIT;
        if (i == 1) {
            *p |= I_FIRST_BIT;
        } else {
            set_component_of(i);
        }
    }
    void fake_root_item() {
        set_key_len(0);   // null key length
        set_size(I2 + K1);   // length of the item
        *p |= I_FIRST_BIT|I_LAST_BIT;
    }
    operator const LeafItem() const { return LeafItem(p); }
    static void setD(uint8_t * q, int c, int x) {
        AssertRel(c, >=, DIR_START);
        AssertRel(c, <, 65535);
        Assert((c & 1) == 1);
        unaligned_write2(q + c, x);
    }
};

/* A branch item has this form:

                 k     x
                 |     |
             tag K key X
             ←B→   ←K→
             <--SIZE--->

             B = BYTES_PER_BLOCK_NUMBER

   We can't omit X here, as we've nowhere to store the first and last bit
   flags which we have in leaf items.
*/

// BItem_wr wants to be "BItem with non-const p and more methods" - we can't
// achieve that nicely with inheritance, so we use a template base class.
template<class T> class BItem_base {
  protected:
    T p;
    int get_key_len() const { return p[BYTES_PER_BLOCK_NUMBER]; }
    static int getD(const uint8_t * q, int c) {
        AssertRel(c, >=, DIR_START);
        AssertRel(c, <, 65535);
        Assert((c & 1) == 1);
        return unaligned_read2(q + c);
    }
    static int getX(const uint8_t * q, int c) { return unaligned_read2(q + c); }
  public:
    /* BItem from block address and offset to item pointer */
    BItem_base(T p_, int c) : p(p_ + getD(p_, c)) { }
    explicit BItem_base(T p_) : p(p_) { }
    T get_address() const { return p; }
    int size() const {
        return get_key_len() + K1 + X2 + BYTES_PER_BLOCK_NUMBER;
    }
    Key key() const { return Key(p + BYTES_PER_BLOCK_NUMBER); }
    uint4 block_given_by() const {
        return unaligned_read4(p);
    }
    int component_of() const {
        return getX(p, get_key_len() + BYTES_PER_BLOCK_NUMBER + K1);
    }
};

class BItem : public BItem_base<const uint8_t *> {
  public:
    /* BItem from block address and offset to item pointer */
    BItem(const uint8_t * p_, int c) : BItem_base<const uint8_t *>(p_, c) { }
    explicit BItem(const uint8_t * p_) : BItem_base<const uint8_t *>(p_) { }
};

class BItem_wr : public BItem_base<uint8_t *> {
    void set_key_len(int x) {
        AssertRel(x, >=, 0);
        AssertRel(x, <, GLASS_BTREE_MAX_KEY_LEN);
        p[BYTES_PER_BLOCK_NUMBER] = x;
    }
    static void setX(uint8_t * q, int c, int x) { unaligned_write2(q + c, x); }
  public:
    /* BItem_wr from block address and offset to item pointer */
    BItem_wr(uint8_t * p_, int c) : BItem_base<uint8_t *>(p_, c) { }
    explicit BItem_wr(uint8_t * p_) : BItem_base<uint8_t *>(p_) { }
    void set_component_of(int i) {
        setX(p, get_key_len() + BYTES_PER_BLOCK_NUMBER + K1, i);
    }
    void set_key_and_block(Key newkey, uint4 n) {
        int len = newkey.length() + K1 + X2;
        // Copy the key size, main part of the key and the count part.
        std::memcpy(p + BYTES_PER_BLOCK_NUMBER, newkey.get_address(), len);
        // Set tag contents to block number
        set_block_given_by(n);
    }
    // Takes size as we may be truncating newkey.
    void set_truncated_key_and_block(Key newkey, int new_comp,
                                     int truncate_size, uint4 n) {
        int i = truncate_size;
        AssertRel(i,<=,newkey.length());
        // Key size
        set_key_len(i);
        // Copy the main part of the key, possibly truncating.
        std::memcpy(p + BYTES_PER_BLOCK_NUMBER + K1, newkey.get_address() + K1, i);
        // Set the component count.
        setX(p, BYTES_PER_BLOCK_NUMBER + K1 + i, new_comp);
        // Set tag contents to block number
        set_block_given_by(n);
    }

    void set_block_given_by(uint4 n) {
        unaligned_write4(p, n);
    }
    void form_null_key(uint4 n) {
        set_block_given_by(n);
        set_key_len(0);        /* null key */
        set_component_of(0);
    }
    operator const BItem() const { return BItem(p); }
    static void setD(uint8_t * q, int c, int x) {
        AssertRel(c, >=, DIR_START);
        AssertRel(c, <, 65535);
        Assert((c & 1) == 1);
        unaligned_write2(q + c, x);
    }
};

}

using Glass::RootInfo;

class GlassChanges;

class GlassTable {
    friend class GlassCursor; /* Should probably fix this. */
    friend class GlassFreeList;

  private:
    GlassTable(const GlassTable &);

    GlassTable & operator=(const GlassTable &);

    void basic_open(const RootInfo * root_info,
                    glass_revision_number_t rev);

    void do_open_to_read(const RootInfo * root_info,
                         glass_revision_number_t rev);

    void do_open_to_write(const RootInfo * root_info,
                          glass_revision_number_t rev = 0);

  public:
    GlassTable(const char * tablename_, const std::string & path_,
               bool readonly_, bool lazy = false);

    GlassTable(const char * tablename_, int fd, off_t offset_,
               bool readonly_, bool lazy = false);

    ~GlassTable();

    void close(bool permanent = false);

    bool readahead_key(const string &key) const;

    bool exists() const;

    void open(int flags_, const RootInfo & root_info,
              glass_revision_number_t rev);

    bool is_open() const { return handle >= 0; }

    bool is_writable() const { return writable; }

    void flush_db();

    void commit(glass_revision_number_t revision, RootInfo * root_info);

    bool sync() {
        return (flags & Xapian::DB_NO_SYNC) ||
               handle < 0 ||
               io_sync(handle);
    }

    void cancel(const RootInfo & root_info, glass_revision_number_t rev);

    bool get_exact_entry(const std::string & key, std::string & tag) const;

    bool key_exists(const std::string &key) const;

    bool read_tag(Glass::Cursor* C_,
                  std::string* tag,
                  bool keep_compressed) const;

    void add(const std::string& key,
             const std::string& tag,
             bool already_compressed = false);

    bool del(const std::string &key);

    int get_flags() const { return flags; }

    void create_and_open(int flags_, const RootInfo & root_info);

    void set_full_compaction(bool parity);

    glass_revision_number_t get_open_revision_number() const {
        return revision_number;
    }

    glass_tablesize_t get_entry_count() const {
        return item_count;
    }

    bool empty() const {
        return (item_count == 0);
    }

    GlassCursor * cursor_get() const;

    bool is_modified() const { return Btree_modified; }

    void set_max_item_size(size_t block_capacity) {
        if (block_capacity > Glass::BLOCK_CAPACITY)
            block_capacity = Glass::BLOCK_CAPACITY;
        using Glass::DIR_START;
        using Glass::D2;
        max_item_size =
            (block_size - DIR_START - block_capacity * D2) / block_capacity;
        // Make sure we don't exceed the limit imposed by the format.
        if (max_item_size > Glass::MAX_ITEM_SIZE)
            max_item_size = Glass::MAX_ITEM_SIZE;
    }

    void set_changes(GlassChanges * changes) {
        changes_obj = changes;
    }

    XAPIAN_NORETURN(static void throw_database_closed());

    string get_path() const {
        return name + GLASS_TABLE_EXTENSION;
    }

  protected:
    bool find(Glass::Cursor *) const;
    int delete_kt();
    void read_block(uint4 n, uint8_t *p) const;
    void write_block(uint4 n, const uint8_t *p,
                     bool appending = false) const;
    XAPIAN_NORETURN(void set_overwritten() const);
    void block_to_cursor(Glass::Cursor *C_, int j, uint4 n) const;
    void alter();
    void compact(uint8_t *p);
    void enter_key_above_leaf(Glass::LeafItem previtem,
                              Glass::LeafItem newitem);
    void enter_key_above_branch(int j, Glass::BItem newitem);
    int mid_point(uint8_t *p) const;
    void add_item_to_leaf(uint8_t *p, Glass::LeafItem kt, int c);
    void add_item_to_branch(uint8_t *p, Glass::BItem kt, int c);
    void add_leaf_item(Glass::LeafItem kt);
    void add_branch_item(Glass::BItem kt, int j);
    void delete_leaf_item(bool repeatedly);
    void delete_branch_item(int j);
    int add_kt(bool found);
    void read_root();
    void split_root(uint4 split_n);
    void form_key(const std::string & key) const;

    const char * tablename;

    glass_revision_number_t revision_number;

    glass_tablesize_t item_count;

    unsigned int block_size;

    int flags;

    bool faked_root_block;

    bool sequential;

    int handle;

    int level;

    uint4 root;

    mutable Glass::LeafItem_wr kt;

    uint8_t * buffer;

    GlassFreeList free_list;

    std::string name;

    int seq_count;

    uint4 changed_n;

    int changed_c;

    size_t max_item_size;

    mutable bool Btree_modified;

    bool full_compaction;

    bool writable;

    mutable bool cursor_created_since_last_modification;

    unsigned long cursor_version;

    GlassChanges * changes_obj;

    bool single_file() const {
        return name.empty();
    }

    /* B-tree navigation functions */
    bool prev(Glass::Cursor *C_, int j) const {
        if (sequential && !single_file())
            return prev_for_sequential(C_, j);
        return prev_default(C_, j);
    }

    bool next(Glass::Cursor *C_, int j) const {
        if (sequential) return next_for_sequential(C_, j);
        return next_default(C_, j);
    }

    /* Default implementations. */
    bool prev_default(Glass::Cursor *C_, int j) const;
    bool next_default(Glass::Cursor *C_, int j) const;

    /* Implementations for sequential mode. */
    bool prev_for_sequential(Glass::Cursor *C_, int dummy) const;
    bool next_for_sequential(Glass::Cursor *C_, int dummy) const;

    static int find_in_leaf(const uint8_t * p,
                            Glass::LeafItem item, int c, bool& exact);
    static int find_in_branch(const uint8_t * p,
                              Glass::LeafItem item, int c);
    static int find_in_branch(const uint8_t * p, Glass::BItem item, int c);

    static uint4 block_given_by(const uint8_t * p, int c);

    mutable Glass::Cursor C[GLASS_BTREE_CURSOR_LEVELS];

    uint8_t * split_p;

    uint4 compress_min;

    mutable CompressionStream comp_stream;

    bool lazy;

    mutable uint4 last_readahead;

    off_t offset;

    /* Debugging methods */
//    void report_block_full(int m, int n, const uint8_t * p);
};

namespace Glass {

template<typename ITEM1, typename ITEM2>
int compare(ITEM1 a, ITEM2 b)
{
    Key key1 = a.key();
    Key key2 = b.key();
    const uint8_t* p1 = key1.data();
    const uint8_t* p2 = key2.data();
    int key1_len = key1.length();
    int key2_len = key2.length();
    int k_smaller = (key2_len < key1_len ? key2_len : key1_len);

    // Compare the common part of the keys.
    int diff = std::memcmp(p1, p2, k_smaller);
    if (diff == 0) {
        // If the common part matches, compare the lengths.
        diff = key1_len - key2_len;
        if (diff == 0) {
            // If the strings match, compare component_of().
            diff = a.component_of() - b.component_of();
        }
    }
    return diff;
}

inline int
compare(BItem a, BItem b)
{
    Key key1 = a.key();
    Key key2 = b.key();
    const uint8_t* p1 = key1.data();
    const uint8_t* p2 = key2.data();
    int key1_len = key1.length();
    int key2_len = key2.length();
    if (key1_len == key2_len) {
        // The keys are the same length, so we can compare the counts in the
        // same operation since they're stored as 2 byte bigendian numbers.
        int len = key1_len + X2;
        return std::memcmp(p1, p2, len);
    }

    int k_smaller = (key2_len < key1_len ? key2_len : key1_len);

    // Compare the common part of the keys.
    int diff = std::memcmp(p1, p2, k_smaller);
    if (diff == 0) {
        // If the common part matches, compare the lengths.
        diff = key1_len - key2_len;
        // We dealt with the "same length" case above so we never need to check
        // component_of here.
    }
    return diff;
}

}

#endif /* OM_HGUARD_GLASS_TABLE_H */