implement delete under brt cursor

git-svn-id: file:///svn/tokudb@291 c7de825b-a66e-492c-adef-691d508d4ae1

newbrt/brt-test.c

@@ -1548,6 +1548,68 @@ void test_brt_cursor_set_range(int n) {
     assert(r==0);
 }
 
+void test_brt_cursor_delete(int n) {
+    printf("test_brt_cursor_delete:%d\n", n);
+
+    int error;
+    char fname[]="testbrt.brt";
+    CACHETABLE ct;
+    BRT brt;
+    BRT_CURSOR cursor;
+
+    unlink(fname);
+
+    error = brt_create_cachetable(&ct, 0);
+    assert(error == 0);
+
+    error = open_brt(fname, 0, 1, &brt, 1<<12, ct, default_compare_fun);  
+    assert(error == 0);
+
+    error = brt_cursor(brt, &cursor);
+    assert(error == 0);
+
+    DBT key, val;
+    int k, v;
+
+    int i;
+    /* insert keys 0, 1, 2, .. (n-1) */
+    for (i=0; i<n; i++) {
+        k = htonl(i);
+        v = i;
+        fill_dbt(&key, &k, sizeof k);
+        fill_dbt(&val, &v, sizeof v);
+        error = brt_insert(brt, &key, &val, 0);
+        assert(error == 0);
+    }
+
+    /* walk the tree and delete under the cursor */
+    for (;;) {
+        init_dbt(&key); key.flags = DB_DBT_MALLOC;
+        init_dbt(&val); val.flags = DB_DBT_MALLOC;
+        error = brt_c_get(cursor, &key, &val, DB_NEXT);
+        if (error == DB_NOTFOUND)
+            break;
+        assert(error == 0);
+        toku_free(key.data);
+        toku_free(val.data);
+
+        error = brt_cursor_delete(cursor, 0);
+        assert(error == 0);
+    }
+
+    error = brt_cursor_delete(cursor, 0);
+    assert(error != 0);
+
+    error = brt_cursor_close(cursor);
+    assert(error == 0);
+
+    error = close_brt(brt);
+    assert(error == 0);
+
+    error = cachetable_close(&ct);
+    assert(error == 0);
+}
+
 int test_brt_cursor_inc = 1000;
 int test_brt_cursor_limit = 10000;
 
@@ -1588,6 +1650,7 @@ void test_brt_cursor() {
     test_brt_cursor_set(1000, DB_SET_RANGE); memory_check_all_free();
     test_brt_cursor_set_range(1000); memory_check_all_free();
     test_brt_cursor_set_range(10000); memory_check_all_free();
+    test_brt_cursor_delete(1000); memory_check_all_free();
 }
 
 void test_large_kv(int bsize, int ksize, int vsize) {
@@ -1848,6 +1911,7 @@ void test_brt_delete() {
 
 static void brt_blackbox_test (void) {
     memory_check = 1;
+
     test_wrongendian_compare(0, 2);          memory_check_all_free();
     test_wrongendian_compare(1, 2);          memory_check_all_free();
     test_wrongendian_compare(1, 257);        memory_check_all_free();

newbrt/brt.c

@@ -2390,3 +2390,19 @@ int brt_c_get (BRT_CURSOR cursor, DBT *kbt, DBT *vbt, int flags) {
     if ((r = unpin_brt_header(cursor->brt))!=0) return r;
     return 0;
 }
+
+int brt_cursor_delete(BRT_CURSOR cursor, int flags __attribute__((__unused__))) {
+    int r;
+
+    if (cursor->path_len > 0) {
+        BRTNODE node = cursor->path[cursor->path_len-1];
+        assert(node->height == 0);
+        int kvsize;
+        r = pma_cursor_delete_under(cursor->pmacurs, &kvsize);
+        if (r == 0)
+            node->u.l.n_bytes_in_buffer -= KEY_VALUE_OVERHEAD + kvsize;
+    } else
+        r = DB_NOTFOUND;
+
+    return r;
+}

newbrt/brt.h

@@ -31,6 +31,7 @@ int show_brt_blocknumbers(BRT);
 typedef struct brt_cursor *BRT_CURSOR;
 int brt_cursor (BRT, BRT_CURSOR*);
 int brt_c_get (BRT_CURSOR cursor, DBT *kbt, DBT *vbt, int brtc_flags);
+int brt_cursor_delete(BRT_CURSOR cursor, int flags);
 int brt_cursor_close (BRT_CURSOR curs);
 
 #endif

newbrt/pma.c

@@ -489,6 +489,23 @@ int pma_cursor_set_range(PMA_CURSOR c, DBT *key, DB *db) {
     return DB_NOTFOUND;
 }
 
+int pma_cursor_delete_under(PMA_CURSOR c, int *kvsize) {
+    int r = DB_NOTFOUND;
+
+    if (c->position >= 0) {
+        PMA pma = c->pma;
+        assert(c->position < pma->N);
+        struct kv_pair *kv = pma->pairs[c->position];
+        if (kv_pair_valid(kv)) {
+            if (kvsize) *kvsize = kv_pair_keylen(kv) + kv_pair_vallen(kv);
+            pma->pairs[c->position] = kv_pair_set_deleted(kv);
+            r = 0;
+        }
+    }
+
+    return r;
+}
+
 #if 0
 int pma_cget_first (PMA_CURSOR c, YBT *key, YBT *val) {
     PMA pma=c->pma;

newbrt/pma.h

@@ -90,21 +90,11 @@ int pma_cursor_set_key(PMA_CURSOR c, DBT *key, DB *db);
 /* set the cursor to the smallest key >= requested key */
 int pma_cursor_set_range(PMA_CURSOR c, DBT *key, DB *db);
 
-/*
- * Get the last key and value in the pma
- */
-int pma_get_last(PMA pma, DBT *key, DBT *val);
+/* delete the key under the cursor */
+int pma_cursor_delete_under(PMA_CURSOR c, int *kvsize);
 
-/* Return PMA_NOTFOUND if the pma is empty. */
-#if 0
-int pma_cget_first (PMA_CURSOR, YBT */*key*/, YBT */*val*/);
-int pma_cursor_first (PMA);
-int pma_cursor_last  (PMA);
-int pma_cursor_set (PMA, bytevec key, int keylen);
-int pma_cursor_next (PMA);
-int pma_cursor_prev (PMA);
-int pma_cursor_get (PMA, bytevec *key, int *keylen, bytevec *data, int *datalen);
-#endif
+/* get the last key and value in the pma */
+int pma_get_last(PMA pma, DBT *key, DBT *val);
 
 int pma_random_pick(PMA, bytevec *key, ITEMLEN *keylen, bytevec *data, ITEMLEN *datalen);