brt cursor get both feature

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

newbrt/brt-test.c

@@ -620,10 +620,10 @@ void test_cursor_last_empty(void) {
     //printf("%s:%d %d alloced\n", __FILE__, __LINE__, get_n_items_malloced()); print_malloced_items();
     init_dbt(&vbt);
     //printf("%s:%d %d alloced\n", __FILE__, __LINE__, get_n_items_malloced()); print_malloced_items();
-    r = brt_c_get(cursor, &kbt, &vbt, DB_LAST);
+    r = brt_cursor_get(cursor, &kbt, &vbt, DB_LAST, 0);
     //printf("%s:%d %d alloced\n", __FILE__, __LINE__, get_n_items_malloced()); print_malloced_items();
     assert(r==DB_NOTFOUND);
-    r = brt_c_get(cursor, &kbt, &vbt, DB_FIRST);
+    r = brt_cursor_get(cursor, &kbt, &vbt, DB_FIRST, 0);
     //printf("%s:%d %d alloced\n", __FILE__, __LINE__, get_n_items_malloced()); print_malloced_items();
     assert(r==DB_NOTFOUND);
     r = close_brt(brt);
@@ -656,9 +656,9 @@ void test_cursor_next (void) {
     init_dbt(&vbt);
     //printf("%s:%d %d alloced\n", __FILE__, __LINE__, get_n_items_malloced()); print_malloced_items();
 
-    printf("%s:%d calling brt_c_get(...)\n", __FILE__, __LINE__);
-    r = brt_c_get(cursor, &kbt, &vbt, DB_NEXT);
-    printf("%s:%d called brt_c_get(...)\n", __FILE__, __LINE__);
+    printf("%s:%d calling brt_cursor_get(...)\n", __FILE__, __LINE__);
+    r = brt_cursor_get(cursor, &kbt, &vbt, DB_NEXT, 0);
+    printf("%s:%d called brt_cursor_get(...)\n", __FILE__, __LINE__);
     //printf("%s:%d %d alloced\n", __FILE__, __LINE__, get_n_items_malloced()); print_malloced_items();
     assert(r==0);
     assert(kbt.size==7);
@@ -666,14 +666,14 @@ void test_cursor_next (void) {
     assert(vbt.size==7);
     assert(memcmp(vbt.data, "byenow", 7)==0);
 
-    r = brt_c_get(cursor, &kbt, &vbt, DB_NEXT);
+    r = brt_cursor_get(cursor, &kbt, &vbt, DB_NEXT, 0);
     assert(r==0);
     assert(kbt.size==6);
     assert(memcmp(kbt.data, "hello", 6)==0);
     assert(vbt.size==6);
     assert(memcmp(vbt.data, "there", 6)==0);
 
-    r = brt_c_get(cursor, &kbt, &vbt, DB_NEXT);
+    r = brt_cursor_get(cursor, &kbt, &vbt, DB_NEXT, 0);
     assert(r==DB_NOTFOUND);
 
     r = close_brt(brt);
@@ -752,7 +752,7 @@ static void test_wrongendian_compare (int wrong_p, unsigned int N) {
 
     for (i=0; i<2; i++) {
 	init_dbt(&kbt); init_dbt(&vbt);
-	r = brt_c_get(cursor, &kbt, &vbt, DB_NEXT);
+	r = brt_cursor_get(cursor, &kbt, &vbt, DB_NEXT, 0);
 	assert(r==0);
 	assert(kbt.size==4 && vbt.size==4);
 	printf("%s:%d %02x%02x%02x%02x -> %02x%02x%02x%02x\n", __FILE__, __LINE__,
@@ -787,7 +787,7 @@ static void test_wrongendian_compare (int wrong_p, unsigned int N) {
     for (i=0; i<N; i++) {
 	int this;
 	init_dbt(&kbt); init_dbt(&vbt);
-	r = brt_c_get(cursor, &kbt, &vbt, DB_NEXT);
+	r = brt_cursor_get(cursor, &kbt, &vbt, DB_NEXT, 0);
 	assert(r==0);
 	assert(kbt.size==4 && vbt.size==4);
 	if (0) printf("%s:%d %02x%02x%02x%02x -> %02x%02x%02x%02x\n", __FILE__, __LINE__,
@@ -822,7 +822,7 @@ void assert_cursor_notfound(BRT brt, int position) {
 
     init_dbt(&kbt); kbt.flags = DB_DBT_MALLOC;
     init_dbt(&vbt); vbt.flags = DB_DBT_MALLOC;
-    r = brt_c_get(cursor, &kbt, &vbt, position);
+    r = brt_cursor_get(cursor, &kbt, &vbt, position, 0);
     assert(r == DB_NOTFOUND);
 
     r = brt_cursor_close(cursor);
@@ -841,7 +841,7 @@ void assert_cursor_value(BRT brt, int position, long long value) {
     if (test_cursor_debug) printf("key: ");
     init_dbt(&kbt); kbt.flags = DB_DBT_MALLOC;
     init_dbt(&vbt); vbt.flags = DB_DBT_MALLOC;
-    r = brt_c_get(cursor, &kbt, &vbt, position);
+    r = brt_cursor_get(cursor, &kbt, &vbt, position, 0);
     assert(r == 0);
     if (test_cursor_debug) printf("%s ", (char*)kbt.data);
     assert(vbt.size == sizeof v);
@@ -867,7 +867,7 @@ void assert_cursor_first_last(BRT brt, long long firstv, long long lastv) {
     if (test_cursor_debug) printf("first key: ");
     init_dbt(&kbt); kbt.flags = DB_DBT_MALLOC;
     init_dbt(&vbt); vbt.flags = DB_DBT_MALLOC;
-    r = brt_c_get(cursor, &kbt, &vbt, DB_FIRST);
+    r = brt_cursor_get(cursor, &kbt, &vbt, DB_FIRST, 0);
     assert(r == 0);
     if (test_cursor_debug) printf("%s ", (char*)kbt.data);
     assert(vbt.size == sizeof v);
@@ -880,7 +880,7 @@ void assert_cursor_first_last(BRT brt, long long firstv, long long lastv) {
     if (test_cursor_debug) printf("last key:");
     init_dbt(&kbt); kbt.flags = DB_DBT_MALLOC;
     init_dbt(&vbt); vbt.flags = DB_DBT_MALLOC;
-    r = brt_c_get(cursor, &kbt, &vbt, DB_LAST);
+    r = brt_cursor_get(cursor, &kbt, &vbt, DB_LAST, 0);
     assert(r == 0);
     if (test_cursor_debug)printf("%s ", (char*)kbt.data);
     assert(vbt.size == sizeof v);
@@ -1078,7 +1078,7 @@ void assert_cursor_walk(BRT brt, int n) {
 
         init_dbt(&kbt); kbt.flags = DB_DBT_MALLOC;
         init_dbt(&vbt); vbt.flags = DB_DBT_MALLOC;
-        r = brt_c_get(cursor, &kbt, &vbt, DB_NEXT);
+        r = brt_cursor_get(cursor, &kbt, &vbt, DB_NEXT, 0);
         if (r != 0)
             break;
         if (test_cursor_debug) printf("%s ", (char*)kbt.data);
@@ -1150,7 +1150,7 @@ void assert_cursor_rwalk(BRT brt, int n) {
 
         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);
+        r = brt_cursor_get(cursor, &kbt, &vbt, DB_PREV, 0);
         if (r != 0)
             break;
         if (test_cursor_debug) printf("%s ", (char*)kbt.data);
@@ -1224,7 +1224,7 @@ void assert_cursor_walk_inorder(BRT brt, int n) {
 
         init_dbt(&kbt); kbt.flags = DB_DBT_MALLOC;
         init_dbt(&vbt); vbt.flags = DB_DBT_MALLOC;
-        r = brt_c_get(cursor, &kbt, &vbt, DB_NEXT);
+        r = brt_cursor_get(cursor, &kbt, &vbt, DB_NEXT, 0);
         if (r != 0)
             break;
         if (test_cursor_debug) printf("%s ", (char*)kbt.data);
@@ -1333,7 +1333,7 @@ void test_brt_cursor_split(int n) {
     for (i=0; i<n/2; i++) {
         init_dbt(&kbt); kbt.flags = DB_DBT_MALLOC;
         init_dbt(&vbt); vbt.flags = DB_DBT_MALLOC;
-        r = brt_c_get(cursor, &kbt, &vbt, DB_NEXT);
+        r = brt_cursor_get(cursor, &kbt, &vbt, DB_NEXT, 0);
         assert(r==0);
         if (test_cursor_debug) printf("%s ", (char*)kbt.data);
         toku_free(kbt.data);
@@ -1356,7 +1356,7 @@ void test_brt_cursor_split(int n) {
     for (;;) {
         init_dbt(&kbt); kbt.flags = DB_DBT_MALLOC;
         init_dbt(&vbt); vbt.flags = DB_DBT_MALLOC;
-        r = brt_c_get(cursor, &kbt, &vbt, DB_NEXT);
+        r = brt_cursor_get(cursor, &kbt, &vbt, DB_NEXT, 0);
         if (r != 0)
             break;
         if (test_cursor_debug) printf("%s ", (char*)kbt.data);
@@ -1431,7 +1431,6 @@ void test_multiple_brt_cursor_walk(int n) {
     const int ncursors = n/cursor_gap;
     BRT_CURSOR cursors[ncursors];
 
- 
     unlink(fname);
 
     int h = log16(n);
@@ -1467,7 +1466,7 @@ void test_multiple_brt_cursor_walk(int n) {
             c = i / cursor_gap;
             init_dbt(&key); key.flags = DB_DBT_MALLOC;
             init_dbt(&val); val.flags = DB_DBT_MALLOC;
-            r = brt_c_get(cursors[c], &key, &val, DB_LAST);
+            r = brt_cursor_get(cursors[c], &key, &val, DB_LAST, 0);
             assert(r == 0);
             toku_free(key.data);
             toku_free(val.data);
@@ -1479,7 +1478,7 @@ void test_multiple_brt_cursor_walk(int n) {
         for (c=0; c<ncursors; c++) {
             init_dbt(&key); key.flags = DB_DBT_MALLOC;
             init_dbt(&val); val.flags = DB_DBT_MALLOC;
-            r = brt_c_get(cursors[c], &key, &val, DB_NEXT);
+            r = brt_cursor_get(cursors[c], &key, &val, DB_NEXT, 0);
             if (r == DB_NOTFOUND) {
                 /* we already consumed 1 previously */
                 assert(i == cursor_gap-1);
@@ -1488,7 +1487,7 @@ void test_multiple_brt_cursor_walk(int n) {
                 int vv;
                 assert(val.size == sizeof vv);
                 memcpy(&vv, val.data, val.size);
-                assert(vv == c*1000 + i + 1);
+                assert(vv == c*cursor_gap + i + 1);
                 toku_free(key.data);
                 toku_free(val.data);
             }
@@ -1549,7 +1548,7 @@ void test_brt_cursor_set(int n, int cursor_op) {
         k = htonl(v);
         fill_dbt(&key, &k, sizeof k);
         init_dbt(&val); val.flags = DB_DBT_MALLOC;
-        r = brt_c_get(cursor, &key, &val, cursor_op);
+        r = brt_cursor_get(cursor, &key, &val, cursor_op, 0);
         assert(r == 0);
         assert(val.size == sizeof vv);
         memcpy(&vv, val.data, val.size);
@@ -1564,7 +1563,7 @@ void test_brt_cursor_set(int n, int cursor_op) {
         k = htonl(i);
         fill_dbt(&key, &k, sizeof k);
         init_dbt(&val); val.flags = DB_DBT_MALLOC;
-        r = brt_c_get(cursor, &key, &val, DB_SET);
+        r = brt_cursor_get(cursor, &key, &val, DB_SET, 0);
         assert(r == DB_NOTFOUND);
     }
 
@@ -1622,7 +1621,7 @@ void test_brt_cursor_set_range(int n) {
         k = htonl(v);
         fill_dbt(&key, &k, sizeof k);
         init_dbt(&val); val.flags = DB_DBT_MALLOC;
-        r = brt_c_get(cursor, &key, &val, DB_SET_RANGE);
+        r = brt_cursor_get(cursor, &key, &val, DB_SET_RANGE, 0);
         if (v > max_key)
             /* there is no smallest key if v > the max key */
             assert(r == DB_NOTFOUND);
@@ -1683,7 +1682,7 @@ void test_brt_cursor_delete(int n) {
     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);
+        error = brt_cursor_get(cursor, &key, &val, DB_NEXT, 0);
         if (error == DB_NOTFOUND)
             break;
         assert(error == 0);
@@ -1707,6 +1706,110 @@ void test_brt_cursor_delete(int n) {
     assert(error == 0);
 }
 
+void test_brt_cursor_get_both(int n) {
+    printf("test_brt_cursor_get_both:%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;
+
+    /* verify get_both on an empty tree fails */
+    k = htonl(n+1);
+    v = n+1;
+    fill_dbt(&key, &k, sizeof k);
+    fill_dbt(&val, &v, sizeof v);
+    error = brt_cursor_get(cursor, &key, &val, DB_GET_BOTH, 0);
+    assert(error == DB_NOTFOUND);
+
+    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);
+    }
+
+    /* verify that keys not in the tree fail */
+    k = htonl(n+1);
+    v = n-1;
+    fill_dbt(&key, &k, sizeof k);
+    fill_dbt(&val, &v, sizeof v);
+    error = brt_cursor_get(cursor, &key, &val, DB_GET_BOTH, 0);
+    assert(error == DB_NOTFOUND);
+
+    /* verify that key match but data mismatch fails */
+    for (i=0; i<n; i++) {
+        k = htonl(i);
+        v = i+1;
+        fill_dbt(&key, &k, sizeof k);
+        fill_dbt(&val, &v, sizeof v);
+        error = brt_cursor_get(cursor, &key, &val, DB_GET_BOTH, 0);
+        assert(error == DB_NOTFOUND);
+    }
+
+    /* verify that key and data matches succeeds */
+    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_cursor_get(cursor, &key, &val, DB_GET_BOTH, 0);
+        assert(error == 0);
+#ifdef DB_CURRENT 
+        init_dbt(&key); key.flags = DB_DBT_MALLOC;
+        init_dbt(&val); val.flags = DB_DBT_MALLOC;
+        error = brt_cursor_get(cursor, &key, &val, DB_CURRENT, 0);
+        assert(error == 0);
+        int vv;
+        assert(val.size == sizeof vv);
+        memcpy(&vv, val.data, val.size);
+        assert(vv == i);
+        toku_free(key.data);
+        toku_free(val.data);
+#endif
+        error = brt_cursor_delete(cursor, 0);
+        assert(error == 0);
+
+        k = htonl(i);
+        v = i;
+        fill_dbt(&key, &k, sizeof k);
+        fill_dbt(&val, &v, sizeof v);
+        error = brt_cursor_get(cursor, &key, &val, DB_GET_BOTH, 0);
+        assert(error == DB_NOTFOUND);
+    }
+
+    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;
 
@@ -1750,6 +1853,7 @@ void test_brt_cursor() {
     test_brt_cursor_delete(1000); memory_check_all_free();
     test_multiple_brt_cursor_walk(10000); memory_check_all_free();
     test_multiple_brt_cursor_walk(100000); memory_check_all_free();
+    test_brt_cursor_get_both(1000); memory_check_all_free();
 }
 
 void test_large_kv(int bsize, int ksize, int vsize) {
@@ -1845,6 +1949,7 @@ void test_brt_delete_present(int n) {
     DBT key, val;
     int k, v;
 
+    /* insert 0 .. n-1 */
     for (i=0; i<n; i++) {
         k = htonl(i); v = i;
         fill_dbt(&key, &k, sizeof k);
@@ -1853,6 +1958,7 @@ void test_brt_delete_present(int n) {
         assert(r == 0);
     }
 
+    /* delete 0 .. n-1 */
     for (i=0; i<n; i++) {
         k = htonl(i);
         fill_dbt(&key, &k, sizeof k);
@@ -1877,7 +1983,7 @@ void test_brt_delete_present(int n) {
 
     init_dbt(&key); key.flags = DB_DBT_MALLOC;
     init_dbt(&val); val.flags = DB_DBT_MALLOC;
-    r = brt_c_get(cursor, &key, &val, DB_FIRST);
+    r = brt_cursor_get(cursor, &key, &val, DB_FIRST, 0);
     assert(r != 0);
 
     r = brt_cursor_close(cursor);
@@ -1905,6 +2011,7 @@ void test_brt_delete_not_present(int n) {
     DBT key, val;
     int k, v;
 
+    /* insert 0 .. n-1 */
     for (i=0; i<n; i++) {
         k = htonl(i); v = i;
         fill_dbt(&key, &k, sizeof k);
@@ -1913,6 +2020,7 @@ void test_brt_delete_not_present(int n) {
         assert(r == 0);
     }
 
+    /* delete 0 .. n-1 */
     for (i=0; i<n; i++) {
         k = htonl(i);
         fill_dbt(&key, &k, sizeof k);
@@ -1920,9 +2028,12 @@ void test_brt_delete_not_present(int n) {
         assert(r == 0);
     }
 
+    /* try to delete key n+1 not in the tree */
     k = htonl(n+1);
     fill_dbt(&key, &k, sizeof k);
     r = brt_delete(t, &key, 0);
+    /* the delete may be buffered or may be executed on a leaf node, so the 
+       return value depends */
     printf("brt_delete k=%d %d\n", k, r);
 
     r = close_brt(t);              assert(r==0);
@@ -1947,6 +2058,7 @@ void test_brt_delete_cursor_first(int n) {
     DBT key, val;
     int k, v;
 
+    /* insert 0 .. n-1 */
     for (i=0; i<n; i++) {
         k = htonl(i); v = i;
         fill_dbt(&key, &k, sizeof k);
@@ -1955,6 +2067,7 @@ void test_brt_delete_cursor_first(int n) {
         assert(r == 0);
     }
 
+    /* delete 0 .. n-2 */
     for (i=0; i<n-1; i++) {
         k = htonl(i);
         fill_dbt(&key, &k, sizeof k);
@@ -1962,7 +2075,7 @@ void test_brt_delete_cursor_first(int n) {
         assert(r == 0);
     }
 
-    /* lookups should all fail */
+    /* lookup of 0 .. n-2 should all fail */
     for (i=0; i<n-1; i++) {
         k = htonl(i);
         fill_dbt(&key, &k, sizeof k);
@@ -1971,7 +2084,7 @@ void test_brt_delete_cursor_first(int n) {
         assert(r == DB_NOTFOUND);
     }
 
-    /* cursor should find the last key */
+    /* cursor should find the last key: n-1 */
     BRT_CURSOR cursor;
 
     r = brt_cursor(t, &cursor);
@@ -1979,7 +2092,7 @@ void test_brt_delete_cursor_first(int n) {
 
     init_dbt(&key); key.flags = DB_DBT_MALLOC;
     init_dbt(&val); val.flags = DB_DBT_MALLOC;
-    r = brt_c_get(cursor, &key, &val, DB_FIRST);
+    r = brt_cursor_get(cursor, &key, &val, DB_FIRST, 0);
     assert(r == 0);
     int vv;
     assert(val.size == sizeof vv);

newbrt/brt.c

@@ -946,7 +946,7 @@ static int brt_nonleaf_put_cmd (BRT t, BRTNODE node, BRT_CMD *cmd,
                     k->app_private, db);
             assert(r == 0);
         } else {
-            r = cachetable_unpin(t->cf, child->thisnodename, 0);
+            r = cachetable_unpin(t->cf, child->thisnodename, 1);
             assert(r == 0);
         }
 
@@ -1314,6 +1314,7 @@ int brt_insert (BRT brt, DBT *key, DBT *val, DB* db) {
 }
 
 int brt_lookup_node (BRT brt, diskoff off, DBT *k, DBT *v, DB *db) {
+    int result;
     void *node_v;
     int r = cachetable_get_and_pin(brt->cf, off, &node_v,
 				   brtnode_flush_callback, brtnode_fetch_callback, (void*)(long)brt->h->nodesize);
@@ -1324,7 +1325,7 @@ int brt_lookup_node (BRT brt, diskoff off, DBT *k, DBT *v, DB *db) {
     int childnum;
 
     if (node->height==0) {
-	int result = pma_lookup(node->u.l.buffer, k, v, db);
+	result = pma_lookup(node->u.l.buffer, k, v, db);
 	//printf("%s:%d looked up something, got answerlen=%d\n", __FILE__, __LINE__, answerlen);
 	r = cachetable_unpin(brt->cf, off, 0);
 	assert(r == 0);
@@ -1337,7 +1338,6 @@ int brt_lookup_node (BRT brt, diskoff off, DBT *k, DBT *v, DB *db) {
 	ITEMLEN hanswerlen;
         int type;
 	if (toku_hash_find (node->u.n.htables[childnum], k->data, k->size, &hanswer, &hanswerlen, &type)==0) {
-	    int result;
 	    if (type == BRT_INSERT) {
                 //printf("Found %d bytes\n", *vallen);
                 ybt_set_value(v, hanswer, hanswerlen, &brt->sval);
@@ -1347,7 +1347,7 @@ int brt_lookup_node (BRT brt, diskoff off, DBT *k, DBT *v, DB *db) {
                 result = DB_NOTFOUND;
             } else {
                 assert(0);
-		result = -1; // Some versions of gcc complain
+                result = -1; // some versions of gcc complain
             }
             r = cachetable_unpin(brt->cf, off, 0);
             assert(r == 0);
@@ -1355,12 +1355,10 @@ int brt_lookup_node (BRT brt, diskoff off, DBT *k, DBT *v, DB *db) {
 	}
     }
     
-    {
-	int result = brt_lookup_node(brt, node->u.n.children[childnum], k, v, db);
+    result = brt_lookup_node(brt, node->u.n.children[childnum], k, v, db);
     r = cachetable_unpin(brt->cf, off, 0);
     assert(r == 0);
     return result;
-    }
 }
 
 
@@ -1777,7 +1775,7 @@ void brt_cursor_leaf_split(BRT_CURSOR cursor, BRT t, BRTNODE oldnode, BRTNODE le
         if (0) printf("brt_cursor_leaf_split %p oldnode %lld newnode %lld\n", cursor, 
                       oldnode->thisnodename, newnode->thisnodename);
 
-        r = cachetable_unpin(t->cf, oldnode->thisnodename, 0);
+        r = cachetable_unpin(t->cf, oldnode->thisnodename, 1);
         assert(r == 0);
         r = cachetable_maybe_get_and_pin(t->cf, newnode->thisnodename, &v);
         assert(r == 0 && v == newnode);
@@ -1857,7 +1855,7 @@ void brt_cursor_nonleaf_split(BRT_CURSOR cursor, BRT t, BRTNODE oldnode, BRTNODE
         if (0) printf("brt_cursor_nonleaf_split %p oldnode %lld newnode %lld\n",
     cursor, oldnode->thisnodename, newnode->thisnodename);
 
-        r = cachetable_unpin(t->cf, oldnode->thisnodename, 0);
+        r = cachetable_unpin(t->cf, oldnode->thisnodename, 1);
         assert(r == 0);
         r = cachetable_maybe_get_and_pin(t->cf, newnode->thisnodename, &v);
         assert(r == 0 && v == newnode); 
@@ -1936,7 +1934,7 @@ int brtcurs_set_position_last (BRT_CURSOR cursor, diskoff off) {
     int r = cachetable_get_and_pin(brt->cf, off, &node_v,
 				   brtnode_flush_callback, brtnode_fetch_callback, (void*)(long)brt->h->nodesize);
     if (r!=0) {
-	if (0) { died0: cachetable_unpin(brt->cf, off, 0); }
+	if (0) { died0: cachetable_unpin(brt->cf, off, 1); }
 	return r;
     }
     BRTNODE node = node_v;
@@ -1997,7 +1995,7 @@ int brtcurs_set_position_first (BRT_CURSOR cursor, diskoff off) {
     int r = cachetable_get_and_pin(brt->cf, off, &node_v,
 				   brtnode_flush_callback, brtnode_fetch_callback, (void*)(long)brt->h->nodesize);
     if (r!=0) {
-	if (0) { died0: cachetable_unpin(brt->cf, off, 0); }
+	if (0) { died0: cachetable_unpin(brt->cf, off, 1); }
 	return r;
     }
     BRTNODE node = node_v;
@@ -2064,7 +2062,7 @@ int brtcurs_set_position_next2(BRT_CURSOR cursor) {
     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);
+    cachetable_unpin(cursor->brt->cf, node->thisnodename, 1);
 
     if (brt_cursor_path_empty(cursor))
         return DB_NOTFOUND;
@@ -2124,7 +2122,7 @@ int brtcurs_set_position_prev2(BRT_CURSOR cursor) {
     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);
+    cachetable_unpin(cursor->brt->cf, node->thisnodename, 1);
 
     if (brt_cursor_path_empty(cursor))
         return DB_NOTFOUND;
@@ -2170,7 +2168,7 @@ int brtcurs_set_position_prev (BRT_CURSOR cursor) {
     return 0;
 }
 
-int brtcurs_set_key(BRT_CURSOR cursor, diskoff off, DBT *key, DB *db) {
+int brtcurs_set_key(BRT_CURSOR cursor, diskoff off, DBT *key, DBT *val, int flag, DB *db) {
     BRT brt = cursor->brt;
     void *node_v;
     int r;
@@ -2200,7 +2198,7 @@ int brtcurs_set_key(BRT_CURSOR cursor, diskoff off, DBT *key, DB *db) {
             }
             break;
         }
-        r = brtcurs_set_key(cursor, node->u.n.children[childnum], key, db);
+        r = brtcurs_set_key(cursor, node->u.n.children[childnum], key, val, flag, db);
         if (r != 0)
             brt_node_remove_cursor(node, childnum, cursor);
     } else {
@@ -2208,7 +2206,14 @@ int brtcurs_set_key(BRT_CURSOR cursor, diskoff off, DBT *key, DB *db) {
         cursor->path[cursor->path_len-1] = node;
         r = pma_cursor(node->u.l.buffer, &cursor->pmacurs);
         if (r == 0) {
+            if (flag == DB_SET)
                 r = pma_cursor_set_key(cursor->pmacurs, key, db);
+            else if (flag == DB_GET_BOTH)
+                r = pma_cursor_set_both(cursor->pmacurs, key, val, db);
+            else {
+                assert(0);
+                r = DB_NOTFOUND;
+            }
             if (r != 0) {
                 int rr = pma_cursor_free(&cursor->pmacurs);
                 assert(rr == 0);
@@ -2218,7 +2223,7 @@ int brtcurs_set_key(BRT_CURSOR cursor, diskoff off, DBT *key, DB *db) {
 
     if (r != 0) {
         cursor->path_len -= 1;
-        cachetable_unpin(brt->cf, off, 0);
+        cachetable_unpin(brt->cf, off, 1);
     }
     return r;
 }
@@ -2280,7 +2285,7 @@ int brtcurs_set_range(BRT_CURSOR cursor, diskoff off, DBT *key, DB *db) {
 
     if (r != 0) {
         cursor->path_len -= 1;
-        cachetable_unpin(brt->cf, off, 0);
+        cachetable_unpin(brt->cf, off, 1);
     }
     return r;
 }
@@ -2292,7 +2297,7 @@ static int unpin_cursor (BRT_CURSOR cursor) {
     for (i=0; i<cursor->path_len; i++) {
         BRTNODE node = cursor->path[i];
         brt_node_remove_cursor(node, cursor->pathcnum[i], cursor);
-	int r2 = cachetable_unpin(brt->cf, cursor->path[i]->thisnodename, 0);
+	int r2 = cachetable_unpin(brt->cf, cursor->path[i]->thisnodename, 1);
 	if (r==0) r=r2;
     }
     if (cursor->pmacurs) {
@@ -2321,7 +2326,7 @@ static void assert_cursor_path(BRT_CURSOR cursor) {
     assert(node->height == 0);
 }
 
-int brt_c_get (BRT_CURSOR cursor, DBT *kbt, DBT *vbt, int flags) {
+ int brt_cursor_get (BRT_CURSOR cursor, DBT *kbt, DBT *vbt, int flags, DB *db) {
     int do_rmw=0;
     int r;
     CACHEKEY *rootp;
@@ -2344,7 +2349,7 @@ 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_last(cursor, *rootp); if (r!=0) goto died0;
-        r=pma_cget_current(cursor->pmacurs, kbt, vbt);
+        r=pma_cursor_get_current(cursor->pmacurs, kbt, vbt);
 	if (r == 0) assert_cursor_path(cursor);
         break;
     case DB_FIRST:
@@ -2352,40 +2357,45 @@ 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;
-        r=pma_cget_current(cursor->pmacurs, kbt, vbt);
+        r=pma_cursor_get_current(cursor->pmacurs, kbt, vbt);
 	if (r == 0) assert_cursor_path(cursor);
         break;
     case DB_NEXT:
 	if (cursor->path_len<=0)
 	    goto do_db_first;
 	r=brtcurs_set_position_next(cursor); if (r!=0) goto died0;
-	r=pma_cget_current(cursor->pmacurs, kbt, vbt); if (r!=0) goto died0;
+	r=pma_cursor_get_current(cursor->pmacurs, kbt, vbt); if (r!=0) goto died0;
 	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;
+        r = pma_cursor_get_current(cursor->pmacurs, kbt, vbt); if (r!=0) goto died0;
         if (r == 0) assert_cursor_path(cursor);
         break;
     case DB_SET:
         r = unpin_cursor(cursor);
         assert(r == 0);
-        r = brtcurs_set_key(cursor, *rootp, kbt, 0);
+        r = brtcurs_set_key(cursor, *rootp, kbt, vbt, DB_SET, db);
         if (r != 0) goto died0;
-        r = pma_cget_current(cursor->pmacurs, kbt, vbt);
+        r = pma_cursor_get_current(cursor->pmacurs, kbt, vbt);
+        if (r != 0) goto died0;
+        break;
+    case DB_GET_BOTH:
+        r = unpin_cursor(cursor);
+        assert(r == 0);
+        r = brtcurs_set_key(cursor, *rootp, kbt, vbt, DB_GET_BOTH, db);
         if (r != 0) goto died0;
         break;
     case DB_SET_RANGE:
         r = unpin_cursor(cursor);
         assert(r == 0);
-        r = brtcurs_set_range(cursor, *rootp, kbt, 0);
+        r = brtcurs_set_range(cursor, *rootp, kbt, db);
         if (r != 0) goto died0;
-        r = pma_cget_current(cursor->pmacurs, kbt, vbt);
+        r = pma_cursor_get_current(cursor->pmacurs, kbt, vbt);
         if (r != 0) goto died0;
         break;
-        break;
     default:
 	fprintf(stderr, "%s:%d c_get(...,%d) not ready\n", __FILE__, __LINE__, flags);
 	abort();
@@ -2395,6 +2405,7 @@ int brt_c_get (BRT_CURSOR cursor, DBT *kbt, DBT *vbt, int flags) {
     return 0;
 }
 
+/* delete the key and value under the cursor */
 int brt_cursor_delete(BRT_CURSOR cursor, int flags __attribute__((__unused__))) {
     int r;
 

newbrt/brt.h

@@ -30,7 +30,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_get (BRT_CURSOR cursor, DBT *kbt, DBT *vbt, int brtc_flags, DB *db);
 int brt_cursor_delete(BRT_CURSOR cursor, int flags);
 int brt_cursor_close (BRT_CURSOR curs);
 

newbrt/pma-test.c

@@ -487,24 +487,24 @@ void test_pma_cursor_3 (void) {
     r=pma_cursor(pma, &c); assert(r==0); assert(c!=0);
 
     r=pma_cursor_set_position_first(c); assert(r==0);
-    r=pma_cget_current(c, &key, &val); assert(r==0);
+    r=pma_cursor_get_current(c, &key, &val); assert(r==0);
     assert(key.size=3); assert(memcmp(key.data,"aa",3)==0);
     assert(val.size=2); assert(memcmp(val.data,"a",2)==0);
 
     r=pma_cursor_set_position_next(c); assert(r==0);
-    r=pma_cget_current(c, &key, &val); assert(r==0);
+    r=pma_cursor_get_current(c, &key, &val); assert(r==0);
     assert(key.size=2); assert(memcmp(key.data,"m",2)==0);
     assert(val.size=3); assert(memcmp(val.data,"mm",3)==0);
     
     r=pma_cursor_set_position_next(c); assert(r==0);
-    r=pma_cget_current(c, &key, &val); assert(r==0);
+    r=pma_cursor_get_current(c, &key, &val); assert(r==0);
     assert(key.size=2); assert(memcmp(key.data,"x",2)==0);
     assert(val.size=3); assert(memcmp(val.data,"xx",3)==0);
     
     r=pma_cursor_set_position_next(c); assert(r==DB_NOTFOUND);
 
     /* After an error, the cursor should still point at the same thing. */
-    r=pma_cget_current(c, &key, &val); assert(r==0);
+    r=pma_cursor_get_current(c, &key, &val); assert(r==0);
     assert(key.size=2); assert(memcmp(key.data,"x",2)==0);
     assert(val.size=3); assert(memcmp(val.data,"xx",3)==0);
 
@@ -525,7 +525,7 @@ void assert_cursor_val(PMA_CURSOR cursor, int v) {
 
     init_dbt(&key); key.flags = DB_DBT_MALLOC;
     init_dbt(&val); val.flags = DB_DBT_MALLOC;
-    error = pma_cget_current(cursor, &key, &val);
+    error = pma_cursor_get_current(cursor, &key, &val);
     assert(error == 0);
     assert( v == *(int *)val.data);
     toku_free(key.data);
@@ -637,7 +637,7 @@ void test_pma_cursor_delete(int n) {
     DBT cursorkey, cursorval;
     init_dbt(&cursorkey); cursorkey.flags = DB_DBT_MALLOC;
     init_dbt(&cursorval); cursorval.flags = DB_DBT_MALLOC;
-    error = pma_cget_current(cursor, &cursorkey, &cursorval);
+    error = pma_cursor_get_current(cursor, &cursorkey, &cursorval);
     assert(error != 0);
 
     error = pma_cursor_set_position_first(cursor);
@@ -646,7 +646,7 @@ void test_pma_cursor_delete(int n) {
     int kk;
     init_dbt(&cursorkey); cursorkey.flags = DB_DBT_MALLOC;
     init_dbt(&cursorval); cursorval.flags = DB_DBT_MALLOC;
-    error = pma_cget_current(cursor, &cursorkey, &cursorval);
+    error = pma_cursor_get_current(cursor, &cursorkey, &cursorval);
     assert(error == 0);
     assert(cursorkey.size == sizeof kk);
     kk = 0;
@@ -663,7 +663,7 @@ void test_pma_cursor_delete(int n) {
     /* cursor get should fail */
     init_dbt(&cursorkey); cursorkey.flags = DB_DBT_MALLOC;
     init_dbt(&cursorval); cursorval.flags = DB_DBT_MALLOC;
-    error = pma_cget_current(cursor, &cursorkey, &cursorval);
+    error = pma_cursor_get_current(cursor, &cursorkey, &cursorval);
     assert(error != 0);
 
     error = pma_cursor_set_position_next(cursor);
@@ -673,7 +673,7 @@ void test_pma_cursor_delete(int n) {
         assert(error == 0);
         init_dbt(&cursorkey); cursorkey.flags = DB_DBT_MALLOC;
         init_dbt(&cursorval); cursorval.flags = DB_DBT_MALLOC;
-        error = pma_cget_current(cursor, &cursorkey, &cursorval);
+        error = pma_cursor_get_current(cursor, &cursorkey, &cursorval);
         assert(error == 0);
         assert(cursorkey.size == sizeof kk);
         kk = 1;
@@ -739,7 +739,7 @@ void test_pma_compare_fun (int wrong_endian_p) {
 	} else {
 	    r=pma_cursor_set_position_next(c); assert(r==0);
 	}
-	r=pma_cget_current(c, &key, &val); assert(r==0);
+	r=pma_cursor_get_current(c, &key, &val); assert(r==0);
 	//printf("Got %s, expect %s\n", (char*)key.data, expected_keys[i]);
 	assert(key.size=3); assert(memcmp(key.data,expected_keys[i],3)==0);
 	assert(val.size=4); assert(memcmp(val.data,expected_keys[i],2)==0);
@@ -868,7 +868,7 @@ void print_cursor(const char *str, PMA_CURSOR cursor) {
     printf("cursor %s: ", str);
     init_dbt(&key); key.flags = DB_DBT_MALLOC;
     init_dbt(&val); val.flags = DB_DBT_MALLOC;
-    error = pma_cget_current(cursor, &key, &val);
+    error = pma_cursor_get_current(cursor, &key, &val);
     assert(error == 0);
     printf("%s ", (char*)key.data);
     toku_free(key.data);
@@ -884,7 +884,7 @@ void walk_cursor(const char *str, PMA_CURSOR cursor) {
     for (;;) {
         init_dbt(&key); key.flags = DB_DBT_MALLOC;
         init_dbt(&val); val.flags = DB_DBT_MALLOC;
-        error = pma_cget_current(cursor, &key, &val);
+        error = pma_cursor_get_current(cursor, &key, &val);
         assert(error == 0);
         printf("%s ", (char*)key.data);
         toku_free(key.data);
@@ -905,7 +905,7 @@ void walk_cursor_reverse(const char *str, PMA_CURSOR cursor) {
     for (;;) {
         init_dbt(&key); key.flags = DB_DBT_MALLOC;
         init_dbt(&val); val.flags = DB_DBT_MALLOC;
-        error = pma_cget_current(cursor, &key, &val);
+        error = pma_cursor_get_current(cursor, &key, &val);
         assert(error == 0);
         printf("%s ", (char*)key.data);
         toku_free(key.data);
@@ -1243,7 +1243,7 @@ void assert_cursor_equal(PMA_CURSOR pmacursor, int v) {
     init_dbt(&key); key.flags = DB_DBT_MALLOC;
     init_dbt(&val); val.flags = DB_DBT_MALLOC;
     int r;
-    r = pma_cget_current(pmacursor, &key, &val);
+    r = pma_cursor_get_current(pmacursor, &key, &val);
     assert(r == 0);
     if (0) printf("key %s\n", (char*) key.data);
     int thev;
@@ -1259,7 +1259,7 @@ void assert_cursor_nokey(PMA_CURSOR pmacursor) {
     init_dbt(&key); key.flags = DB_DBT_MALLOC;
     init_dbt(&val); val.flags = DB_DBT_MALLOC;
     int r;
-    r = pma_cget_current(pmacursor, &key, &val);
+    r = pma_cursor_get_current(pmacursor, &key, &val);
     assert(r != 0);
 }
 
@@ -1597,7 +1597,7 @@ void test_pma_cursor_set_key() {
         if (i % 10 == 0) {
             assert(error == 0);
             init_dbt(&val); val.flags = DB_DBT_MALLOC;
-            error = pma_cget_current(cursor, &key, &val);
+            error = pma_cursor_get_current(cursor, &key, &val);
             assert(error == 0);
             int vv;
             assert(val.size == sizeof vv);
@@ -1655,7 +1655,7 @@ void test_pma_cursor_set_range() {
             assert(i > largest_key);
         } else {
             init_dbt(&val); val.flags = DB_DBT_MALLOC;
-            error = pma_cget_current(cursor, &key, &val);
+            error = pma_cursor_get_current(cursor, &key, &val);
             assert(error == 0);
             int vv;
             assert(val.size == sizeof vv);
@@ -1675,6 +1675,152 @@ void test_pma_cursor_set_range() {
     assert(error == 0);
 }
 
+void test_pma_cursor_delete_under() {
+    printf("test_pma_cursor_delete_under\n");
+
+    int error;
+    PMA pma;
+
+    error = pma_create(&pma, default_compare_fun);
+    assert(error == 0);
+
+    PMA_CURSOR cursor;
+    error = pma_cursor(pma, &cursor);
+    assert(error == 0);
+
+    int kvsize;
+
+    /* delete under an uninitialized cursor should fail */
+    error = pma_cursor_delete_under(cursor, &kvsize);
+    assert(error == DB_NOTFOUND);
+
+    const int n = 1000;
+
+    DBT key, val;
+    int k, v;
+
+    int i;
+
+    /* insert 0 .. 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 = pma_insert(pma, &key, &val, 0);
+        assert(error == 0);
+    }
+
+    for (i=0;;i++) {
+        error = pma_cursor_set_position_next(cursor);
+        if (error != 0) {
+            assert(error == DB_NOTFOUND);
+            break;
+        } 
+        init_dbt(&key); key.flags = DB_DBT_MALLOC;
+        init_dbt(&val); val.flags = DB_DBT_MALLOC;
+        error = pma_cursor_get_current(cursor, &key, &val);
+        assert(error == 0);
+        int vv;
+        assert(val.size == sizeof vv);
+        memcpy(&vv, val.data, val.size);
+        assert(vv == i);
+        toku_free(key.data);
+        toku_free(val.data);
+
+        /* delete under should succeed */
+        error = pma_cursor_delete_under(cursor, &kvsize);
+        assert(error == 0);
+
+        /* 2nd delete under should fail */
+        error = pma_cursor_delete_under(cursor, &kvsize);
+        assert(error == DB_NOTFOUND);
+    }
+    assert(i == n);
+
+    error = pma_cursor_free(&cursor);
+    assert(error == 0);
+    assert(pma_n_entries(pma) == 0);
+
+    error = pma_free(&pma);
+    assert(error == 0);
+}
+
+void test_pma_cursor_set_both() {
+    printf("test_pma_cursor_set_both\n");
+
+    int error;
+    PMA pma;
+
+    error = pma_create(&pma, default_compare_fun);
+    assert(error == 0);
+
+    PMA_CURSOR cursor;
+    error = pma_cursor(pma, &cursor);
+    assert(error == 0);
+
+    const int n = 1000;
+
+    DBT key, val;
+    int k, v;
+
+    int i;
+
+    /* insert 0->0, 1->1, .. n-1->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 = pma_insert(pma, &key, &val, 0);
+        assert(error == 0);
+    }
+
+    /* verify key not in pma fails */
+    k = n+1; v = 0;
+    fill_dbt(&key, &k, sizeof k); 
+    fill_dbt(&val, &v, sizeof v);
+    error = pma_cursor_set_both(cursor, &key, &val, 0);
+    assert(error == DB_NOTFOUND);
+    
+    /* key match, data mismatch should fail */
+    for (i=0; i<n; i++) {
+        k = htonl(i);
+        v = i+1;
+        fill_dbt(&key, &k, sizeof k); 
+        fill_dbt(&val, &v, sizeof v);
+        error = pma_cursor_set_both(cursor, &key, &val, 0);
+        assert(error == DB_NOTFOUND);
+    }
+
+    /* key match, data match should succeed */
+    for (i=0; i<n; i++) {
+        k = htonl(i);
+        v = i;
+        fill_dbt(&key, &k, sizeof k); 
+        fill_dbt(&val, &v, sizeof v);
+        error = pma_cursor_set_both(cursor, &key, &val, 0);
+        assert(error == 0);
+
+        init_dbt(&key); key.flags = DB_DBT_MALLOC;
+        init_dbt(&val); val.flags = DB_DBT_MALLOC;
+        error = pma_cursor_get_current(cursor, &key, &val);
+        assert(error == 0);
+        int vv;
+        assert(val.size == sizeof vv);
+        memcpy(&vv, val.data, val.size);
+        assert(vv == i);
+        toku_free(key.data);
+        toku_free(val.data);
+    }
+
+    error = pma_cursor_free(&cursor);
+    assert(error == 0);
+
+    error = pma_free(&pma);
+    assert(error == 0);
+}
+
 void pma_tests (void) {
     memory_check=1;
     test_keycompare();            memory_check_all_free();
@@ -1697,6 +1843,8 @@ void pma_tests (void) {
     test_pma_already_there();     memory_check_all_free();
     test_pma_cursor_set_key();    memory_check_all_free();
     test_pma_cursor_set_range();  memory_check_all_free();    
+    test_pma_cursor_delete_under();  memory_check_all_free();    
+    test_pma_cursor_set_both();   memory_check_all_free();
 }
 
 int main (int argc __attribute__((__unused__)), char *argv[] __attribute__((__unused__))) {

newbrt/pma.c

@@ -442,7 +442,7 @@ int pma_cursor_set_position_next (PMA_CURSOR c) {
     return DB_NOTFOUND;
 }
 
-int pma_cget_current (PMA_CURSOR c, DBT *key, DBT *val) {
+int pma_cursor_get_current(PMA_CURSOR c, DBT *key, DBT *val) {
     if (c->position == -1)
         return DB_NOTFOUND;
     PMA pma = c->pma;
@@ -458,16 +458,37 @@ int pma_cursor_set_key(PMA_CURSOR c, DBT *key, DB *db) {
     PMA pma = c->pma;
     int here = pmainternal_find(pma, key, db);
     assert(0<=here ); assert(here<=pma_index_limit(pma));
-    if (here==pma_index_limit(pma)) 
-        return DB_NOTFOUND;
+    int r = DB_NOTFOUND;
+    if (here < pma->N) {
         DBT k2;
         struct kv_pair *pair = pma->pairs[here];
-    if (kv_pair_valid(pair) && pma->compare_fun(db, key, fill_dbt(&k2, kv_pair_key(pair), kv_pair_keylen(pair)))==0) {
+        if (kv_pair_valid(pair) && 
+            pma->compare_fun(db, key, fill_dbt(&k2, kv_pair_key(pair), kv_pair_keylen(pair)))==0) {
             __pma_delete_resume(c->pma, c->position);
             c->position = here;
-        return 0;
-    } else
-        return DB_NOTFOUND;
+            r = 0;
+        }
+    } 
+    return r;
+}
+
+int pma_cursor_set_both(PMA_CURSOR c, DBT *key, DBT *val, DB *db) {
+    PMA pma = c->pma;
+    int here = pmainternal_find(pma, key, db);
+    assert(0<=here ); assert(here<=pma_index_limit(pma));
+    int r = DB_NOTFOUND;
+    if (here < pma->N) {
+        DBT k2, v2;
+        struct kv_pair *pair = pma->pairs[here];
+        if (kv_pair_valid(pair) && 
+            pma->compare_fun(db, key, fill_dbt(&k2, kv_pair_key(pair), kv_pair_keylen(pair))) == 0 &&
+            pma->compare_fun(db, val, fill_dbt(&v2, kv_pair_val(pair), kv_pair_vallen(pair))) == 0) {
+            __pma_delete_resume(c->pma, c->position);
+            c->position = here;
+            r = 0;
+        }
+    } 
+    return r;
 }
 
 int pma_cursor_set_range(PMA_CURSOR c, DBT *key, DB *db) {
@@ -476,51 +497,36 @@ int pma_cursor_set_range(PMA_CURSOR c, DBT *key, DB *db) {
     assert(0<=here ); assert(here<=pma_index_limit(pma));
 
     /* find the first valid pair where key[here] >= key */
+    int r = DB_NOTFOUND;
     while (here < pma->N) {
         struct kv_pair *pair = pma->pairs[here];
         if (kv_pair_valid(pair)) {
             __pma_delete_resume(c->pma, c->position);
             c->position = here;
-            return 0;
+            r = 0;
+            break;
         }
         here += 1;
     }
-
-    return DB_NOTFOUND;
+    return r;
 }
 
 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);
+            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;
-    c->position=0;
-    if (pma->n_pairs_present==0) return DB_NOTFOUND;
-    while (pma->pairs[c->position].key==0 && c->position<pma->N) {
-	c->position++;
-    }
-    assert(c->position<pma->N && pma->pairs[c->position].key!=0);
-    ybt_set_value(key, pma->pairs[c->position].key, pma->pairs[c->position].keylen, &c->skey);
-    ybt_set_value(val, pma->pairs[c->position].val, pma->pairs[c->position].vallen, &c->sval);
-    return 0;
-}
-#endif
-
 int pma_cursor_free (PMA_CURSOR *cursp) {
     PMA_CURSOR curs=*cursp;
     PMA pma = curs->pma;

newbrt/pma.h

@@ -71,26 +71,26 @@ int pma_bulk_insert(PMA pma, DBT *keys, DBT *vals, int n_newpairs);
 int pma_cursor (PMA, PMA_CURSOR *);
 int pma_cursor_free (PMA_CURSOR*);
 
-/*
- * get the pma that a pma cursor is bound to
- *
- * c - the pma cursor
- * pma - the pma that the cursor is bound to
- */
+/* get the pma that a pma cursor is bound to */
 int pma_cursor_get_pma(PMA_CURSOR c, PMA *pma);
 int pma_cursor_set_position_last (PMA_CURSOR c);
 int pma_cursor_set_position_first (PMA_CURSOR c);
 int pma_cursor_set_position_next (PMA_CURSOR c); /* Requires the cursor is init'd.  Returns DB_NOTFOUND if we fall off the end. */
 int pma_cursor_set_position_prev (PMA_CURSOR c);
-int pma_cget_current (PMA_CURSOR c, DBT *key, DBT *val);
 
-/* set the cursor by key */
+/* get the key and data under the cursor */
+int pma_cursor_get_current(PMA_CURSOR c, DBT *key, DBT *val);
+
+/* set the cursor to the matching key and value pair */
+int pma_cursor_set_both(PMA_CURSOR c, DBT *key, DBT *val, DB *db);
+
+/* move the cursor to the kv pair matching the key */
 int pma_cursor_set_key(PMA_CURSOR c, DBT *key, DB *db);
 
-/* set the cursor to the smallest key >= requested key */
+/* set the cursor to the smallest key in the pma >= key */
 int pma_cursor_set_range(PMA_CURSOR c, DBT *key, DB *db);
 
-/* delete the key under the cursor */
+/* delete the key value pair under the cursor, return the size of the pair */
 int pma_cursor_delete_under(PMA_CURSOR c, int *kvsize);
 
 /* get the last key and value in the pma */
@@ -117,7 +117,7 @@ void pma_iterate (PMA, void(*)(bytevec,ITEMLEN,bytevec,ITEMLEN, void*), void*);
       ITEMLEN datalenvar = pmanode_vallen(table, __i);                 \
       body;                                                            \
 } } })
-#endif
 
 int keycompare (bytevec key1, ITEMLEN key1len, bytevec key2, ITEMLEN key2len);
 
+#endif

src/Makefile

 CFLAGS = -W -Wall -Wno-unused -g -fPIC
 CPPFLAGS = -I../include -I../newbrt
+
 install: libdb.so
 	cp libdb.so ../lib/
+
+clean:
+	rm -rf *.so *.o
+
 ydb.o: ../include/db.h ../newbrt/cachetable.h
 DBBINS = ydb.o ../newbrt/brt.o ../newbrt/brt-serialize.o ../newbrt/cachetable.o ../newbrt/hashtable.o ../newbrt/header-io.o ../newbrt/key.o ../newbrt/memory.o ../newbrt/pma.o ../newbrt/ybt.o
 libdb.so: $(DBBINS)

src/ydb.c

@@ -260,7 +260,7 @@ struct __toku_dbc_internal {
 };
 			  
 int __toku_c_get (DBC *c, DBT *key, DBT *data, u_int32_t flag) {
-    int r = brt_c_get(c->i->c, key, data, flag);
+    int r = brt_cursor_get(c->i->c, key, data, flag, c->i->db);
     return r;
 }