btree_erase.c

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

        if (btree_node_is_leaf(node))
            return (0);
        else 
            return (db_fetch_page(page_get_owner(page), 
                        btree_node_get_ptr_left(node), 0));
    }

    /*
     * if one of the siblings is missing, or both of them are 
     * too empty, we have to merge them
     */
    if ((!leftpage || fewleft) && (!rightpage || fewright)) {
        if (lanchor!=page_get_self(parent)) {
            return (my_merge_pages(page, rightpage, ranchor, scratchpad));
        }
        else {
            return (my_merge_pages(leftpage, page, lanchor, scratchpad));
        }
    }

    /*
     * otherwise choose the better of a merge or a shift
     */
    if (leftpage && fewleft && rightpage && !fewright) {
        if (!(ranchor==page_get_self(parent)) && 
                (page_get_self(page)==page_get_self(scratchpad->mergepage))) {
            return (my_merge_pages(leftpage, page, lanchor, scratchpad));
        }
        else {
            return (my_shift_pages(page, rightpage, ranchor, scratchpad));
        }
    }

    /*
     * ... still choose the better of a merge or a shift...
     */
    if (leftpage && !fewleft && rightpage && fewright) {
        if (!(lanchor==page_get_self(parent)) &&
                (page_get_self(page)==page_get_self(scratchpad->mergepage))) {
            return (my_merge_pages(page, rightpage, ranchor, scratchpad));
        }
        else {
            return (my_shift_pages(leftpage, page, lanchor, scratchpad));
        }
    }

    /*
     * choose the more effective of two shifts
     */
    if (lanchor==ranchor) {
        if (btree_node_get_count(leftnode)<=btree_node_get_count(rightnode)) {
            return (my_shift_pages(page, rightpage, ranchor, scratchpad));
        }
        else {
            return (my_shift_pages(leftpage, page, lanchor, scratchpad));
        }
    }

    /*
     * choose the shift with more local effect
     */
    if (lanchor==page_get_self(parent)) {
        return (my_shift_pages(leftpage, page, lanchor, scratchpad));
    }
    else {
        return (my_shift_pages(page, rightpage, ranchor, scratchpad));
    }
}

static ham_page_t *
my_merge_pages(ham_page_t *page, ham_page_t *sibpage, ham_offset_t anchor, 
        erase_scratchpad_t *scratchpad)
{
    ham_status_t st;
    ham_s32_t slot;
    ham_size_t c, keysize;
    ham_db_t *db=page_get_owner(page);
    ham_page_t *ancpage;
    btree_node_t *node, *sibnode, *ancnode;
    int_key_t *bte_lhs, *bte_rhs;

    keysize=db_get_keysize(db);
    node   =ham_page_get_btree_node(page);
    sibnode=ham_page_get_btree_node(sibpage);

    if (anchor) {
        ancpage=db_fetch_page(db, anchor, 0);
        ancnode=ham_page_get_btree_node(ancpage);
    }
    else {
        ancpage=0;
        ancnode=0;
    }

    /*
     * uncouple all cursors
     */
    if ((st=db_uncouple_all_cursors(page)))
        return (0);
    if ((st=db_uncouple_all_cursors(sibpage)))
        return (0);
    if (ancpage)
        if ((st=db_uncouple_all_cursors(ancpage)))
            return (0);

    /*
     * internal node: append the anchornode separator value to 
     * this node
     */
    if (!btree_node_is_leaf(node)) {
        int_key_t *bte;
        ham_key_t key; 

        bte =btree_node_get_key(db, sibnode, 0);
        memset(&key, 0, sizeof(key));
        key._flags=key_get_flags(bte);
        key.data  =key_get_key(bte);
        key.size  =key_get_size(bte);

        st=btree_get_slot(db, ancpage, &key, &slot);
        if (st) {
            db_set_error(db, st);
            return 0;
        }

        bte_lhs=btree_node_get_key(db, node,
            btree_node_get_count(node));
        bte_rhs=btree_node_get_key(db, ancnode, slot);

        st=my_copy_key(db, bte_lhs, bte_rhs);
        if (st) {
            db_set_error(db, st);
            return 0;
        }
        key_set_ptr(bte_lhs, btree_node_get_ptr_left(sibnode));
        btree_node_set_count(node, btree_node_get_count(node)+1);
    }

    c=btree_node_get_count(sibnode);
    bte_lhs=btree_node_get_key(db, node, 
            btree_node_get_count(node));
    bte_rhs=btree_node_get_key(db, sibnode, 0);

    /*
     * shift items from the sibling to this page
     */
    memcpy(bte_lhs, bte_rhs, (sizeof(int_key_t)-1+keysize)*c);
            
    page_set_dirty(page, 1);
    page_set_dirty(sibpage, 1);
    btree_node_set_count(node, btree_node_get_count(node)+c);
    btree_node_set_count(sibnode, 0);

    /*
     * update the linked list of pages
     */
    if (btree_node_get_left(node)==page_get_self(sibpage)) {
        if (btree_node_get_left(sibnode)) {
            ham_page_t *p=db_fetch_page(db, 
                    btree_node_get_left(sibnode), 0);
            btree_node_t *n=ham_page_get_btree_node(p);
            btree_node_set_right(n, btree_node_get_right(sibnode));
            btree_node_set_left(node, btree_node_get_left(sibnode));
            page_set_dirty(p, 1);
        }
        else
            btree_node_set_left(node, 0);
    }
    else if (btree_node_get_right(node)==page_get_self(sibpage)) {
        if (btree_node_get_right(sibnode)) {
            ham_page_t *p=db_fetch_page(db, 
                    btree_node_get_right(sibnode), 0);
            btree_node_t *n=ham_page_get_btree_node(p);

            btree_node_set_right(node, btree_node_get_right(sibnode));
            btree_node_set_left(n, btree_node_get_left(sibnode));
            page_set_dirty(p, 1);
        }
        else
            btree_node_set_right(node, 0);
    }
    
    /*
     * return this page for deletion
     */
    if (scratchpad->mergepage && 
           (page_get_self(scratchpad->mergepage)==page_get_self(page) ||
            page_get_self(scratchpad->mergepage)==page_get_self(sibpage))) 
        scratchpad->mergepage=0;

    /* 
     * delete the page
     */
    st=txn_free_page(db_get_txn(db), sibpage);
    if (st) {
        db_set_error(db, st);
        return (0);
    }

    return (sibpage);
}

static ham_page_t *
my_shift_pages(ham_page_t *page, ham_page_t *sibpage, ham_offset_t anchor,
        erase_scratchpad_t *scratchpad)
{
    ham_s32_t slot=0;
    ham_status_t st;
    ham_bool_t intern;
    ham_size_t i, s, c, keysize;
    ham_db_t *db=page_get_owner(page);
    ham_page_t *ancpage;
    btree_node_t *node, *sibnode, *ancnode;
    int_key_t *bte_lhs, *bte_rhs;

    node   =ham_page_get_btree_node(page);
    sibnode=ham_page_get_btree_node(sibpage);
    if (btree_node_get_count(node)==btree_node_get_count(sibnode))
        return (0);
    keysize=db_get_keysize(db);
    intern =!btree_node_is_leaf(node);
    ancpage=db_fetch_page(db, anchor, 0);
    ancnode=ham_page_get_btree_node(ancpage);

    /*
     * uncouple all cursors
     */
    if ((st=db_uncouple_all_cursors(page)))
        return (0);
    if ((st=db_uncouple_all_cursors(sibpage)))
        return (0);
    if (ancpage)
        if ((st=db_uncouple_all_cursors(ancpage)))
            return (0);

    /*
     * shift from sibling to this node
     */
    if (btree_node_get_count(sibnode)>=btree_node_get_count(node)) {
        /*
         * internal node: insert the anchornode separator value to 
         * this node
         */
        if (intern) {
            int_key_t *bte;
            ham_key_t key;

            bte=btree_node_get_key(db, sibnode, 0);
            memset(&key, 0, sizeof(key));
            key._flags=key_get_flags(bte);
            key.data  =key_get_key(bte);
            key.size  =key_get_size(bte);
            st=btree_get_slot(db, ancpage, &key, &slot);
            if (st) {
                db_set_error(db, st);
                return 0;
            }
    
            /*
             * append the anchor node to the page
             */
            bte_rhs=btree_node_get_key(db, ancnode, slot);
            bte_lhs=btree_node_get_key(db, node,
                btree_node_get_count(node));

            st=my_copy_key(db, bte_lhs, bte_rhs);
            if (st) {
                db_set_error(db, st);
                return 0;
            }

            /*
             * the pointer of this new node is ptr_left of the sibling
             */
            key_set_ptr(bte_lhs, btree_node_get_ptr_left(sibnode));

            /*
             * new pointer left of the sibling is sibling[0].ptr
             */
            btree_node_set_ptr_left(sibnode, key_get_ptr(bte));

            /*
             * update the anchor node with sibling[0]
             */
            (void)my_replace_key(ancpage, slot, 
                    bte, INTERNAL_KEY);

            /*
             * shift the whole sibling to the left
             */
            for (i=0; i<(ham_size_t)btree_node_get_count(sibnode)-1; i++) {
                bte_lhs=btree_node_get_key(db, sibnode, i);
                bte_rhs=btree_node_get_key(db, sibnode, i+1);
                memcpy(bte_lhs, bte_rhs, sizeof(int_key_t)-1+keysize);
            }

            /*
             * adjust counters
             */
            btree_node_set_count(node, btree_node_get_count(node)+1);
            btree_node_set_count(sibnode, btree_node_get_count(sibnode)-1);
        }

        c=(btree_node_get_count(sibnode)-btree_node_get_count(node))/2;
        if (c==0)
            goto cleanup;
        if (intern)
            c--;

        /*
         * internal node: append the anchor key to the page 
         */
        if (intern) {
            bte_lhs=btree_node_get_key(db, node, 
                    btree_node_get_count(node));
            bte_rhs=btree_node_get_key(db, ancnode, slot);

            st=my_copy_key(db, bte_lhs, bte_rhs);
            if (st) {
                db_set_error(db, st);
                return (0);
            }

            key_set_ptr(bte_lhs, btree_node_get_ptr_left(sibnode));
            btree_node_set_count(node, btree_node_get_count(node)+1);
        }

        /*
         * shift items from the sibling to this page, then 
         * delete the shifted items
         */
        bte_lhs=btree_node_get_key(db, node, 
                btree_node_get_count(node));
        bte_rhs=btree_node_get_key(db, sibnode, 0);

        memmove(bte_lhs, bte_rhs, (sizeof(int_key_t)-1+keysize)*c);

        bte_lhs=btree_node_get_key(db, sibnode, 0);
        bte_rhs=btree_node_get_key(db, sibnode, c);
        memmove(bte_lhs, bte_rhs, (sizeof(int_key_t)-1+keysize)*
                (btree_node_get_count(sibnode)-c));

        /*
         * internal nodes: don't forget to set ptr_left of the sibling, and
         * replace the anchor key
         */
        if (intern) {
            int_key_t *bte;
            bte=btree_node_get_key(db, sibnode, 0);
            btree_node_set_ptr_left(sibnode, key_get_ptr(bte));
            if (anchor) {
                ham_key_t key;
                memset(&key, 0, sizeof(key));
                key._flags=key_get_flags(bte);
                key.data  =key_get_key(bte);
                key.size  =key_get_size(bte);
                st=btree_get_slot(db, ancpage, &key, &slot);
                if (st) {
                    db_set_error(db, st);
                    return (0);
                }
                /* don't replace if the slot is outside of the key range */
#if 0
                if (slot<btree_node_get_count(ancnode)-1) {
#endif
                    st=my_replace_key(ancpage, slot, 
                        bte, INTERNAL_KEY);
                    if (st) {
                        db_set_error(db, st);
                        return (0);
                    }
/*}*/
            }
            /*
             * shift once more
             */
            bte_lhs=btree_node_get_key(db, sibnode, 0);
            bte_rhs=btree_node_get_key(db, sibnode, 1);
            memmove(bte_lhs, bte_rhs, (sizeof(int_key_t)-1+keysize)*
                    (btree_node_get_count(sibnode)-1));
        }
        /*
         * in a leaf - update the anchor
         */
        else {
            ham_key_t key;
            int_key_t *bte;
            bte=btree_node_get_key(db, sibnode, 0);
            memset(&key, 0, sizeof(key));