btree_cursor.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 cursors - implementation
 *
 */

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

static ham_status_t
my_set_to_nil(ham_bt_cursor_t *c)
{
    /*
     * uncoupled cursor: free the cached pointer
     */
    if (bt_cursor_get_flags(c)&BT_CURSOR_FLAG_UNCOUPLED) {
        ham_key_t *key=bt_cursor_get_uncoupled_key(c);
        if (key->data)
            ham_mem_free(cursor_get_db(c), key->data);
        ham_mem_free(cursor_get_db(c), key);
        bt_cursor_set_uncoupled_key(c, 0);
        bt_cursor_set_flags(c,
                bt_cursor_get_flags(c)&(~BT_CURSOR_FLAG_UNCOUPLED));
    }
    /*
     * coupled cursor: uncouple, remove from page
     */
    else if (bt_cursor_get_flags(c)&BT_CURSOR_FLAG_COUPLED) {
        page_remove_cursor(bt_cursor_get_coupled_page(c),
                (ham_cursor_t *)c);
        bt_cursor_set_flags(c,
                bt_cursor_get_flags(c)&(~BT_CURSOR_FLAG_COUPLED));
    }

    return (0);
}

static ham_status_t
my_move_first(ham_btree_t *be, ham_bt_cursor_t *c, ham_u32_t flags)
{
    ham_status_t st;
    ham_page_t *page;
    btree_node_t *node;
    ham_db_t *db=cursor_get_db(c);

    /*
     * get a NIL cursor
     */
    st=my_set_to_nil(c);
    if (st)
        return (st);

    db_set_error(db, 0);

    /*
     * get the root page
     */
    if (!btree_get_rootpage(be))
        return (db_set_error(db, HAM_KEY_NOT_FOUND));
    page=db_fetch_page(db, btree_get_rootpage(be), 0);
    if (!page)
        return (db_get_error(db));

    /*
     * while we've not reached the leaf: pick the smallest element
     * and traverse down
     */
    while (1) {
        node=ham_page_get_btree_node(page);
        /* check for an empty root page */
        if (btree_node_get_count(node)==0)
            return (db_set_error(db, HAM_KEY_NOT_FOUND));
        /* leave the loop when we've reached the leaf page */
        if (btree_node_is_leaf(node))
            break;

        page=db_fetch_page(db, btree_node_get_ptr_left(node), 0);
        if (!page) {
            if (!db_get_error(db))
                db_set_error(db, HAM_KEY_NOT_FOUND);
            return (db_get_error(db));
        }
    }

    /*
     * couple this cursor to the smallest key in this page
     */
    page_add_cursor(page, (ham_cursor_t *)c);
    bt_cursor_set_coupled_page(c, page);
    bt_cursor_set_coupled_index(c, 0);
    bt_cursor_set_flags(c,
            bt_cursor_get_flags(c)|BT_CURSOR_FLAG_COUPLED);

    return (0);
}

static ham_status_t
my_move_last(ham_btree_t *be, ham_bt_cursor_t *c, ham_u32_t flags)
{
    ham_status_t st;
    ham_page_t *page;
    btree_node_t *node;
    ham_db_t *db=cursor_get_db(c);

    /*
     * get a NIL cursor
     */
    st=my_set_to_nil(c);
    if (st)
        return (st);

    db_set_error(db, 0);

    /*
     * get the root page
     */
    if (!btree_get_rootpage(be))
        return (db_set_error(db, HAM_KEY_NOT_FOUND));
    page=db_fetch_page(db, btree_get_rootpage(be), 0);
    if (!page)
        return (db_get_error(db));

    /*
     * while we've not reached the leaf: pick the largest element
     * and traverse down
     */
    while (1) {
        int_key_t *key;

        node=ham_page_get_btree_node(page);
        /* check for an empty root page */
        if (btree_node_get_count(node)==0)
            return (db_set_error(db, HAM_KEY_NOT_FOUND));
        /* leave the loop when we've reached a leaf page */
        if (btree_node_is_leaf(node))
            break;

        key=btree_node_get_key(db, node, btree_node_get_count(node)-1);

        page=db_fetch_page(db, key_get_ptr(key), 0);
        if (!page) {
            if (!db_get_error(db))
                db_set_error(db, HAM_KEY_NOT_FOUND);
            return (db_get_error(db));
        }
    }

    /*
     * couple this cursor to the largest key in this page
     */
    page_add_cursor(page, (ham_cursor_t *)c);
    bt_cursor_set_coupled_page(c, page);
    bt_cursor_set_coupled_index(c, btree_node_get_count(node)-1);
    bt_cursor_set_flags(c,
            bt_cursor_get_flags(c)|BT_CURSOR_FLAG_COUPLED);

    return (0);
}

static ham_status_t
my_move_next(ham_btree_t *be, ham_bt_cursor_t *c, ham_u32_t flags)
{
    ham_status_t st;
    ham_page_t *page;
    btree_node_t *node;
    ham_db_t *db=cursor_get_db(c);

    /*
     * uncoupled cursor: couple it
     */
    if (bt_cursor_get_flags(c)&BT_CURSOR_FLAG_UNCOUPLED) {
        st=bt_cursor_couple(c);
        if (st)
            return (st);
    }
    else if (!(bt_cursor_get_flags(c)&BT_CURSOR_FLAG_COUPLED))
        return (HAM_CURSOR_IS_NIL);

    db_set_error(db, 0);

    page=bt_cursor_get_coupled_page(c);
    node=ham_page_get_btree_node(page);

    /*
     * if the index+1 is till in the coupled page, just increment the
     * index
     */
    if (bt_cursor_get_coupled_index(c)+1<btree_node_get_count(node)) {
        bt_cursor_set_coupled_index(c, bt_cursor_get_coupled_index(c)+1);
        return (0);
    }

    /*
     * otherwise uncouple the cursor and load the right sibling page
     */
    page_remove_cursor(page, (ham_cursor_t *)c);
    bt_cursor_set_flags(c, bt_cursor_get_flags(c)&(~BT_CURSOR_FLAG_COUPLED));

    if (!btree_node_get_right(node))
        return db_set_error(db, HAM_KEY_NOT_FOUND);

    page=db_fetch_page(db, btree_node_get_right(node), 0);
    if (!page)
        return (db_get_error(db));

    /*
     * couple this cursor to the smallest key in this page
     */
    page_add_cursor(page, (ham_cursor_t *)c);
    bt_cursor_set_coupled_page(c, page);
    bt_cursor_set_coupled_index(c, 0);
    bt_cursor_set_flags(c,
            bt_cursor_get_flags(c)|BT_CURSOR_FLAG_COUPLED);

    return (0);
}

static ham_status_t
my_move_previous(ham_btree_t *be, ham_bt_cursor_t *c, ham_u32_t flags)
{
    ham_status_t st;
    ham_page_t *page;
    btree_node_t *node;
    ham_db_t *db=cursor_get_db(c);

    /*
     * uncoupled cursor: couple it
     */
    if (bt_cursor_get_flags(c)&BT_CURSOR_FLAG_UNCOUPLED) {
        st=bt_cursor_couple(c);
        if (st)
            return (st);
    }
    else if (!(bt_cursor_get_flags(c)&BT_CURSOR_FLAG_COUPLED))
        return (HAM_CURSOR_IS_NIL);


    db_set_error(db, 0);

    page=bt_cursor_get_coupled_page(c);
    node=ham_page_get_btree_node(page);

    /*
     * if the index-1 is till in the coupled page, just decrement the
     * index
     */
    if (bt_cursor_get_coupled_index(c)!=0) {
        bt_cursor_set_coupled_index(c, bt_cursor_get_coupled_index(c)-1);
        return (0);
    }

    /*
     * otherwise uncouple the cursor and load the left sibling page
     */
    page_remove_cursor(page, (ham_cursor_t *)c);
    bt_cursor_set_flags(c, bt_cursor_get_flags(c)&(~BT_CURSOR_FLAG_COUPLED));

    if (!btree_node_get_left(node))
        return db_set_error(db, HAM_KEY_NOT_FOUND);

    page=db_fetch_page(db, btree_node_get_left(node), 0);
    if (!page)
        return (db_get_error(db));
    node=ham_page_get_btree_node(page);

    /*
     * couple this cursor to the highest key in this page
     */
    page_add_cursor(page, (ham_cursor_t *)c);
    bt_cursor_set_coupled_page(c, page);
    bt_cursor_set_coupled_index(c, btree_node_get_count(node)-1);
    bt_cursor_set_flags(c,
            bt_cursor_get_flags(c)|BT_CURSOR_FLAG_COUPLED);

    return (0);
}

ham_status_t
bt_cursor_couple(ham_bt_cursor_t *c)
{
    ham_key_t key;
    ham_status_t st;
    ham_db_t *db=cursor_get_db(c);
    ham_txn_t txn;
    ham_bool_t local_txn=db_get_txn(db) ? HAM_FALSE : HAM_TRUE;

    ham_assert(bt_cursor_get_flags(c)&BT_CURSOR_FLAG_UNCOUPLED,
            ("coupling a cursor which is not uncoupled"));

    if (local_txn) {
        st=ham_txn_begin(&txn, db);
        if (st)
            return (st);
    }

    /*
     * make a 'find' on the cached key; if we succeed, the cursor
     * is automatically coupled
     */
    memset(&key, 0, sizeof(key));

    if (!util_copy_key(db, bt_cursor_get_uncoupled_key(c), &key)) {
        if (local_txn)
            (void)ham_txn_abort(&txn);
        return (db_get_error(db));
    }
    
    st=bt_cursor_find(c, &key, 0);
    if (st) {
        if (local_txn)
            (void)ham_txn_abort(&txn);
        return (st);
    }

    /*
     * free the cached key
     */
    if (key.data)
        ham_mem_free(db, key.data);

    if (local_txn)
        return (ham_txn_commit(&txn, 0));

    return (0);
}

ham_status_t
bt_cursor_uncouple(ham_bt_cursor_t *c, ham_u32_t flags)
{
    ham_status_t st=0;
    btree_node_t *node;
    int_key_t *entry;
    ham_key_t *key;
    ham_db_t *db=bt_cursor_get_db(c);
    ham_txn_t txn;
    ham_bool_t local_txn=db_get_txn(db) ? HAM_FALSE : HAM_TRUE;

    if (bt_cursor_get_flags(c)&BT_CURSOR_FLAG_UNCOUPLED)
        return (0);
    if (!(bt_cursor_get_flags(c)&BT_CURSOR_FLAG_COUPLED))
        return (0);

    ham_assert(bt_cursor_get_coupled_page(c)!=0,
            ("uncoupling a cursor which has no coupled page"));

    if (local_txn) {
        st=ham_txn_begin(&txn, db);
        if (st)
            return (st);
    }

    /*
     * get the btree-entry of this key
     */
    node=ham_page_get_btree_node(bt_cursor_get_coupled_page(c));
    ham_assert(btree_node_is_leaf(node), ("iterator points to internal node"));
    entry=btree_node_get_key(db, node, bt_cursor_get_coupled_index(c));

    /*
     * copy the key
     */
    key=(ham_key_t *)ham_mem_alloc(db, sizeof(*key));
    if (!key) {
        if (local_txn)
            (void)ham_txn_abort(&txn);
        return (db_set_error(db, HAM_OUT_OF_MEMORY));
    }
    memset(key, 0, sizeof(*key));
    key=util_copy_key_int2pub(db, entry, key);
    if (!key) {
        if (local_txn)
            (void)ham_txn_abort(&txn);
        return (db_get_error(bt_cursor_get_db(c)));
    }

    /*
     * uncouple the page
     */
    if (!(flags&BT_CURSOR_UNCOUPLE_NO_REMOVE))
        page_remove_cursor(bt_cursor_get_coupled_page(c),
                (ham_cursor_t *)c);

    /*
     * set the flags and the uncoupled key
     */
    bt_cursor_set_flags(c, bt_cursor_get_flags(c)&(~BT_CURSOR_FLAG_COUPLED));
    bt_cursor_set_flags(c, bt_cursor_get_flags(c)|BT_CURSOR_FLAG_UNCOUPLED);
    bt_cursor_set_uncoupled_key(c, key);

    if (local_txn)
        return (ham_txn_commit(&txn, 0));

    return (0);
}

ham_status_t
bt_cursor_create(ham_db_t *db, ham_txn_t *txn, ham_u32_t flags,
            ham_bt_cursor_t **cu)
{
    ham_bt_cursor_t *c;

    (void)flags;

    *cu=0;

    c=(ham_bt_cursor_t *)ham_mem_alloc(db, sizeof(*c));
    if (!c)
        return (db_set_error(db, HAM_OUT_OF_MEMORY));

    memset(c, 0, sizeof(*c));
    c->_fun_clone=bt_cursor_clone;
    c->_fun_close=bt_cursor_close;
    c->_fun_replace=bt_cursor_replace;
    c->_fun_move=bt_cursor_move;
    c->_fun_find=bt_cursor_find;
    c->_fun_insert=bt_cursor_insert;
    c->_fun_erase=bt_cursor_erase;
    bt_cursor_set_db(c, db);
    bt_cursor_set_txn(c, txn);

    *cu=c;
    return (0);
}

ham_status_t
bt_cursor_clone(ham_bt_cursor_t *oldcu, ham_bt_cursor_t **newcu)
{
    ham_status_t st=0;
    ham_bt_cursor_t *c;
    ham_db_t *db=bt_cursor_get_db(oldcu);
    ham_txn_t txn;
    ham_bool_t local_txn=db_get_txn(db) ? HAM_FALSE : HAM_TRUE;

    *newcu=0;

    c=(ham_bt_cursor_t *)ham_mem_alloc(db, sizeof(*c));
    if (!c)
        return (db_set_error(db, HAM_OUT_OF_MEMORY));
    memcpy(c, oldcu, sizeof(*c));
    cursor_set_next(c, 0);
    cursor_set_previous(c, 0);

    if (local_txn) {
        st=ham_txn_begin(&txn, db);