btree_cursor.c

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

        if (st)
            return (st);
    }

    /*
     * if the old cursor is coupled: couple the new cursor, too
     */
    if (bt_cursor_get_flags(oldcu)&BT_CURSOR_FLAG_COUPLED) {
         ham_page_t *page=bt_cursor_get_coupled_page(oldcu);
         page_add_cursor(page, (ham_cursor_t *)c);
         bt_cursor_set_coupled_page(c, page);
    }
    /*
     * if the old cursor is uncoupled: copy the key
     */
    else if (bt_cursor_get_flags(oldcu)&BT_CURSOR_FLAG_UNCOUPLED) {
        ham_key_t *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));

        if (!util_copy_key(bt_cursor_get_db(c), 
                bt_cursor_get_uncoupled_key(oldcu), key)) {
            if (key->data)
                ham_mem_free(db, key->data);
            ham_mem_free(db, key);
            if (local_txn)
                (void)ham_txn_abort(&txn);
            return (db_get_error(bt_cursor_get_db(c)));
        }
        bt_cursor_set_uncoupled_key(c, key);
    }

    *newcu=c;

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

    return (0);
}

ham_status_t
bt_cursor_close(ham_bt_cursor_t *c)
{
    (void)my_set_to_nil(c);

    ham_mem_free(cursor_get_db(c), c);

    return (0);
}

ham_status_t
bt_cursor_replace(ham_bt_cursor_t *c, ham_record_t *record,
            ham_u32_t flags)
{
    ham_status_t st;
    btree_node_t *node;
    int_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;
    ham_u32_t oldflags;
    ham_page_t *page;

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

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

    /*
     * make sure that the page is not unloaded
     */
    page=bt_cursor_get_coupled_page(c);
    page_add_ref(page);

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

    /*
     * copy the key flags, and remove all flags concerning the key size
     */
    oldflags=key_get_flags(key);
    key_set_flags(key, key_get_flags(key)&(~KEY_BLOB_SIZE_TINY));
    key_set_flags(key, key_get_flags(key)&(~KEY_BLOB_SIZE_SMALL));
    key_set_flags(key, key_get_flags(key)&(~KEY_BLOB_SIZE_EMPTY));

    /*
     * delete the record?
     */
    if (record->size==0) {
        if (!((oldflags&KEY_BLOB_SIZE_TINY) ||
              (oldflags&KEY_BLOB_SIZE_SMALL) ||
              (oldflags&KEY_BLOB_SIZE_EMPTY))) {
            /* remove the cached extended key */
            if (db_get_extkey_cache(db))
                (void)extkey_cache_remove(db_get_extkey_cache(db),
                        key_get_ptr(key));
            (void)blob_free(db, key_get_ptr(key), 0);
        }

        key_set_ptr(key, 0);
        key_set_flags(key, key_get_flags(key)|KEY_BLOB_SIZE_EMPTY);
    }

    /*
     * or replace with a big record?
     */
    else if (record->size>sizeof(ham_offset_t)) {
        ham_offset_t rid;

        /*
         * make sure that we only call blob_replace(), if there IS a blob
         * to replace! otherwise call blob_allocate()
         */
        if (!((oldflags&KEY_BLOB_SIZE_TINY) ||
              (oldflags&KEY_BLOB_SIZE_SMALL) ||
              (oldflags&KEY_BLOB_SIZE_EMPTY))) {
            /* remove the cached extended key */
            if (db_get_extkey_cache(db))
                (void)extkey_cache_remove(db_get_extkey_cache(db),
                        key_get_ptr(key));
            st=blob_replace(db, key_get_ptr(key), record->data,
                        record->size, 0, &rid);
        }
        else
            st=blob_allocate(db, record->data, record->size, 0, &rid);
        if (st) {
            page_release_ref(page);
            if (local_txn)
                (void)ham_txn_abort(&txn);
            return (st);
        }

        key_set_ptr(key, rid);
    }

    /*
     * or replace with a small record?
     */
    else {
        ham_offset_t rid;

        if (!((oldflags&KEY_BLOB_SIZE_TINY) ||
              (oldflags&KEY_BLOB_SIZE_SMALL) ||
              (oldflags&KEY_BLOB_SIZE_EMPTY))) {
            /* remove the cached extended key */
            if (db_get_extkey_cache(db))
                (void)extkey_cache_remove(db_get_extkey_cache(db),
                        key_get_ptr(key));
            (void)blob_free(db, key_get_ptr(key), 0);
        }

        memcpy(&rid, record->data, record->size);
        if (record->size<sizeof(ham_offset_t)) {
            char *p=(char *)&rid;
            p[sizeof(ham_offset_t)-1]=record->size;
            key_set_flags(key, key_get_flags(key)|KEY_BLOB_SIZE_TINY);
        }
        else
            key_set_flags(key, key_get_flags(key)|KEY_BLOB_SIZE_SMALL);

        key_set_ptr(key, rid);
    }

    page_set_dirty(bt_cursor_get_coupled_page(c), 1);
    page_release_ref(page);

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

    return (0);
}

ham_status_t
bt_cursor_move(ham_bt_cursor_t *c, ham_key_t *key,
            ham_record_t *record, ham_u32_t flags)
{
    ham_status_t st=0;
    ham_page_t *page;
    btree_node_t *node;
    ham_db_t *db=bt_cursor_get_db(c);
    ham_btree_t *be=(ham_btree_t *)db_get_backend(db);
    int_key_t *entry;
    ham_txn_t txn;
    ham_bool_t local_txn=db_get_txn(db) ? HAM_FALSE : HAM_TRUE;

    if (!be)
        return (HAM_INV_INDEX);

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

    /*
     * if the cursor is NIL, and the user requests a NEXT, we set it to FIRST;
     * if the user requests a PREVIOUS, we set it to LAST, resp.
     */
    if (bt_cursor_is_nil(c)) {
        if (flags&HAM_CURSOR_NEXT) {
            flags&=~HAM_CURSOR_NEXT;
            flags|=HAM_CURSOR_FIRST;
        }
        else if (flags&HAM_CURSOR_PREVIOUS) {
            flags&=~HAM_CURSOR_PREVIOUS;
            flags|=HAM_CURSOR_LAST;
        }
    }

    if (flags&HAM_CURSOR_FIRST)
        st=my_move_first(be, c, flags);
    else if (flags&HAM_CURSOR_LAST)
        st=my_move_last(be, c, flags);
    else if (flags&HAM_CURSOR_NEXT)
        st=my_move_next(be, c, flags);
    else if (flags&HAM_CURSOR_PREVIOUS)
        st=my_move_previous(be, c, flags);
    /* no move, but cursor is nil? return error */
    else if (bt_cursor_is_nil(c)) {
        if (local_txn)
            (void)ham_txn_abort(&txn);
        if (key || record) 
            return (HAM_CURSOR_IS_NIL);
        else
            return (0);
    }
    if (st) {
        if (local_txn)
            (void)ham_txn_abort(&txn);
        return (st);
    }

    /*
     * during util_read_key and util_read_record, new pages might be needed,
     * and the page at which we're pointing could be moved out of memory; 
     * that would mean that the cursor would be uncoupled, and we're losing
     * the 'entry'-pointer. therefore we 'lock' the page by incrementing 
     * the reference counter
     */
    ham_assert(bt_cursor_get_flags(c)&BT_CURSOR_FLAG_COUPLED, 
            ("move: cursor is not coupled"));
    page=bt_cursor_get_coupled_page(c);
    page_add_ref(page);
    node=ham_page_get_btree_node(page);
    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));

    if (key) {
        st=util_read_key(db, entry, key, 0);
        if (st) {
            page_release_ref(page);
            if (local_txn)
                (void)ham_txn_abort(&txn);
            return (st);
        }
    }

    if (record) {
        record->_intflags=key_get_flags(entry);
		record->_rid=key_get_ptr(entry);
        st=util_read_record(db, record, 0);
        if (st) {
            page_release_ref(page);
            if (local_txn)
                (void)ham_txn_abort(&txn);
            return (st);
        }
    }

    page_release_ref(page);

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

    return (0);
}

ham_status_t
bt_cursor_find(ham_bt_cursor_t *c, ham_key_t *key, ham_u32_t flags)
{
    ham_status_t st;
    ham_backend_t *be=db_get_backend(bt_cursor_get_db(c));
    ham_txn_t txn;
    ham_db_t *db=cursor_get_db(c);
    ham_bool_t local_txn=db_get_txn(db) ? HAM_FALSE : HAM_TRUE;

    if (!be)
        return (HAM_INV_INDEX);
    if (!key)
        return (HAM_INV_PARAMETER);

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

    st=my_set_to_nil(c);
    if (st) {
        if (local_txn)
            (void)ham_txn_abort(&txn);
        return (st);
    }
    /*
    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));
    }
    */

    st=btree_find_cursor((ham_btree_t *)be, c, key, 0, flags);
    if (st) {
        /* cursor is now NIL */
        if (local_txn)
            (void)ham_txn_abort(&txn);
        return (st);
    }

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

    return (0);
}

ham_status_t
bt_cursor_insert(ham_bt_cursor_t *c, ham_key_t *key,
            ham_record_t *record, ham_u32_t flags)
{
    ham_status_t st;
    ham_db_t *db=bt_cursor_get_db(c);
    ham_btree_t *be=(ham_btree_t *)db_get_backend(db);
    ham_txn_t txn;
    ham_bool_t local_txn=db_get_txn(db) ? HAM_FALSE : HAM_TRUE;

    if (!be)
        return (HAM_INV_INDEX);
    if (!key)
        return (HAM_INV_PARAMETER);
    if (!record)
        return (HAM_INV_PARAMETER);

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

    /*
     * set the cursor to nil
     */
    st=my_set_to_nil(c);
    if (st) {
        if (local_txn)
            (void)ham_txn_abort(&txn);
        return (st);
    }

    /*
     * then call the btree insert function
     */
    st=btree_insert_cursor(be, key, record, c, flags);
    if (st) {
        if (local_txn)
            (void)ham_txn_abort(&txn);
        return (st);
    }

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

    return (0);
}

ham_status_t
bt_cursor_erase(ham_bt_cursor_t *c, ham_offset_t *rid,
            ham_u32_t *intflags, ham_u32_t flags)
{
    ham_status_t st;
    ham_db_t *db=bt_cursor_get_db(c);
    ham_btree_t *be=(ham_btree_t *)db_get_backend(db);
    ham_txn_t txn;
    ham_bool_t local_txn=db_get_txn(db) ? HAM_FALSE : HAM_TRUE;

    if (!be)
        return (HAM_INV_INDEX);

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

    /*
     * coupled cursor: uncouple it
     */
    if (bt_cursor_get_flags(c)&BT_CURSOR_FLAG_COUPLED) {
        st=bt_cursor_uncouple(c, 0);
        if (st)
            return (st);
    }
    else if (!(bt_cursor_get_flags(c)&BT_CURSOR_FLAG_UNCOUPLED))
        return (HAM_CURSOR_IS_NIL);

    st=btree_erase(be, bt_cursor_get_uncoupled_key(c), rid, intflags, flags);
    if (st) {
        if (local_txn)
            (void)ham_txn_abort(&txn);
        return (st);
    }

    /*
     * set cursor to nil
     */
    st=my_set_to_nil(c);
    if (st)
        return (st);

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

    return (0);
}