/* flint_table.h: Btree implementation
 *
 * Copyright 1999,2000,2001 BrightStation PLC
 * Copyright 2002,2003,2004,2005,2006,2007,2008 Olly Betts
 *
 * 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 <algorithm>
#include <string>
using std::string;

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

#include "stringutils.h"
#include "utils.h"

#include <zlib.h>

#define DONT_COMPRESS -1

/** The largest possible value of a key_len.
 *
 *  This gives the upper limit of the size of a key that may be stored in the
 *  B-tree (252 bytes with the present implementation).
 */
#define FLINT_BTREE_MAX_KEY_LEN 252

// 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(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(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);
    }
    /** Get this item's tag as a block number (this block should not be at
     *  level 0).
     */
    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.
	memmove(p + I2 + K1, newkey.get_address() + K1, i);
	// Copy the count part.
	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);
    }

    /** Set this item's tag to point to block n (this block should not be at
     *  level 0).
     */
    void set_block_given_by(uint4 n) {
	setint4(p, size() - BYTES_PER_BLOCK_NUMBER, n);
    }
    void set_size(int l) { setint2(p, 0, l); }
    /** Form an item with a null key and with block number n in the tag.
     */
    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 string & key_) {
	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.
	    string msg("Key too long: length was ");
	    msg += om_tostring(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);
	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) {
	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 managing a Btree table in a Flint database.
 *
 *  A table is a store holding a set of key/tag pairs.
 *
 *  A key is used to access a block of data in a flint table.
 * 
 *  Keys are of limited length.
 *
 *  Keys may not be empty (each Btree has a special empty key for internal use).
 *
 *  A tag is a piece of data associated with a given key.  The contents
 *  of the tag are opaque to the Btree.
 *
 *  Tags may be of arbitrary length (the Btree imposes a very large limit).
 *  Note though that they will be loaded into memory in their entirety, so
 *  should not be permitted to grow without bound in normal usage.
 *
 *  Tags which are null strings _are_ valid, and are different from a
 *  tag simply not being in the table.
 */
class XAPIAN_VISIBILITY_DEFAULT FlintTable {
    friend class FlintCursor; /* Should probably fix this. */
    private:
	/// Copying not allowed
        FlintTable(const FlintTable &);

	/// Assignment not allowed
        FlintTable & operator=(const FlintTable &);

    public:
	/** Create a new Btree object.
	 *
	 *  This does not create the table on disk - the create_and_open()
	 *  method must be called to create the table on disk.
	 *
	 *  This also does not open the table - either the create_and_open()
	 *  or open() methods must be called before use is made of the table.
	 *
	 *  @param path_	Path at which the table is stored.
	 *  @param readonly_	whether to open the table for read only access.
	 *  @param compress_strategy_	DONT_COMPRESS, Z_DEFAULT_STRATEGY,
	 *				Z_FILTERED, Z_HUFFMAN_ONLY, or Z_RLE.
	 *  @param lazy		If true, don't create the table until it's
	 *			needed.
	 */
	FlintTable(string path_, bool readonly_,
		   int compress_strategy_ = DONT_COMPRESS, bool lazy = false);

	/** Close the Btree.
	 *
	 *  Any outstanding changes (ie, changes made without commit() having
	 *  subsequently been called) will be lost.
	 */
	~FlintTable();

	/** Close the Btree.  This closes and frees any of the btree
	 *  structures which have been created and opened.
	 */
	void close();

	/** Determine whether the btree exists on disk.
	 */
	bool exists() const;

	/** Open the btree at the latest revision.
	 *
	 *  @exception Xapian::DatabaseCorruptError will be thrown if the table
	 *	is in a corrupt state.
	 *  @exception Xapian::DatabaseOpeningError will be thrown if the table
	 *	cannot be opened (but is not corrupt - eg, permission problems,
	 *	not present, etc).
	 */
	void open();

	/** Open the btree at a given revision.
	 *
	 *  Like Btree::open, but try to open at the given revision number
	 *  and fail if that isn't possible.
	 *
	 *  @param revision_      - revision number to open.
	 *
	 *  @return true if table is successfully opened at desired revision;
	 *          false if table cannot be opened at desired revision (but
	 *          table is otherwise consistent).
	 *
	 *  @exception Xapian::DatabaseCorruptError will be thrown if the table
	 *	is in a corrupt state.
	 *  @exception Xapian::DatabaseOpeningError will be thrown if the table
	 *	cannot be opened (but is not corrupt - eg, permission problems,
	 *	not present, etc).
	 */
	bool open(flint_revision_number_t revision_);

	/** Commit any outstanding changes to the table.
	 *
	 *  Commit changes made by calling add() and del() to the Btree.
	 *
	 *  If an error occurs during the operation, this will be signalled
	 *  by an exception.  In case of error, changes made will not be
	 *  committed to the Btree - they will be discarded.
	 *
	 *  @param new_revision  The new revision number to store.  This must
	 *          be greater than the latest revision number (see
	 *          get_latest_revision_number()), or an exception will be
	 *          thrown.
	 */
	void commit(flint_revision_number_t revision);

	/** Cancel any outstanding changes.
	 *
	 *  This will discard any modifications which haven't been committed
	 *  by calling commit().
	 */
	void cancel();

	/** Read an entry from the table, if and only if it is exactly that
	 *  being asked for.
	 *
	 *  If the key is found in the table, the tag will be filled with
	 *  the data associated with the key.  If the key is not found,
	 *  the tag will be unmodified.
	 *
	 *  @param key  The key to look for in the table.
	 *  @param tag  A tag object to fill with the value if found.
	 *
	 *  @return true if key is found in table,
	 *          false if key is not found in table.
	 */
	bool get_exact_entry(const string & key, string & tag) const;

	/** Check if a key exists in the Btree.
	 *
	 *  This is just like get_exact_entry() except it doesn't read the tag
	 *  value so is more efficient if you only want to check that the key
	 *  exists.
	 *
	 *  @param key  The key to look for in the table.
	 *
	 *  @return true if key is found in table,
	 *          false if key is not found in table.
	 */
	bool key_exists(const string &key) const;

	/** Find a key in the Btree and read its tag.
	 *
	 *  If the key is found the tag is copied to tag.  If the key is not
	 *  found tag is left unchanged.
	 * 
	 *  The result is true iff the specified key is found in the Btree.
	 *
	 *  e.g.
	 *
	 *    string t;
	 *    btree.find_tag("TODAY", &t); // get today's date
	 */
	bool find_tag(const string &key, string * tag) const;

	/** Read the tag value for the key pointed to by cursor C_.
	 *
	 *  @param keep_compressed  Don't uncompress the tag - e.g. useful
	 *			    if it's just being opaquely copied.
	 *
	 *  @return	true if current_tag holds compressed data (always
	 *		false if keep_compressed was false).
	 */
	bool read_tag(Cursor_ * C_, string *tag, bool keep_compressed) const;

	/** Add a key/tag pair to the table, replacing any existing pair with
	 *  the same key.
	 *
	 *  If an error occurs during the operation, this will be signalled
	 *  by a return value of false.  All modifications since the
	 *  previous commit() will be lost.
	 *
	 *  If key is empty, then the null item is replaced.  If key.length()
	 *  exceeds the limit on key size, false is returned.
	 *
	 *  e.g.    ok = btree.add("TODAY", "Mon 9 Oct 2000");
	 *
	 *  @param key   The key to store in the table.
	 *  @param tag   The tag to store in the table.
	 *  @param already_compressed	true if tag is already compressed,
	 *		for example because it is being opaquely copied
	 *		(default: false).
	 *
	 *  @return true if the operation completed successfully, false
	 *          otherwise.
	 */
	bool add(const string &key, string tag, bool already_compressed = false);

	/** Delete an entry from the table.
	 *
	 *  The entry will be removed from the table, if it exists.  If
	 *  it does not exist, no action will be taken.  The item with
	 *  an empty key can't be removed, and false is returned.
	 *
	 *  If an error occurs during the operation, this will be signalled
	 *  by a return value of false.  All modifications since the
	 *  previous commit() will be lost.
	 *
	 *  e.g.    ok = btree.del("TODAY")
	 *
	 *  @param key   The key to remove from the table.
	 *
	 *  @return true if the operation completed successfully, false
	 *          otherwise.
	 */
	bool del(const string &key);

	/// Erase this table from disk.
	void erase();

	/** Set the block size.
	 *
	 *  It's only safe to do this before the table is created.
	 */
	void set_block_size(unsigned int block_size_);

	/** Get the block size.
	 */
	unsigned int get_block_size() const { return block_size; }

	/** Create a new empty btree structure on disk and open it at the
	 *  initial revision.
	 *
	 *  The table must be writable - it doesn't make sense to create
	 *  a table that is read-only!
	 *
	 *  The block size must be less than 64K, where K = 1024. It is unwise
	 *  to use a small block size (less than 1024 perhaps), so we enforce a
	 *  minimum block size of 2K.
	 *
	 *  Example:
	 *
	 *    Btree btree("X-");
	 *    btree.create_and_open(8192);
	 *    // Files will be X-DB, X-baseA (and X-baseB).
	 *
	 *  @param blocksize     - Size of blocks to use.
	 *
	 *  @exception Xapian::DatabaseCreateError if the table can't be
	 *	created.
	 *  @exception Xapian::InvalidArgumentError if the requested blocksize
	 *	is unsuitable.
	 */
	void create_and_open(unsigned int blocksize);

	void set_full_compaction(bool parity);

	/** Get the latest revision number stored in this table.
	 *
	 *  This gives the higher of the revision numbers held in the base
	 *  files of the B-tree, or just the revision number if there's only
	 *  one base file.
	 *
	 *  It is possible that there are other, older, revisions of this
	 *  table available, and indeed that the revision currently open
	 *  is one of these older revisions.
	 */
	flint_revision_number_t get_latest_revision_number() const {
	    return latest_revision_number;
	}

	/** Get the revision number at which this table
	 *  is currently open.
	 *
	 *  It is possible that there are other, more recent or older
	 *  revisions available.
	 *
	 *  @return the current revision number.
	 */
	flint_revision_number_t get_open_revision_number() const {
	    return revision_number;
	}

	/** Return a count of the number of entries in the table.
	 *
	 *  The count does not include the ever-present item with null key.
	 *
	 *  @return The number of entries in the table.
	 */
	flint_tablesize_t get_entry_count() const {
	    return item_count;
	}

	/** Get a cursor for reading from the table.
	 *
	 *  The cursor is owned by the caller - it is the caller's
	 *  responsibility to ensure that it is deleted.
	 */
	FlintCursor * cursor_get() const;

	/** Determine whether the object contains uncommitted modifications.
	 *
	 *  @return true if there have been modifications since the last
	 *          the last call to commit().
	 */
	bool is_modified() const { return Btree_modified; }

	/** Set the maximum item size given the block capacity.
	 *
	 *  At least this many items of maximum size must fit into a block.
	 *  The default is BLOCK_CAPACITY (which is currently 4).
	 */
	void set_max_item_size(size_t block_capacity) {
	    if (block_capacity > 4) block_capacity = 4;
	    max_item_size = (block_size - 11 /*DIR_START*/ - block_capacity * D2)
		/ block_capacity;
	}

    protected:

	/** Perform the opening operation to read.
	 *
	 *  Return true iff the open succeeded.
	 */
	bool do_open_to_read(bool revision_supplied, flint_revision_number_t revision_);

	/** Perform the opening operation to write.
	 *
	 *  Return true iff the open succeeded.
	 */
	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 string & key) const;

	/** revision number of the opened B-tree. */
	flint_revision_number_t revision_number;

	/** keeps a count of the number of items in the B-tree. */
	uint4 item_count;

	/** block size of the B tree in bytes */
	unsigned int block_size;

	/** Revision number of the other base, or zero if there is only one
	 *  base file.
	 */
	mutable flint_revision_number_t latest_revision_number;

	/** set to true if baseA and baseB both exist as valid bases.
	 *
	 *  The unused base is deleted as soon as a write to the Btree takes
	 *  place. */
	mutable bool both_bases;

	/** the value 'A' or 'B' of the current base */
	int base_letter;

	/** true if the root block is faked (not written to disk).
	 * false otherwise.  This is true when the btree hasn't been
	 * modified yet.
	 */
	bool faked_root_block;

	/** true iff the data has been written in a single write in
	 * sequential order.
	 */
	bool sequential;

	/// corresponding file handle
	int handle;

	/// number of levels, counting from 0
	int level;

	/// the root block of the B-tree
	uint4 root;

	/// buffer of size block_size for making up key-tag items
	mutable Item_wr_ kt;

	/// buffer of size block_size for reforming blocks
	byte * buffer;

	/// For writing back as file baseA or baseB.
	FlintTable_base base;

	/// The base letter ('B' or 'A') of the next base.
	char other_base_letter;

	/// The path name of the B tree.
	string name;

	/** count of the number of successive instances of purely
	 * sequential addition, starting at SEQ_START_POINT (neg) and
	 * going up to zero. */
	int seq_count;

	/** the last block to be changed by an addition */
	uint4 changed_n;

	/** directory offset corresponding to last block to be changed
	 *  by an addition */
	int changed_c;

	/// maximum size of an item (key-tag pair)
	size_t max_item_size;

	/// Set to true the first time the B-tree is modified.
	mutable bool Btree_modified;

	/// set to true when full compaction is to be achieved
	bool full_compaction;

	/// Set to true when the database is opened to write.
	bool writable;

	/* 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);

	/** block_given_by(p, c) finds the item at block address p, directory
	 *  offset c, and returns its tag value as an integer.
	 */
	static uint4 block_given_by(const byte * p, int c);

	mutable Cursor_ C[BTREE_CURSOR_LEVELS];

	/** Buffer used when splitting a block.
	 *
	 *  This buffer holds the split off part of the block.  It's only used
	 *  when updating (in FlintTable::add_item().
	 */
	byte * split_p;

	/** DONT_COMPRESS or Z_DEFAULT_STRATEGY, Z_FILTERED, Z_HUFFMAN_ONLY,
	 *  Z_RLE. */
	int compress_strategy;

	/// If true, don't create the table until it's needed.
	bool lazy;

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

#endif /* OM_HGUARD_FLINT_TABLE_H */