btree_erase.c

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

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

        /*
         * update the page counter
         */
        btree_node_set_count(node, 
                btree_node_get_count(node)+c);
        btree_node_set_count(sibnode, 
                btree_node_get_count(sibnode)-c-(intern ? 1 : 0));
    }
    /*
     * shift from this node to the sibling
     */
    else {
        /*
        * internal node: insert the anchornode separator value to 
        * this node
        */
        if (intern) {
            ham_size_t i;
            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;
            }

            /*
             * shift sibling by 1 to the right 
             */
            for (i=btree_node_get_count(sibnode); i>0; 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);
            }

            /*
             * copy the old anchor element to sibling[0]
             */
            bte_lhs=btree_node_get_key(db, sibnode, 0);
            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);
            }

            /*
             * sibling[0].ptr = sibling.ptr_left
             */
            key_set_ptr(bte_lhs, btree_node_get_ptr_left(sibnode));

            /*
             * sibling.ptr_left = node[node.count-1].ptr
             */
            bte_lhs=btree_node_get_key(db, node, 
            btree_node_get_count(node)-1);
            btree_node_set_ptr_left(sibnode, key_get_ptr(bte_lhs));

            /*
             * new anchor element is node[node.count-1].key
             */
            st=my_replace_key(ancpage, slot, bte_lhs, 
                    NOFLUSH|INTERNAL_KEY);
            if (st) {
                db_set_error(db, st);
                return (0);
            }

            /*
             * page: one item less; sibling: one item more
             */
            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(node)-btree_node_get_count(sibnode))/2;
        if (c==0)
            goto cleanup;
        if (intern)
            c--;

        /*
         * internal pages: insert the anchor element
         */
        if (intern) {
            ham_size_t i;

            /*
             * shift sibling by 1 to the right 
             */
            for (i=btree_node_get_count(sibnode); i>0; 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);
            }

            bte_lhs=btree_node_get_key(db, sibnode, 0);
            bte_rhs=btree_node_get_key(db, ancnode, slot);

            st=my_replace_key(sibpage, 0, bte_rhs, 
                    NOFLUSH|(btree_node_is_leaf(node)?0:INTERNAL_KEY));
            if (st) {
                db_set_error(db, st);
                return (0);
            }

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

        s=btree_node_get_count(node)-c-1;

        /*
         * shift items from this page to the sibling, then delete the
         * items from this page
         */
        bte_lhs=btree_node_get_key(db, sibnode, c);
        bte_rhs=btree_node_get_key(db, sibnode, 0);
        memmove(bte_lhs, bte_rhs, (sizeof(int_key_t)-1+keysize)*
                btree_node_get_count(sibnode));

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

        btree_node_set_count(node,
                btree_node_get_count(node)-c);
        btree_node_set_count(sibnode, 
                btree_node_get_count(sibnode)+c);

        /*
         * internal nodes: the pointer of the highest item
         * in the node will become the ptr_left of the sibling
         */
        if (intern) {
            bte_lhs=btree_node_get_key(db, node, 
                    btree_node_get_count(node)-1);
            btree_node_set_ptr_left(sibnode, key_get_ptr(bte_lhs));

            /*
             * free the extended blob of this key
             */
            if (key_get_flags(bte_lhs)&KEY_IS_EXTENDED) {
                ham_offset_t blobid=key_get_extended_rid(db, bte_lhs);
                ham_assert(blobid, (""));
                if (db_get_extkey_cache(db))
                    (void)extkey_cache_remove(db_get_extkey_cache(db), blobid);

                st=blob_free(db, blobid, 0);
                if (st)
                    return (0);
            }
            btree_node_set_count(node, btree_node_get_count(node)-1);
        }

        /*
         * replace the old anchor key with the new anchor key 
         */
        if (anchor) {
            int_key_t *bte;
            ham_key_t key;
            memset(&key, 0, sizeof(key));

            if (intern)
                bte =btree_node_get_key(db, node, s);
            else
                bte =btree_node_get_key(db, sibnode, 0);

            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);
            }

            st=my_replace_key(ancpage, slot+1, bte, INTERNAL_KEY);
            if (st) {
                db_set_error(db, st);
                return (0);
            }
        }
    }

cleanup:
    /*
     * mark pages as dirty
     */
    page_set_dirty(page, 1);
    page_set_dirty(ancpage, 1);
    page_set_dirty(sibpage, 1);

    scratchpad->mergepage=0;

    return (0);
}

static ham_status_t
my_copy_key(ham_db_t *db, int_key_t *lhs, int_key_t *rhs)
{
    memcpy(lhs, rhs, sizeof(int_key_t)-1+db_get_keysize(db));

    /*
     * if the key is extended, we copy the extended blob; otherwise, we'd
     * have to add reference counting to the blob, because two keys are now 
     * using the same blobid. this would be too complicated.
     */
    if (key_get_flags(rhs)&KEY_IS_EXTENDED) {
        ham_status_t st;
        ham_record_t record;
        ham_offset_t rhsblobid, lhsblobid;

        memset(&record, 0, sizeof(record));

        rhsblobid=key_get_extended_rid(db, rhs);
        st=blob_read(db, rhsblobid, &record, 0);
        if (st)
            return (st);

        st=blob_allocate(db, record.data, record.size, 0, &lhsblobid);
        if (st)
            return (st);
        key_set_extended_rid(db, lhs, lhsblobid);
    }

    return (0);
}

static ham_status_t
my_replace_key(ham_page_t *page, ham_s32_t slot, 
        int_key_t *rhs, ham_u32_t flags)
{
    int_key_t *lhs;
    ham_status_t st;
    ham_db_t *db=page_get_owner(page);
    btree_node_t *node=ham_page_get_btree_node(page);

    /*
     * uncouple all cursors
     */
    if ((st=db_uncouple_all_cursors(page)))
        return (db_set_error(db, st));

    lhs=btree_node_get_key(db, node, slot);

    /* 
     * if we overwrite an extended key: delete the existing extended blob
     */
    if (key_get_flags(lhs)&KEY_IS_EXTENDED) {
        ham_offset_t blobid=key_get_extended_rid(db, lhs);
        ham_assert(blobid, (""));

        st=blob_free(db, blobid, 0);
        if (st)
            return (st);

        /* remove the cached extended key */
        if (db_get_extkey_cache(db)) 
            (void)extkey_cache_remove(db_get_extkey_cache(db), blobid);
    }

    key_set_flags(lhs, key_get_flags(rhs));
    memcpy(key_get_key(lhs), key_get_key(rhs), 
            db_get_keysize(db));

    /*
     * internal keys are not allowed to have blob-flags, because only the
     * leaf-node can manage the blob. Therefore we have to disable those 
     * flags if we modify an internal key.
     */
    if (flags&INTERNAL_KEY)
        key_set_flags(lhs, key_get_flags(lhs)&
                ~(KEY_BLOB_SIZE_TINY|KEY_BLOB_SIZE_SMALL|KEY_BLOB_SIZE_EMPTY));

    /*
     * if this key is extended, we copy the extended blob; otherwise, we'd
     * have to add reference counting to the blob, because two keys are now 
     * using the same blobid. this would be too complicated.
     */
    if (key_get_flags(rhs)&KEY_IS_EXTENDED) {
        ham_status_t st;
        ham_record_t record;
        ham_offset_t rhsblobid, lhsblobid;

        memset(&record, 0, sizeof(record));

        rhsblobid=key_get_extended_rid(db, rhs);
        st=blob_read(db, rhsblobid, &record, 0);
        if (st)
            return (st);

        st=blob_allocate(db, record.data, record.size, 0, &lhsblobid);
        if (st)
            return (st);
        key_set_extended_rid(db, lhs, lhsblobid);
    }

    key_set_size(lhs, key_get_size(rhs));

    page_set_dirty(page, 1);

    return (HAM_SUCCESS);
}

static ham_status_t
my_remove_entry(ham_page_t *page, ham_s32_t slot, 
        erase_scratchpad_t *scratchpad)
{
    ham_status_t st;
    int_key_t *bte_lhs, *bte_rhs, *bte;
    btree_node_t *node;
    ham_size_t keysize;
    ham_db_t *db;

    db=page_get_owner(page);
    node=ham_page_get_btree_node(page);
    keysize=db_get_keysize(db);
    bte=btree_node_get_key(db, node, slot);

    /*
     * uncouple all cursors
     */
    if ((st=db_uncouple_all_cursors(page)))
        return (db_set_error(db, st));

    ham_assert(slot>=0, ("invalid slot %ld", slot));
    ham_assert(slot<btree_node_get_count(node), ("invalid slot %ld", slot));

    /*
     * get rid of the extended key (if there is one)
     *
     * also remove the key from the cache
     */
    if (key_get_flags(bte)&KEY_IS_EXTENDED) {
        ham_offset_t blobid=key_get_extended_rid(db, bte);
        (void)blob_free(db, blobid, 0); 

        /* remove the cached extended key */
        if (db_get_extkey_cache(db)) 
            (void)extkey_cache_remove(db_get_extkey_cache(db), blobid);
    }

    /*
     * if we delete the last item, it's enough to decrement the item 
     * counter and return...
     */
    if (slot!=btree_node_get_count(node)-1) {
        bte_lhs=btree_node_get_key(db, node, slot);
        bte_rhs=btree_node_get_key(db, node, slot+1);
        memmove(bte_lhs, bte_rhs, ((sizeof(int_key_t)-1+keysize))*
                (btree_node_get_count(node)-slot-1));
    }

    btree_node_set_count(node, btree_node_get_count(node)-1);

    page_set_dirty(page, 1);

    return (0);
}