backends/flint/flint_table.h

Go to the documentation of this file.

/* flint_table.h: Btree implementation
 *
 * Copyright 1999,2000,2001 BrightStation PLC
 * Copyright 2002,2003,2004,2005,2006,2007,2008,2009,2010 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_FLINT_TABLE_H
#define OM_HGUARD_FLINT_TABLE_H

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

#include "flint_types.h"
#include "flint_btreebase.h"
#include "flint_cursor.h"

#include "noreturn.h"
#include "str.h"
#include "stringutils.h"
#include "unaligned.h"

#include <algorithm>
#include <string>

#include <zlib.h>

#define DONT_COMPRESS -1

#define FLINT_BTREE_MAX_KEY_LEN 252

#define BLOCK_CAPACITY 4

// FIXME: This named constant probably isn't used everywhere it should be...
#define 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
*/

#define K1 1
#define I2 2
#define D2 2
#define C2 2

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

#define getK(p, c)    getint1(p, c)
#define setD(p, c, 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,
*/

class XAPIAN_VISIBILITY_DEFAULT Key_ {
    const byte *p;
public:
    explicit Key_(const byte * p_) : p(p_) { }
    const byte * 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 {
        return getK(p, 0) - C2 - K1;
    }
    char operator[](size_t i) const {
        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)) { }
    Item_base_(T p_) : p(p_) { }
    T get_address() const { return p; }
    int size() const { return getint2(p, 0) & 0x7fff; } /* I in diagram above */
    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 {
        return getint4(p, size() - BYTES_PER_BLOCK_NUMBER);
    }
};

class Item_ : public Item_base_<const byte *> {
public:
    /* Item_ from block address and offset to item pointer */
    Item_(const byte * p_, int c) : Item_base_<const byte *>(p_, c) { }
    Item_(const byte * p_) : Item_base_<const byte *>(p_) { }
};

class Item_wr_ : public Item_base_<byte *> {
    void set_key_len(int x) { setint1(p, I2, x); }
public:
    /* Item_wr_ from block address and offset to item pointer */
    Item_wr_(byte * p_, int c) : Item_base_<byte *>(p_, c) { }
    Item_wr_(byte * p_) : Item_base_<byte *>(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();
        int newsize = I2 + K1 + i + C2;
        // Item size (4 since tag contains block number)
        setint2(p, 0, newsize + 4);
        // 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) { 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 + 4);  /* total length */
    }
    void form_key(const std::string & key_) {
        std::string::size_type key_len = key_.length();
        if (key_len > FLINT_BTREE_MAX_KEY_LEN) {
            // We check term length when a term is added to a document but
            // flint 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(FLINT_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...
#define BTREE_CURSOR_LEVELS 10

class XAPIAN_VISIBILITY_DEFAULT FlintTable {
    friend class FlintCursor; /* Should probably fix this. */
    private:
        FlintTable(const FlintTable &);

        FlintTable & operator=(const FlintTable &);

        bool really_empty() const;

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

        ~FlintTable();

        void close(bool permanent=false);

        bool exists() const;

        void open();

        bool open(flint_revision_number_t revision_);

        void flush_db();

        void commit(flint_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);

        flint_revision_number_t get_latest_revision_number() const {
            return latest_revision_number;
        }

        flint_revision_number_t get_open_revision_number() const {
            return revision_number;
        }

        flint_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();
        }

        FlintCursor * 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 - 11 /*DIR_START*/ - block_capacity * D2)
                / block_capacity;
        }

    protected:

        bool do_open_to_read(bool revision_supplied, flint_revision_number_t revision_);

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

        bool find(Cursor_ *) const;
        int delete_kt();
        void read_block(uint4 n, byte *p) const;
        void write_block(uint4 n, const byte *p) const;
        XAPIAN_NORETURN(void set_overwritten() const);
        void block_to_cursor(Cursor_ *C_, int j, uint4 n) const;
        void alter();
        void compact(byte *p);
        void enter_key(int j, Key_ prevkey, Key_ newkey);
        int mid_point(byte *p);
        void add_item_to_block(byte *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;

        const char * tablename;

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

        void lazy_alloc_deflate_zstream() const;

        void lazy_alloc_inflate_zstream() const;

        flint_revision_number_t revision_number;

        uint4 item_count;

        unsigned int block_size;

        mutable flint_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;

        byte * buffer;

        FlintTable_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 byte * p, Key_ key, bool leaf, int c);

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

        mutable Cursor_ C[BTREE_CURSOR_LEVELS];

        byte * split_p;

        int compress_strategy;

        mutable z_stream *deflate_zstream;

        mutable z_stream *inflate_zstream;

        bool lazy;

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

        XAPIAN_NORETURN(static void throw_database_closed());
};

#endif /* OM_HGUARD_FLINT_TABLE_H */

---