chert_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,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_CHERT_TABLE_H
#define OM_HGUARD_CHERT_TABLE_H

#include <xapian/error.h>

#include "chert_types.h"
#include "chert_btreebase.h"
#include "chert_cursor.h"

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

#include <algorithm>
#include <string>

#include <zlib.h>

// FIXME: 65536 in Asserts below is the max chert block size.  We should
// abstract this out, and use the current block_size to catch overruns better.
inline int
getint1(const unsigned char *p, int c)
{
    AssertRel(c, >=, 0);
    AssertRel(c, <, 65536);
    return p[c];
}

inline void
setint1(unsigned char *p, int c, int x)
{
    AssertRel(c, >=, 0);
    AssertRel(c, <, 65536);
    p[c] = x;
}

inline int
getint2(const unsigned char *p, int c)
{
    AssertRel(c, >=, 0);
    AssertRel(c, <, 65536 - 1);
    return unaligned_read2(p + c);
}

inline void
setint2(unsigned char *p, int c, int x)
{
    AssertRel(c, >=, 0);
    AssertRel(c, <, 65536 - 1);
    unaligned_write2(p + c, uint16_t(x));
}

inline int
getint4(const unsigned char *p, int c)
{
    AssertRel(c, >=, 0);
    AssertRel(c, <, 65536 - 3);
    return unaligned_read4(p + c);
}

inline void
setint4(unsigned char *p, int c, int x)
{
    AssertRel(c, >=, 0);
    AssertRel(c, <, 65536 - 3);
    unaligned_write4(p + c, uint32_t(x));
}

const int DONT_COMPRESS = -1;

#define CHERT_BTREE_MAX_KEY_LEN 252

const size_t BLOCK_CAPACITY = 4;

// 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
       C2      the 2 byte counter that ends each key and begins each tag
*/

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

/*  and when getting K1 or setting D2, we use getK, setD defined as: */

inline int getK(const unsigned char *p, int c) { return getint1(p, c); }
inline void setD(unsigned char *p, int c, int x) { setint2(p, c, x); }

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

   Recall that item has this form:

           i k
           | |
           I K key x C tag
             <--K-->
           <------I------>


   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,

   components_of(p, c) returns the number marked 'C' above,
*/

inline unsigned REVISION(const uint8_t * b) { return aligned_read4(b); }
inline int GET_LEVEL(const uint8_t * b) { return getint1(b, 4); }
inline int MAX_FREE(const uint8_t * b) { return getint2(b, 5); }
inline int TOTAL_FREE(const uint8_t * b) { return getint2(b, 7); }
inline int DIR_END(const uint8_t * b) { return getint2(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) { setint1(b, 4, x); }
inline void SET_MAX_FREE(uint8_t * b, int x) { setint2(b, 5, x); }
inline void SET_TOTAL_FREE(uint8_t * b, int x) { setint2(b, 7, x); }
inline void SET_DIR_END(uint8_t * b, int x) { setint2(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".
const size_t CHERT_MAX_ITEM_SIZE = 0x7fff;

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

// Item_wr wants to be "Item 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 Item_base {
protected:
    T p;
public:
    /* Item from block address and offset to item pointer */
    Item_base(T p_, int c) : p(p_ + getint2(p_, c)) { }
    explicit Item_base(T p_) : p(p_) { }
    T get_address() const { return p; }
    int size() const {
        int item_size = getint2(p, 0) & CHERT_MAX_ITEM_SIZE;
        AssertRel(item_size,>=,5);
        return item_size;
    }
    bool get_compressed() const { return *p & 0x80; }
    int component_of() const {
        return getint2(p, getK(p, I2) + I2 - C2);
    }
    int components_of() const {
        return getint2(p, getK(p, I2) + I2);
    }
    Key key() const { return Key(p + I2); }
    void append_chunk(std::string * tag) const {
        /* number of bytes to extract from current component */
        int cd = getK(p, I2) + I2 + C2;
        int l = size() - cd;
        tag->append(reinterpret_cast<const char *>(p + cd), l);
    }
    uint4 block_given_by() const {
        AssertRel(size(),>=,BYTES_PER_BLOCK_NUMBER);
        return getint4(p, size() - BYTES_PER_BLOCK_NUMBER);
    }
};

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

class Item_wr : public Item_base<uint8_t *> {
    void set_key_len(int x) { setint1(p, I2, x); }
public:
    /* Item_wr from block address and offset to item pointer */
    Item_wr(uint8_t * p_, int c) : Item_base<uint8_t *>(p_, c) { }
    explicit Item_wr(uint8_t * p_) : Item_base<uint8_t *>(p_) { }
    void set_component_of(int i) {
        setint2(p, getK(p, I2) + I2 - C2, i);
    }
    void set_components_of(int m) {
        setint2(p, getK(p, I2) + I2, m);
    }
    // Takes size as we may be truncating newkey.
    void set_key_and_block(Key newkey, int truncate_size, uint4 n) {
        int i = truncate_size;
        // Read the length now because we may be copying the key over itself.
        // FIXME that's stupid!  sort this out
        int newkey_len = newkey.length();
        AssertRel(i,<=,newkey_len);
        int newsize = I2 + K1 + i + C2;
        // Item size (BYTES_PER_BLOCK_NUMBER since tag contains block number)
        setint2(p, 0, newsize + BYTES_PER_BLOCK_NUMBER);
        // Key size
        setint1(p, I2, newsize - I2);
        // Copy the main part of the key, possibly truncating.
        std::memmove(p + I2 + K1, newkey.get_address() + K1, i);
        // Copy the count part.
        std::memmove(p + I2 + K1 + i, newkey.get_address() + K1 + newkey_len, C2);
        // Set tag contents to block number
//      set_block_given_by(n);
        setint4(p, newsize, n);
    }

    void set_block_given_by(uint4 n) {
        setint4(p, size() - BYTES_PER_BLOCK_NUMBER, n);
    }
    void set_size(int l) {
        AssertRel(l,>=,5);
        // We should never be able to pass too large a size here, but don't
        // corrupt the database if this somehow happens.
        if (rare(l &~ CHERT_MAX_ITEM_SIZE))
            throw Xapian::DatabaseError("item too large!");
        setint2(p, 0, l);
    }
    void form_null_key(uint4 n) {
        setint4(p, I2 + K1, n);
        set_key_len(K1);        /* null key */
        set_size(I2 + K1 + BYTES_PER_BLOCK_NUMBER);  /* total length */
    }
    void form_key(const std::string & key_) {
        std::string::size_type key_len = key_.length();
        if (key_len > CHERT_BTREE_MAX_KEY_LEN) {
            // We check term length when a term is added to a document but
            // chert 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(CHERT_BTREE_MAX_KEY_LEN) " bytes";
            throw Xapian::InvalidArgumentError(msg);
        }

        set_key_len(key_len + K1 + C2);
        std::memmove(p + I2 + K1, key_.data(), key_len);
        set_component_of(1);
    }
    // FIXME passing cd here is icky
    void set_tag(int cd, const char *start, int len, bool compressed) {
        std::memmove(p + cd, start, len);
        set_size(cd + len);
        if (compressed) *p |= 0x80;
    }
    void fake_root_item() {
        set_key_len(K1 + C2);   // null key length
        set_size(I2 + K1 + 2 * C2);   // length of the item
        set_component_of(1);
        set_components_of(1);
    }
};

// Allow for BTREE_CURSOR_LEVELS levels in the B-tree.
// With 10, overflow is practically impossible
// FIXME: but we want it to be completely impossible...
const int BTREE_CURSOR_LEVELS = 10;

class ChertTable {
    friend class ChertCursor; /* Should probably fix this. */
    private:
        ChertTable(const ChertTable &);

        ChertTable & operator=(const ChertTable &);

        bool really_empty() const;

    public:
        ChertTable(const char * tablename_, const std::string & path_,
                   bool readonly_, int compress_strategy_ = DONT_COMPRESS,
                   bool lazy = false);

        ~ChertTable();

        void close(bool permanent = false);

        bool readahead_key(const string &key) const;

        bool exists() const;

        void open();

        bool open(chert_revision_number_t revision_);

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

        void flush_db();

        void commit(chert_revision_number_t revision, int changes_fd = -1,
                    const std::string * changes_tail = NULL);

        void write_changed_blocks(int changes_fd);

        void cancel();

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

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

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

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

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

        void erase();

        void set_block_size(unsigned int block_size_);

        unsigned int get_block_size() const { return block_size; }

        void create_and_open(unsigned int blocksize);

        void set_full_compaction(bool parity);

        chert_revision_number_t get_latest_revision_number() const {
            return latest_revision_number;
        }

        chert_revision_number_t get_open_revision_number() const {
            return revision_number;
        }

        chert_tablesize_t get_entry_count() const {
            return item_count;
        }

        bool empty() const {
            // Prior to 1.1.4/1.0.18, item_count was stored in 32 bits, so we
            // can't trust it as there could be more than 1<<32 entries.
            //
            // In theory it should wrap, so if non-zero the table isn't empty,
            // but the table this was first noticed in wasn't off by a multiple
            // of 1<<32.

            // An empty table will always have level == 0, and most non-empty
            // tables will have more levels, so use that as a short-cut.
            return (level == 0) && really_empty();
        }

        ChertCursor * cursor_get() const;

        bool is_modified() const { return Btree_modified; }

        void set_max_item_size(size_t block_capacity) {
            if (block_capacity > BLOCK_CAPACITY) block_capacity = BLOCK_CAPACITY;
            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 > CHERT_MAX_ITEM_SIZE)
                max_item_size = CHERT_MAX_ITEM_SIZE;
        }

        XAPIAN_NORETURN(static void throw_database_closed());

        string get_path() const {
            return name;
        }

    protected:

        bool do_open_to_read(bool revision_supplied, chert_revision_number_t revision_);

        bool do_open_to_write(bool revision_supplied,
                              chert_revision_number_t revision_,
                              bool create_db = false);
        bool basic_open(bool revision_supplied, chert_revision_number_t revision);

        bool find(Cursor *) const;
        int delete_kt();
        void read_block(uint4 n, uint8_t *p) const;
        void write_block(uint4 n, const uint8_t *p) const;
        XAPIAN_NORETURN(void set_overwritten() const);
        void block_to_cursor(Cursor *C_, int j, uint4 n) const;
        void alter();
        void compact(uint8_t *p);
        void enter_key(int j, Key prevkey, Key newkey);
        int mid_point(uint8_t *p) const;
        void add_item_to_block(uint8_t *p, Item_wr kt, int c);
        void add_item(Item_wr kt, int j);
        void delete_item(int j, bool repeatedly);
        int add_kt(bool found);
        void read_root();
        void split_root(uint4 split_n);
        void form_key(const std::string & key) const;

        char other_base_letter() const {
           return (base_letter == 'A') ? 'B' : 'A';
        }

        const char * tablename;

        void lazy_alloc_deflate_zstream() const;

        void lazy_alloc_inflate_zstream() const;

        chert_revision_number_t revision_number;

        chert_tablesize_t item_count;

        unsigned int block_size;

        mutable chert_revision_number_t latest_revision_number;

        mutable bool both_bases;

        char base_letter;

        bool faked_root_block;

        bool sequential;

        int handle;

        int level;

        uint4 root;

        mutable Item_wr kt;

        uint8_t * buffer;

        ChertTable_base base;

        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;

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

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

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

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

        static int find_in_block(const uint8_t * p, Key key, bool leaf, int c);

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

        mutable Cursor C[BTREE_CURSOR_LEVELS];

        uint8_t * split_p;

        int compress_strategy;

        mutable z_stream *deflate_zstream;

        mutable z_stream *inflate_zstream;

        bool lazy;

        mutable uint4 last_readahead;

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

#endif /* OM_HGUARD_CHERT_TABLE_H */