add brt cursor prev feature and test cases

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

newbrt/brt-internal.h

@@ -106,6 +106,7 @@ struct brt_cursor {
     int pathcnum[CURSOR_PATHLEN_LIMIT]; /* which child did we descend to from here? */
     PMA_CURSOR pmacurs; /* The cursor into the leaf.  NULL if the cursor doesn't exist. */
     BRT_CURSOR prev,next;
+    int op;
 };
 
 /* print the cursor path */

newbrt/brt-test.c

@@ -9,6 +9,7 @@
 #include <sys/time.h>
 #include <stdlib.h>
 #include <unistd.h>
+#include <arpa/inet.h>
 
 extern long long n_items_malloced;
 
@@ -1134,6 +1135,78 @@ void test_brt_cursor_walk(int n) {
     assert(r==0);
 }
 
+void assert_cursor_rwalk(BRT brt, int n) {
+    BRT_CURSOR cursor;
+    int i;
+    int r;
+
+    r = brt_cursor(brt, &cursor);
+    assert(r==0);
+
+    if (test_cursor_debug) printf("key: ");
+    for (i=n-1; ; i--) {
+        DBT kbt, vbt;
+        long long v;
+
+        init_dbt(&kbt); kbt.flags = DB_DBT_MALLOC;
+        init_dbt(&vbt); vbt.flags = DB_DBT_MALLOC;
+        r = brt_c_get(cursor, &kbt, &vbt, DB_PREV);
+        if (r != 0)
+            break;
+        if (test_cursor_debug) printf("%s ", (char*)kbt.data);
+        assert(vbt.size == sizeof v);
+        memcpy(&v, vbt.data, vbt.size);
+        assert(v == i);
+        toku_free(kbt.data);
+        toku_free(vbt.data);
+    }
+    if (test_cursor_debug) printf("\n");
+    assert(i == -1);
+
+    r = brt_cursor_close(cursor);
+    assert(r==0);
+}
+
+void test_brt_cursor_rwalk(int n) {
+    const char *fname="testbrt.brt";
+    CACHETABLE ct;
+    BRT brt;
+    int r;
+    int i;
+
+    printf("test_brt_cursor_rwalk:%d\n", n);
+
+    unlink(fname);
+
+    r = brt_create_cachetable(&ct, 0);
+    assert(r==0);
+
+    r = open_brt(fname, 0, 1, &brt, 1<<12, ct, default_compare_fun);  
+    assert(r==0);
+
+    /* insert a bunch of kv pairs */
+    for (i=0; i<n; i++) {
+        int k; long long v;
+        DBT kbt, vbt;
+
+        k = htonl(i);
+        fill_dbt(&kbt, &k, sizeof k);
+        v = i;
+        fill_dbt(&vbt, &v, sizeof v);
+        r = brt_insert(brt, &kbt, &vbt, 0);
+        assert(r==0);
+    }
+
+    /* walk the tree */
+    assert_cursor_rwalk(brt, n);
+
+    r = close_brt(brt);
+    assert(r==0);
+
+    r = cachetable_close(&ct);
+    assert(r==0);
+}
+
 void assert_cursor_walk_inorder(BRT brt, int n) {
     BRT_CURSOR cursor;
     int i;
@@ -1302,12 +1375,48 @@ void test_brt_cursor_split(int n) {
     assert(r==0);
 }
 
+void test_multiple_brt_cursors(int n) {
+    printf("test_multiple_brt_cursors:%d\n", n);
+
+    int r;
+    char fname[]="testbrt.brt";
+    CACHETABLE ct;
+    BRT brt;
+    BRT_CURSOR cursors[n];
+
+    unlink(fname);
+
+    r = brt_create_cachetable(&ct, 0);
+    assert(r==0);
+
+    r = open_brt(fname, 0, 1, &brt, 1<<12, ct, default_compare_fun);  
+    assert(r==0);
+
+    int i;
+    for (i=0; i<n; i++) {
+        r = brt_cursor(brt, &cursors[i]);
+        assert(r == 0);
+    }
+
+    for (i=0; i<n; i++) {
+        r = brt_cursor_close(cursors[i]);
+        assert(r == 0);
+    }
+
+    r = close_brt(brt);
+    assert(r==0);
+
+    r = cachetable_close(&ct);
+    assert(r==0);
+}
+
 void test_brt_cursor() {
     int n;
 
-    if (1) for (n=0; n<10000; n += 100) {
-        test_brt_cursor_last(n); memory_check_all_free();
-    }
+    test_multiple_brt_cursors(1);
+    test_multiple_brt_cursors(2);
+    test_multiple_brt_cursors(3);
+
     if (1) for (n=0; n<10000; n += 100) {
         test_brt_cursor_first(n); memory_check_all_free();
     }
@@ -1315,10 +1424,13 @@ void test_brt_cursor() {
         test_brt_cursor_rfirst(n); memory_check_all_free();
     }
     if (1) for (n=0; n<10000; n += 100) {
-        test_brt_cursor_first_last(n); memory_check_all_free();
+        test_brt_cursor_walk(n); memory_check_all_free();
     }
     if (1) for (n=0; n<10000; n += 100) {
-        test_brt_cursor_walk(n); memory_check_all_free();
+        test_brt_cursor_last(n); memory_check_all_free();
+    }
+    if (1) for (n=0; n<10000; n += 100) {
+        test_brt_cursor_first_last(n); memory_check_all_free();
     }
     if (1) for (n=0; n<10000; n += 100) {
         test_brt_cursor_split(n); memory_check_all_free();
@@ -1326,6 +1438,9 @@ void test_brt_cursor() {
     if (1) for (n=0; n<10000; n += 100) {
         test_brt_cursor_rand(n); memory_check_all_free();
     }
+    if (1) for (n=0; n<10000; n += 100) {
+        test_brt_cursor_rwalk(n); memory_check_all_free();
+    }
 }
 
 void test_large_kv(int bsize, int ksize, int vsize) {
@@ -1390,7 +1505,7 @@ void test_brt_delete_empty() {
     assert(r==0);
 
     DBT key;
-    int k = 1;
+    int k = htonl(1);
     fill_dbt(&key, &k, sizeof k);
     r = brt_delete(t, &key, 0);
     assert(r != 0);
@@ -1422,7 +1537,7 @@ void test_brt_delete_present(int n) {
     int k, v;
 
     for (i=0; i<n; i++) {
-        k = i; v = n + i;
+        k = htonl(i); v = i;
         fill_dbt(&key, &k, sizeof k);
         fill_dbt(&val, &v, sizeof v);
         r = brt_insert(t, &key, &val, 0);
@@ -1430,7 +1545,7 @@ void test_brt_delete_present(int n) {
     }
 
     for (i=0; i<n; i++) {
-        k = i;
+        k = htonl(i);
         fill_dbt(&key, &k, sizeof k);
         r = brt_delete(t, &key, 0);
         assert(r == 0);
@@ -1438,7 +1553,7 @@ void test_brt_delete_present(int n) {
 
     /* lookups should all fail */
     for (i=0; i<n; i++) {
-        k = i;
+        k = htonl(i);
         fill_dbt(&key, &k, sizeof k);
         init_dbt(&val); val.flags = DB_DBT_MALLOC;
         r = brt_lookup(t, &key, &val, 0);
@@ -1482,7 +1597,7 @@ void test_brt_delete_not_present(int n) {
     int k, v;
 
     for (i=0; i<n; i++) {
-        k = i; v = n + i;
+        k = htonl(i); v = i;
         fill_dbt(&key, &k, sizeof k);
         fill_dbt(&val, &v, sizeof v);
         r = brt_insert(t, &key, &val, 0);
@@ -1490,13 +1605,13 @@ void test_brt_delete_not_present(int n) {
     }
 
     for (i=0; i<n; i++) {
-        k = i;
+        k = htonl(i);
         fill_dbt(&key, &k, sizeof k);
         r = brt_delete(t, &key, 0);
         assert(r == 0);
     }
 
-    k = n+1;
+    k = htonl(n+1);
     fill_dbt(&key, &k, sizeof k);
     r = brt_delete(t, &key, 0);
     printf("brt_delete k=%d %d\n", k, r);
@@ -1524,7 +1639,7 @@ void test_brt_delete_cursor_first(int n) {
     int k, v;
 
     for (i=0; i<n; i++) {
-        k = i; v = ~i;
+        k = htonl(i); v = i;
         fill_dbt(&key, &k, sizeof k);
         fill_dbt(&val, &v, sizeof v);
         r = brt_insert(t, &key, &val, 0);
@@ -1532,7 +1647,7 @@ void test_brt_delete_cursor_first(int n) {
     }
 
     for (i=0; i<n-1; i++) {
-        k = i;
+        k = htonl(i);
         fill_dbt(&key, &k, sizeof k);
         r = brt_delete(t, &key, 0);
         assert(r == 0);
@@ -1540,7 +1655,7 @@ void test_brt_delete_cursor_first(int n) {
 
     /* lookups should all fail */
     for (i=0; i<n-1; i++) {
-        k = i;
+        k = htonl(i);
         fill_dbt(&key, &k, sizeof k);
         init_dbt(&val); val.flags = DB_DBT_MALLOC;
         r = brt_lookup(t, &key, &val, 0);
@@ -1560,7 +1675,7 @@ void test_brt_delete_cursor_first(int n) {
     int vv;
     assert(val.size == sizeof vv);
     memcpy(&vv, val.data, val.size);
-    assert(vv == ~(n-1));
+    assert(vv == n-1);
     toku_free(key.data);
     toku_free(val.data);
 

newbrt/brt.c

@@ -1804,13 +1804,18 @@ void brt_cursor_nonleaf_expand(BRT_CURSOR cursor, BRT t __attribute__((unused)),
         if (cursor->pathcnum[i] < childnum)
             return;
         if (cursor->pathcnum[i] > childnum) {
-            setnewchild:
+        setnewchild:
             oldchildnum = cursor->pathcnum[i];
             newchildnum = oldchildnum + 1;
             brt_node_remove_cursor(node, oldchildnum, cursor);
             brt_node_add_cursor(node, newchildnum, cursor);
             cursor->pathcnum[i] = newchildnum;
-        } else if (i+1 < cursor->path_len) {
+            return;
+        } 
+        if (i == cursor->path_len-1 && cursor->op == DB_PREV) {
+            goto setnewchild;
+        }
+        if (i+1 < cursor->path_len) {
             assert(cursor->path[i+1] == left || cursor->path[i+1] == right);
             if (cursor->path[i+1] == right) {
                 goto setnewchild;
@@ -2105,6 +2110,64 @@ int brtcurs_set_position_next (BRT_CURSOR cursor) {
     return 0;
 }
 
+int brtcurs_set_position_prev2(BRT_CURSOR cursor) {
+    BRTNODE node;
+    int childnum;
+    int r;
+    int more;
+
+    assert(cursor->path_len > 0);
+
+    /* pop the node and childnum from the cursor path */
+    node = cursor->path[cursor->path_len-1];
+    childnum = cursor->pathcnum[cursor->path_len-1];
+    cursor->path_len -= 1;
+    cachetable_unpin(cursor->brt->cf, node->thisnodename, 0);
+
+    if (brt_cursor_path_empty(cursor))
+        return DB_NOTFOUND;
+
+    /* set position last in the next left tree */
+    node = cursor->path[cursor->path_len-1];
+    childnum = cursor->pathcnum[cursor->path_len-1];
+    assert(node->height > 0);
+    brt_node_remove_cursor(node, childnum, cursor);
+
+    childnum -= 1;
+    while (childnum >= 0) {
+        cursor->pathcnum[cursor->path_len-1] = childnum;
+        brt_node_add_cursor(node, childnum, cursor);
+        for (;;) {
+            more = node->u.n.n_bytes_in_hashtable[childnum];
+            if (more == 0)
+                break;
+            brt_flush_child(cursor->brt, node, childnum, cursor);
+            node = cursor->path[cursor->path_len-1];
+            childnum = cursor->pathcnum[cursor->path_len-1];
+        }
+        r = brtcurs_set_position_last(cursor, node->u.n.children[childnum]);
+        if (r == 0)
+            return 0;
+        assert(node == cursor->path[cursor->path_len-1]);
+        brt_node_remove_cursor(node, childnum, cursor);
+        childnum -= 1;
+    }
+
+    return brtcurs_set_position_prev2(cursor);
+}
+
+int brtcurs_set_position_prev (BRT_CURSOR cursor) {
+    int r = pma_cursor_set_position_prev(cursor->pmacurs);
+    if (r==DB_NOTFOUND) {
+	if (cursor->path_len==1) 
+            return DB_NOTFOUND;
+        r = pma_cursor_free(&cursor->pmacurs);
+        assert(r == 0);
+        return brtcurs_set_position_prev2(cursor);
+    }
+    return 0;
+}
+
 static int unpin_cursor (BRT_CURSOR cursor) {
     BRT brt=cursor->brt;
     int i;
@@ -2157,12 +2220,13 @@ int brt_c_get (BRT_CURSOR cursor, DBT *kbt, DBT *vbt, int flags) {
 	do_rmw=1;
 	flags &= ~DB_RMW;
     }
+    cursor->op = flags;
     switch (flags) {
     case DB_LAST:
+    do_db_last:
 	r=unpin_cursor(cursor); if (r!=0) goto died0;
 	assert(cursor->pmacurs == 0);
         r=brtcurs_set_position_last(cursor, *rootp); if (r!=0) goto died0;
-	if (0) brt_cursor_print(cursor);
         r=pma_cget_current(cursor->pmacurs, kbt, vbt);
 	if (r == 0) assert_cursor_path(cursor);
         break;
@@ -2171,18 +2235,23 @@ int brt_c_get (BRT_CURSOR cursor, DBT *kbt, DBT *vbt, int flags) {
 	r=unpin_cursor(cursor); if (r!=0) goto died0;
 	assert(cursor->pmacurs == 0);
         r=brtcurs_set_position_first(cursor, *rootp); if (r!=0) goto died0;
-	if (0) brt_cursor_print(cursor);
         r=pma_cget_current(cursor->pmacurs, kbt, vbt);
 	if (r == 0) assert_cursor_path(cursor);
         break;
     case DB_NEXT:
-	if (cursor->path_len<=0) {
+	if (cursor->path_len<=0)
 	    goto do_db_first;
-	}
-	assert(cursor->path_len>0);
 	r=brtcurs_set_position_next(cursor); if (r!=0) goto died0;
 	r=pma_cget_current(cursor->pmacurs, kbt, vbt); if (r!=0) goto died0;
-	break;
+	if (r == 0) assert_cursor_path(cursor);
+        break;
+    case DB_PREV:
+        if (cursor->path_len<= 0)
+            goto do_db_last;
+        r = brtcurs_set_position_prev(cursor); if (r!=0) goto died0;
+        r = pma_cget_current(cursor->pmacurs, kbt, vbt); if (r!=0) goto died0;
+        if (r == 0) assert_cursor_path(cursor);
+        break;
     default:
 	fprintf(stderr, "%s:%d c_get(...,%d) not ready\n", __FILE__, __LINE__, flags);
 	abort();