btree_erase.c

About: hamsterdb is a database engine written in ANSI C. It supports a B+Tree index structure, uses

/**
 * Copyright (C) 2005-2007 Christoph Rupp ([e-mail]).
 * All rights reserved. See file LICENSE for licence and copyright
 * information.
 *
 * btree erasing
 *
 */

#include <string.h>
#include "db.h"
#include "error.h"
#include "page.h"
#include "btree.h"
#include "mem.h"
#include "util.h"
#include "keys.h"
#include "extkeys.h"
#include "blob.h"

/*
 * the erase_scratchpad_t structure helps us to propagate return values
 * from the bottom of the tree to the root.
 */
typedef struct
{
    /*
     * the backend pointer
     */
    ham_btree_t *be;

    /*
     * the flags of the ham_erase()-call
     */
    ham_u32_t flags;

    /*
     * the key which will be deleted
     */
    ham_key_t *key;

    /*
     * a pointer to the record id of the deleted key
     */
    ham_offset_t *rid;

    /*
     * the internal flags of the key
     */
    ham_u32_t *intflags;

    /*
     * a page which needs rebalancing
     */
    ham_page_t *mergepage;

} erase_scratchpad_t;

/*
 * recursively descend down the tree, delete the item and re-balance 
 * the tree on the way back up
 * returns the page which is deleted, if available
 */
static ham_page_t *
my_erase_recursive(ham_page_t *page, ham_offset_t left, ham_offset_t right, 
        ham_offset_t lanchor, ham_offset_t ranchor, ham_page_t *parent,
        erase_scratchpad_t *scratchpad);

/*
 * collapse the root node
 */
static ham_status_t
my_collapse_root(ham_page_t *root, erase_scratchpad_t *scratchpad);

/*
 * rebalance a page - either shifts elements to a sibling, or merges 
 * the page with a sibling
 */
static ham_page_t *
my_rebalance(ham_page_t *page, ham_offset_t left, ham_offset_t right, 
        ham_offset_t lanchor, ham_offset_t ranchor, ham_page_t *parent,
        erase_scratchpad_t *scratchpad);

/*
 * merge two pages
 */
static ham_page_t *
my_merge_pages(ham_page_t *page, ham_page_t *sibling, ham_offset_t anchor,
        erase_scratchpad_t *scratchpad);

/*
 * shift items from a sibling to this page, till both pages have an equal 
 * number of items
 */
static ham_page_t *
my_shift_pages(ham_page_t *page, ham_page_t *sibpage, ham_offset_t anchor,
        erase_scratchpad_t *scratchpad);

/*
 * copy a key
 */
static ham_status_t
my_copy_key(ham_db_t *db, int_key_t *lhs, int_key_t *rhs);

/*
 * replace two keys in a page 
 */
static ham_status_t
my_replace_key(ham_page_t *page, ham_s32_t slot, 
        int_key_t *newentry, ham_u32_t flags);

/*
 * remove an item from a page 
 */
static ham_status_t
my_remove_entry(ham_page_t *page, ham_s32_t slot, 
        erase_scratchpad_t *scratchpad);

/*
 * flags for my_replace_key
 */
#define NOFLUSH 1
#define INTERNAL_KEY 2


ham_status_t
btree_erase(ham_btree_t *be, ham_key_t *key, 
        ham_offset_t *rid, ham_u32_t *intflags, ham_u32_t flags)
{
    ham_status_t st=0;
    ham_page_t *root, *p;
    ham_offset_t rootaddr;
    ham_db_t *db=btree_get_db(be);
    erase_scratchpad_t scratchpad;

    /* 
     * initialize the scratchpad 
     */
    memset(&scratchpad, 0, sizeof(scratchpad));
    scratchpad.be=be;
    scratchpad.key=key;
    scratchpad.rid=rid;
    scratchpad.intflags=intflags;
    scratchpad.flags=flags;

    /* 
     * get the root-page...
     */
    rootaddr=btree_get_rootpage(be);
    if (!rootaddr)
        return (db_set_error(db, HAM_KEY_NOT_FOUND));
    root=db_fetch_page(db, rootaddr, flags);

    db_set_error(db, 0);

    /* 
     * ... and start the recursion 
     */
    p=my_erase_recursive(root, 0, 0, 0, 0, 0, &scratchpad);
    if (db_get_error(db))
        return (db_get_error(db));
    if (p) {
        st=my_collapse_root(p, &scratchpad);
        if (st)
            return (st);

        /* 
         * delete the old root page 
         */
        st=txn_free_page(db_get_txn(db), root);
        if (st)
            return (st);
    }

    return (st);
}

static ham_page_t *
my_erase_recursive(ham_page_t *page, ham_offset_t left, ham_offset_t right, 
        ham_offset_t lanchor, ham_offset_t ranchor, ham_page_t *parent,
        erase_scratchpad_t *scratchpad)
{
    ham_s32_t slot;
    ham_bool_t isfew;
    ham_status_t st;
    ham_offset_t next_left, next_right, next_lanchor, next_ranchor;
    ham_page_t *newme, *child, *tempp=0;
    ham_db_t *db=page_get_owner(page);
    btree_node_t *node=ham_page_get_btree_node(page);
    ham_size_t maxkeys=btree_get_maxkeys(scratchpad->be);

    /* 
     * empty node? then most likely we're in the empty root page.
     */
    if (btree_node_get_count(node)==0) {
        db_set_error(db, HAM_KEY_NOT_FOUND);
        return 0;
    }

    /*
     * mark the nodes which may need rebalancing
     */
    if (btree_get_rootpage(scratchpad->be)==page_get_self(page))
        isfew=(btree_node_get_count(node)>1);
    else
        isfew=(btree_node_get_count(node)>btree_get_minkeys(maxkeys)); 

    if (isfew)
        scratchpad->mergepage=0;
    else if (!scratchpad->mergepage)
        scratchpad->mergepage=page;

    if (!btree_node_is_leaf(node)) {
        child=btree_traverse_tree(db, page, scratchpad->key, &slot);
        ham_assert(child!=0, ("guru meditation error"));
    }
    else {
        st=btree_get_slot(db, page, scratchpad->key, &slot);
        if (st) {
            db_set_error(db, st);
            return 0;
        }
        child=0;
    }

    /*
     * if this page is not a leaf: recursively descend down the tree
     */
    if (!btree_node_is_leaf(node)) {
        /*
         * calculate neighbor and anchor nodes
         */
        if (slot==-1) {
            if (!left)
                next_left=0;
            else {
                int_key_t *bte; 
                btree_node_t *n;
                tempp=db_fetch_page(db, left, 0);
                n=ham_page_get_btree_node(tempp);
                bte=btree_node_get_key(db, n, btree_node_get_count(n)-1);
                next_left=key_get_ptr(bte);
            }
            next_lanchor=lanchor;
        }
        else {
            if (slot==0)
                next_left=btree_node_get_ptr_left(node);
            else {
                int_key_t *bte; 
                bte=btree_node_get_key(db, node, slot-1);
                next_left=key_get_ptr(bte);
            }
            next_lanchor=page_get_self(page);
        }

        if (slot==btree_node_get_count(node)-1) {
            if (!right)
                next_right=0;
            else {
                int_key_t *bte; 
                btree_node_t *n;
                tempp=db_fetch_page(db, right, 0);
                n=ham_page_get_btree_node(tempp);
                bte=btree_node_get_key(db, n, 0);
                next_right=key_get_ptr(bte);
            }
            next_ranchor=ranchor;
        }
        else {
            int_key_t *bte; 
            bte=btree_node_get_key(db, node, slot+1);
            next_right=key_get_ptr(bte);
            next_ranchor=page_get_self(page);
        }

        newme=my_erase_recursive(child, next_left, next_right, next_lanchor, 
                    next_ranchor, page, scratchpad);
    }
    /*
     * otherwise (page is a leaf) delete the key...
     */
    else {
        /*
         * check if this entry really exists
         */
        newme=0;
        if (slot!=-1) {
            int cmp;
            int_key_t *bte;

            bte=btree_node_get_key(db, node, slot);

            cmp=db_compare_keys(db, page, 
                    -1, scratchpad->key->_flags, scratchpad->key->data, 
                    scratchpad->key->size,
                    slot, key_get_flags(bte), key_get_key(bte), 
                    key_get_size(bte));
            if (db_get_error(db))
                return (0);
            if (cmp==0) {
                *scratchpad->rid=key_get_ptr(bte);
                *scratchpad->intflags=key_get_flags(bte);
                newme=page;
            }
            else {
                db_set_error(db, HAM_KEY_NOT_FOUND);
                return (0);
            }
        }
        if (!newme) {
            db_set_error(db, HAM_KEY_NOT_FOUND);
            scratchpad->mergepage=0;
            return (0);
        }
    }

    /*
     * ... and rebalance the tree, if necessary
     */
    if (newme) {
        if (slot==-1)
            slot=0;
        st=my_remove_entry(page, slot, scratchpad);
        if (st)
            return (0);
    }

    /*
     * no need to rebalance in case of an error
     */
    if (!db_get_error(db))
        return (my_rebalance(page, left, right, lanchor, ranchor, parent, 
                scratchpad));
    else
        return (0);
}

static ham_status_t
my_collapse_root(ham_page_t *newroot, erase_scratchpad_t *scratchpad)
{
    btree_set_rootpage(scratchpad->be, page_get_self(newroot));
    btree_set_dirty(scratchpad->be, HAM_TRUE);
    db_set_dirty(page_get_owner(newroot), 1);
    page_set_type(newroot, PAGE_TYPE_B_ROOT);
    return (0);
}

static ham_page_t *
my_rebalance(ham_page_t *page, ham_offset_t left, ham_offset_t right, 
        ham_offset_t lanchor, ham_offset_t ranchor, ham_page_t *parent,
        erase_scratchpad_t *scratchpad)
{
    btree_node_t *node=ham_page_get_btree_node(page);
    ham_page_t *leftpage, *rightpage;
    btree_node_t *leftnode=0, *rightnode=0;
    ham_bool_t fewleft=HAM_FALSE, fewright=HAM_FALSE;
    ham_size_t maxkeys=btree_get_maxkeys(scratchpad->be);
    ham_size_t minkeys=btree_get_minkeys(maxkeys);

    if (!scratchpad->mergepage)
        return (0);

    /*
     * get the left and the right sibling of this page
     */
    leftpage =left
                ? db_fetch_page(page_get_owner(page), 
                        btree_node_get_left(node), 0)
                : 0;
    if (leftpage) {
        leftnode =ham_page_get_btree_node(leftpage);
        fewleft  =(btree_node_get_count(leftnode)<=minkeys) 
                ? HAM_TRUE : HAM_FALSE;
    }
    rightpage=right
                ? db_fetch_page(page_get_owner(page), 
                        btree_node_get_right(node), 0)
                : 0;
    if (rightpage) {
        rightnode=ham_page_get_btree_node(rightpage);
        fewright =(btree_node_get_count(rightnode)<=minkeys) 
                ? HAM_TRUE : HAM_FALSE;
    }

    /*
     * if we have no siblings, then we're rebalancing the root page
     */
    if (!leftpage && !rightpage) {