Integrate the omtcursors into the BRT. So far only DB_NEXT is implemented. Addresses #855, #856.

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

newbrt/brt-internal.h

@@ -13,6 +13,7 @@
 #include "kv-pair.h"
 #include "leafentry.h"
 
+typedef void *OMTVALUE;
 #include "omt.h"
 
 #ifndef BRT_FANOUT
@@ -111,6 +112,8 @@ struct brt_header {
     unsigned int *flags_array; // an array of flags.  Element 0 holds the element if no subdatabases allowed.
     
     FIFO fifo; // all the abort and commit commands.  If the header gets flushed to disk, we write the fifo contents beyond the unused_memory.
+
+    uint64_t root_put_counter;
 };
 
 struct brt {
@@ -190,6 +193,8 @@ struct brt_cursor {
     DBT val;
     int is_temporary_cursor;  // If it is a temporary cursor then use the following skey and sval to return tokudb-managed values in dbts.  Otherwise use the brt's skey and skval.
     void *skey, *sval;
+    OMTCURSOR omtcursor;
+    uint64_t  root_put_counter; // what was the count on the BRT when we validated the cursor?
 };
 
 // logs the memory allocation, but not the creation of the new node

newbrt/log.c

@@ -1039,7 +1039,7 @@ int toku_txn_note_brt (TOKUTXN txn, BRT brt) {
 static int remove_brt (OMTVALUE txnv, u_int32_t UU(idx), void *brtv) {
     TOKUTXN txn = txnv;
     BRT     brt = brtv;
-    OMTVALUE brtv_again=0;  // TODO BBB This is not entirely safe. Verify initialization needed.
+    OMTVALUE brtv_again=NULL;
     u_int32_t index;
     int r = toku_omt_find_zero(txn->open_brts, find_filenum, brt, &brtv_again, &index, NULL);
     assert(r==0);
@@ -1058,7 +1058,7 @@ int toku_txn_note_close_brt (BRT brt) {
 static int remove_txn (OMTVALUE brtv, u_int32_t UU(idx), void *txnv) {
     BRT brt     = brtv;
     TOKUTXN txn = txnv;
-    OMTVALUE txnv_again=0; // TODO BBB This is not entirely safe. Verify initialization needed.
+    OMTVALUE txnv_again=NULL;
     u_int32_t index;
     int r = toku_omt_find_zero(brt->txns, find_ptr, txn, &txnv_again, &index, NULL);
     assert(r==0);

newbrt/omt-internal.h

@@ -39,7 +39,7 @@ struct omt {
 };
 
 //Initial max size of root-to-leaf path
-#define TOKU_OMTCURSOR_INITIAL_SIZE 64
+#define TOKU_OMTCURSOR_INITIAL_SIZE 4
 
 // Cursor for order maintenance tree
 struct omtcursor {

newbrt/omt.c

@@ -3,6 +3,7 @@
 #include <errno.h>
 #include <sys/types.h>
 
+typedef void *OMTVALUE;
 #include "omt.h"
 #include "omt-internal.h"
 #include "../newbrt/memory.h"

newbrt/omt.h

@@ -116,7 +116,6 @@
 
 // The programming API:
 
-typedef void *OMTVALUE;
 typedef struct omt *OMT;
 
 typedef struct omtcursor *OMTCURSOR;

newbrt/roll.c

@@ -55,7 +55,7 @@ static int do_insertion (enum brt_cmd_type type, TXNID xid, FILENUM filenum, BYT
 				data
 				? toku_fill_dbt(&data_dbt, data->data, data->len)
 				: toku_init_dbt(&data_dbt) }};
-    OMTVALUE brtv=NULL; // TODO BBB This is not entirely safe. Verify initialization needed.
+    OMTVALUE brtv=NULL;
     r = toku_omt_find_zero(txn->open_brts, find_brt_from_filenum, &filenum, &brtv, NULL, NULL);
 
     if (r==DB_NOTFOUND) {

newbrt/tests/Makefile

@@ -68,7 +68,9 @@ REGRESSION_TESTS = \
 	log-test5 \
 	log-test6 \
 	memtest \
+	omt-cursor-test \
 	omt-test \
+	shortcut \
 	test-assert \
 	test-brt-delete-both \
 	test-brt-overflow \

newbrt/tests/omt-cursor-test.c

+#include <sys/types.h>
+#include <assert.h>
+typedef struct value *OMTVALUE;
+#include "omt.h"
+
+enum { N=10 };
+struct value { int x; } vs[N];
+OMTVALUE ps[N];
+
+static void test (void) {
+    OMT o;
+    OMTCURSOR curs, curs2, curs3;
+    int i, r;
+    OMTVALUE v;
+    for (i=0; i<N; i++) {
+	vs[i].x=i;
+	ps[i]=&vs[i];
+    }
+
+    // destroy the omt first
+    r = toku_omt_create_from_sorted_array(&o, ps, 10);  assert(r==0);
+    r = toku_omt_cursor_create(&curs);                      assert(r==0);
+    r = toku_omt_fetch(o, 5, &v, curs);                     assert(r==0);
+    toku_omt_destroy(&o);
+    toku_omt_cursor_destroy(&curs);
+    
+    // destroy the cursor first
+    r = toku_omt_create_from_sorted_array(&o, ps, 10);  assert(r==0);
+    r = toku_omt_cursor_create(&curs);                      assert(r==0);
+    r = toku_omt_fetch(o, 5, &v, curs);                     assert(r==0);
+    assert(v->x==5);
+    r = toku_omt_cursor_next(curs, &v);
+    assert(r==0 && v->x==6);
+    r = toku_omt_cursor_prev(curs, &v);
+    assert(r==0 && v->x==5);
+    toku_omt_cursor_destroy(&curs);
+    toku_omt_destroy(&o);
+
+    // Create two cursors, destroy omt first
+    r = toku_omt_create_from_sorted_array(&o, ps, 10);  assert(r==0);
+    r = toku_omt_cursor_create(&curs);                      assert(r==0);
+    r = toku_omt_fetch(o, 5, &v, curs);                     assert(r==0);
+    r = toku_omt_cursor_create(&curs2);                      assert(r==0);
+    r = toku_omt_fetch(o, 4, &v, curs2);                     assert(r==0);    
+    r = toku_omt_cursor_next(curs, &v);    assert(r==0 && v->x==6);
+    toku_omt_destroy(&o);
+    toku_omt_cursor_destroy(&curs);
+    toku_omt_cursor_destroy(&curs2);
+
+    // Create two cursors, destroy them first
+    r = toku_omt_create_from_sorted_array(&o, ps, 10);  assert(r==0);
+    r = toku_omt_cursor_create(&curs);                      assert(r==0);
+    r = toku_omt_fetch(o, 5, &v, curs);                     assert(r==0);
+    r = toku_omt_cursor_create(&curs2);                      assert(r==0);
+    r = toku_omt_fetch(o, 4, &v, curs2);                     assert(r==0);    
+    r = toku_omt_cursor_next(curs, &v);    assert(r==0 && v->x==6);
+    r = toku_omt_cursor_prev(curs2, &v);   assert(r==0 && v->x==3);
+    toku_omt_cursor_destroy(&curs);
+    r = toku_omt_cursor_prev(curs2, &v);   assert(r==0 && v->x==2);
+    toku_omt_cursor_destroy(&curs2);
+    toku_omt_destroy(&o);
+    
+    // Create three cursors, destroy them first
+    r = toku_omt_create_from_sorted_array(&o, ps, 10);  assert(r==0);
+    r = toku_omt_cursor_create(&curs);                      assert(r==0);
+    r = toku_omt_fetch(o, 5, &v, curs);                     assert(r==0);
+    r = toku_omt_cursor_create(&curs2);                      assert(r==0);
+    r = toku_omt_fetch(o, 4, &v, curs2);                     assert(r==0);    
+    r = toku_omt_cursor_create(&curs3);                      assert(r==0);
+    r = toku_omt_fetch(o, 9, &v, curs3);                     assert(r==0);    
+    r = toku_omt_cursor_next(curs, &v);    assert(r==0 && v->x==6);
+    r = toku_omt_cursor_prev(curs2, &v);   assert(r==0 && v->x==3);
+    r = toku_omt_cursor_next(curs3, &v);   assert(r!=0 && !toku_omt_cursor_is_valid(curs3));
+    toku_omt_cursor_destroy(&curs);
+    r = toku_omt_cursor_prev(curs2, &v);   assert(r==0 && v->x==2);
+    r = toku_omt_cursor_prev(curs2, &v);   assert(r==0 && v->x==1);
+    r = toku_omt_fetch(o, 1, &v, curs3);                     assert(r==0);
+    r = toku_omt_cursor_prev(curs3, &v);   assert(r==0 && v->x==0);
+    r = toku_omt_cursor_prev(curs3, &v);   assert(r!=0 && !toku_omt_cursor_is_valid(curs3));
+    toku_omt_cursor_destroy(&curs2);
+    toku_omt_destroy(&o);
+    toku_omt_cursor_destroy(&curs3);
+}
+
+int main (int argc __attribute__((__unused__)), char *argv[] __attribute__((__unused__))) {
+    test();
+    return 0;
+}

newbrt/tests/omt-test.c

@@ -3,8 +3,8 @@
 #include <errno.h>
 #include <sys/types.h>
 
-// typedef struct value *TESTVALUE;
-typedef struct value *TESTVALUE;
+typedef struct value *OMTVALUE;
+typedef OMTVALUE TESTVALUE;
 #include "omt.h"
 #include "../newbrt/memory.h"
 #include "../newbrt/toku_assert.h"
@@ -221,7 +221,7 @@ void test_create_from_sorted_array(enum create_type create_choice, enum close_wh
     omt = NULL;
 
     if (create_choice == BATCH_INSERT) {
-        r = toku_omt_create_from_sorted_array(&omt, (OMTVALUE) values, length);
+        r = toku_omt_create_from_sorted_array(&omt, values, length);
         CKERR(r);
     }
     else if (create_choice == INSERT_AT) {
@@ -257,7 +257,7 @@ void test_fetch_verify (OMT omtree, TESTVALUE* val, u_int32_t len ) {
     for (i = 0; i < len; i++) {
         assert(oldv!=val[i]);
         v = NULL;
-        r = toku_omt_fetch(omtree, i, (OMTVALUE) &v, NULL);
+        r = toku_omt_fetch(omtree, i, &v, NULL);
         CKERR(r);
         assert(v != NULL);
         assert(v != oldv);
@@ -265,7 +265,7 @@ void test_fetch_verify (OMT omtree, TESTVALUE* val, u_int32_t len ) {
         assert(v->number == val[i]->number);
 
         v = oldv;
-        r = toku_omt_fetch(omtree, i, (OMTVALUE) &v, c);
+        r = toku_omt_fetch(omtree, i, &v, c);
         CKERR(r);
         assert(v != NULL);
         assert(v != oldv);
@@ -274,7 +274,7 @@ void test_fetch_verify (OMT omtree, TESTVALUE* val, u_int32_t len ) {
         assert(toku_omt_cursor_is_valid(c));
  
         v = oldv;
-        r = toku_omt_cursor_current(c, (OMTVALUE) &v);
+        r = toku_omt_cursor_current(c, &v);
         CKERR(r);
         assert(v != NULL);
         assert(v != oldv);
@@ -284,7 +284,7 @@ void test_fetch_verify (OMT omtree, TESTVALUE* val, u_int32_t len ) {
 
         v = oldv;
         j = i + 1;
-        while ((r = toku_omt_cursor_next(c, (OMTVALUE) &v)) == 0) {
+        while ((r = toku_omt_cursor_next(c, &v)) == 0) {
             assert(toku_omt_cursor_is_valid(c));
             assert(v != NULL);
             assert(v != oldv);
@@ -298,7 +298,7 @@ void test_fetch_verify (OMT omtree, TESTVALUE* val, u_int32_t len ) {
 
         assert(oldv!=val[i]);
         v = NULL;
-        r = toku_omt_fetch(omtree, i, (OMTVALUE) &v, c);
+        r = toku_omt_fetch(omtree, i, &v, c);
         CKERR(r);
         assert(v != NULL);
         assert(v != oldv);
@@ -307,7 +307,7 @@ void test_fetch_verify (OMT omtree, TESTVALUE* val, u_int32_t len ) {
 
         v = oldv;
         j = i - 1;
-        while ((r = toku_omt_cursor_prev(c, (OMTVALUE) &v)) == 0) {
+        while ((r = toku_omt_cursor_prev(c, &v)) == 0) {
             assert(toku_omt_cursor_is_valid(c));
             assert(v != NULL);
             assert(v != oldv);
@@ -323,11 +323,11 @@ void test_fetch_verify (OMT omtree, TESTVALUE* val, u_int32_t len ) {
 
     for (i = len; i < len*2; i++) {
         v = oldv;
-        r = toku_omt_fetch(omtree, i, (OMTVALUE) &v, NULL);
+        r = toku_omt_fetch(omtree, i, &v, NULL);
         CKERR2(r, ERANGE);
         assert(v == oldv);
         v = NULL;
-        r = toku_omt_fetch(omtree, i, (OMTVALUE) &v, c);
+        r = toku_omt_fetch(omtree, i, &v, c);
         CKERR2(r, ERANGE);
         assert(v == NULL);
     }
@@ -355,10 +355,10 @@ int iterate_helper(TESTVALUE v, u_int32_t idx, void* extra) {
 void test_iterate_verify(OMT omtree, TESTVALUE* vals, u_int32_t len) {
     int r;
     iterate_helper_error_return = 0;
-    r = toku_omt_iterate(omtree, (OMTVALUE) iterate_helper, (void*)vals);
+    r = toku_omt_iterate(omtree, iterate_helper, (void*)vals);
     CKERR(r);
     iterate_helper_error_return = 0xFEEDABBA;
-    r = toku_omt_iterate(omtree, (OMTVALUE) iterate_helper, NULL);
+    r = toku_omt_iterate(omtree, iterate_helper, NULL);
     if (!len) {
         CKERR2(r, 0);
     }
@@ -473,7 +473,7 @@ void test_create_insert(enum close_when_done close) {
         u_int32_t idx = UINT32_MAX;
 
         assert(length==toku_omt_size(omt));
-        r = toku_omt_insert(omt, to_insert, (OMTVALUE) insert_helper, to_insert, &idx);
+        r = toku_omt_insert(omt, to_insert, insert_helper, to_insert, &idx);
         CKERR(r);
         assert(idx <= length);
         if (idx > 0) {
@@ -493,7 +493,7 @@ void test_create_insert(enum close_when_done close) {
         test_iterate_verify(omt, values, length);
 
         idx = UINT32_MAX;
-        r = toku_omt_insert(omt, to_insert, (OMTVALUE) insert_helper, to_insert, &idx);
+        r = toku_omt_insert(omt, to_insert, insert_helper, to_insert, &idx);
         CKERR2(r, DB_KEYEXIST);
         assert(idx < length);
         assert(values[idx]->number == to_insert->number);
@@ -678,10 +678,10 @@ void test_find_dir(int dir, void* extra, int (*h)(OMTVALUE, void*),
 
     omt_val = NULL;
     if (dir == 0) {
-        r = toku_omt_find_zero(omt, h, extra,      (OMTVALUE) &omt_val, &idx, c);
+        r = toku_omt_find_zero(omt, h, extra,      &omt_val, &idx, c);
     }
     else {
-        r = toku_omt_find(     omt, h, extra, dir, (OMTVALUE) &omt_val, &idx, c);
+        r = toku_omt_find(     omt, h, extra, dir, &omt_val, &idx, c);
     }
     CKERR2(r, r_expect);
     if (idx_will_change) {
@@ -704,10 +704,10 @@ void test_find_dir(int dir, void* extra, int (*h)(OMTVALUE, void*),
         TESTVALUE tmp;
         assert(idx_will_change);
         omt_val_curs = NULL;
-        r = toku_omt_cursor_current(c, (OMTVALUE) &omt_val_curs);
+        r = toku_omt_cursor_current(c, &omt_val_curs);
         CKERR(r);
         assert(toku_omt_cursor_is_valid(c));
-        r = toku_omt_fetch(omt, idx, (OMTVALUE) &tmp, NULL);
+        r = toku_omt_fetch(omt, idx, &tmp, NULL);
         CKERR(r);
         if (found) assert(tmp==omt_val);
         assert(omt_val_curs != NULL);
@@ -742,10 +742,10 @@ void test_find_dir(int dir, void* extra, int (*h)(OMTVALUE, void*),
     omt_val  = NULL;
     idx      = old_idx;
     if (dir == 0) {
-        r = toku_omt_find_zero(omt, h, extra,      (OMTVALUE) &omt_val, 0, NULL);
+        r = toku_omt_find_zero(omt, h, extra,      &omt_val, 0, NULL);
     }
     else {
-        r = toku_omt_find(     omt, h, extra, dir, (OMTVALUE) &omt_val, 0, NULL);
+        r = toku_omt_find(     omt, h, extra, dir, &omt_val, 0, NULL);
     }
     CKERR2(r, r_expect);
     assert(idx == old_idx);

newbrt/tests/shortcut.c

+#include <assert.h>
+#include <string.h>
+#include <unistd.h>
+#include "brt.h"
+#include "key.h"
+
+static const char fname[]= __FILE__ ".brt";
+
+static TOKUTXN const null_txn = 0;
+CACHETABLE ct;
+BRT brt;
+BRT_CURSOR cursor;
+
+static int test_brt_cursor_keycompare(DB *db __attribute__((unused)), const DBT *a, const DBT *b) {
+    return toku_keycompare(a->data, a->size, b->data, b->size);
+}
+int main (int argc __attribute__((__unused__)), char *argv[]  __attribute__((__unused__))) {
+    int r;
+    DB a_db;
+    DB *db = &a_db;
+    DBT key,val;
+
+    unlink(fname);
+
+    r = toku_brt_create_cachetable(&ct, 0, ZERO_LSN, NULL_LOGGER);                               assert(r==0);
+    r = toku_open_brt(fname, 0, 1, &brt, 1<<12, ct, null_txn, test_brt_cursor_keycompare, db);   assert(r==0);
+    r = toku_brt_cursor(brt, &cursor, 0);                                                        assert(r==0);
+
+    int i;
+    for (i=0; i<1000; i++) {
+	char string[100];
+	snprintf(string, sizeof(string), "%04d", i);
+	r = toku_brt_insert(brt, toku_fill_dbt(&key, string, 5), toku_fill_dbt(&val, string, 5), 0);       assert(r==0);
+    }
+
+    r = toku_brt_cursor_get(cursor, &key, &val, DB_NEXT, null_txn);                              assert(r==0);
+    assert(strcmp(key.data, "0000")==0);
+    assert(strcmp(val.data, "0000")==0);
+
+    r = toku_brt_cursor_get(cursor, &key, &val, DB_NEXT, null_txn);                              assert(r==0);
+    assert(strcmp(key.data, "0001")==0);
+    assert(strcmp(val.data, "0001")==0);
+
+    // This will invalidate due to the root counter bumping, but the OMT itself will still be valid.
+    r = toku_brt_insert(brt, toku_fill_dbt(&key, "d", 2), toku_fill_dbt(&val, "w", 2), 0);       assert(r==0);
+
+    r = toku_brt_cursor_get(cursor, &key, &val, DB_NEXT, null_txn);                              assert(r==0);
+    assert(strcmp(key.data, "0002")==0);
+    assert(strcmp(val.data, "0002")==0);
+
+    r = toku_brt_cursor_close(cursor);                                                           assert(r==0);
+    r = toku_close_brt(brt, 0);                                                                  assert(r==0);
+    r = toku_cachetable_close(&ct);                                                              assert(r==0);
+    return 0;
+}