Closes #855 Merged back into trunk

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

newbrt/brt-internal.h

@@ -13,8 +13,6 @@
 #include "kv-pair.h"
 #include "leafentry.h"
 
-typedef void* OMTVALUE;
-
 #include "omt.h"
 
 #ifndef BRT_FANOUT

newbrt/brt-test-helpers.c

@@ -91,7 +91,7 @@ int toku_testsetup_insert_to_leaf (BRT brt, DISKOFF diskoff, char *key, int keyl
     BRT_CMD_S cmd = {BRT_INSERT, 0, .u.id={toku_fill_dbt(&keydbt, key, keylen),
 					   toku_fill_dbt(&valdbt, val, vallen)}};
     struct cmd_leafval_bessel_extra be = {brt, &cmd, node->flags & TOKU_DB_DUPSORT};
-    r = toku_omt_find_zero(node->u.l.buffer, toku_cmd_leafval_bessel, &be, &storeddatav, &idx);
+    r = toku_omt_find_zero(node->u.l.buffer, toku_cmd_leafval_bessel, &be, &storeddatav, &idx, NULL);
 
 
     if (r==0) {

newbrt/brt.c

@@ -469,7 +469,7 @@ static int brtleaf_split (TOKULOGGER logger, FILENUM filenum, BRT t, BRTNODE nod
     if (splitk) {
 	memset(splitk, 0, sizeof *splitk);
 	OMTVALUE lev;
-	r=toku_omt_fetch(node->u.l.buffer, toku_omt_size(node->u.l.buffer)-1, &lev);
+	r=toku_omt_fetch(node->u.l.buffer, toku_omt_size(node->u.l.buffer)-1, &lev, NULL);
 	assert(r==0); // that fetch should have worked.
 	LEAFENTRY le=lev;
 	if (node->flags&TOKU_DB_DUPSORT) {
@@ -1461,7 +1461,7 @@ static int brt_leaf_put_cmd (BRT t, BRTNODE node, BRT_CMD cmd,
     FILENUM filenum = toku_cachefile_filenum(t->cf);
 
     LEAFENTRY storeddata;
-    OMTVALUE storeddatav;
+    OMTVALUE storeddatav=NULL; // TODO BBB This is not entirely safe. Verify initialization needed.
 
     u_int32_t idx;
     int r;
@@ -1476,7 +1476,7 @@ static int brt_leaf_put_cmd (BRT t, BRTNODE node, BRT_CMD cmd,
     case BRT_INSERT:
 	
 	r = toku_omt_find_zero(node->u.l.buffer, toku_cmd_leafval_bessel, &be,
-			       &storeddatav, &idx);
+			       &storeddatav, &idx, NULL);
 	if (r==DB_NOTFOUND) {
 	    storeddata = 0;
 	} else if (r!=0) {
@@ -1494,7 +1494,7 @@ static int brt_leaf_put_cmd (BRT t, BRTNODE node, BRT_CMD cmd,
 
 	// Delete the one item
 	r = toku_omt_find_zero(node->u.l.buffer, toku_cmd_leafval_bessel,  &be,
-			       &storeddatav, &idx);
+			       &storeddatav, &idx, NULL);
 	if (r == DB_NOTFOUND) break;
 	if (r != 0) return r;
 	storeddata=storeddatav;
@@ -1515,7 +1515,7 @@ static int brt_leaf_put_cmd (BRT t, BRTNODE node, BRT_CMD cmd,
 	// Delete all the matches
 
 	r = toku_omt_find_zero(node->u.l.buffer, toku_cmd_leafval_bessel, &be,
-			       &storeddatav, &idx);
+			       &storeddatav, &idx, NULL);
 	if (r == DB_NOTFOUND) break;
 	if (r != 0) return r;
 	storeddata=storeddatav;
@@ -1532,7 +1532,7 @@ static int brt_leaf_put_cmd (BRT t, BRTNODE node, BRT_CMD cmd,
 	    BRT_CMD_S ncmd = { cmd->type, cmd->xid, .u.id={cmd->u.id.key, toku_fill_dbt(&valdbt, save_val, vallen)}};
 	    struct cmd_leafval_bessel_extra nbe = {t, &ncmd, 1};
 	    r = toku_omt_find(node->u.l.buffer, toku_cmd_leafval_bessel,  &nbe, +1,
-			      &storeddatav, &idx);
+			      &storeddatav, &idx, NULL);
 	    
 	    toku_free(save_val);
 	    if (r!=0) break;
@@ -2708,7 +2708,7 @@ static int brt_search_leaf_node(BRT brt, BRTNODE node, brt_search_t *search, DBT
 			  bessel_from_search_t,
 			  search,
 			  direction,
-			  &datav, &idx);
+			  &datav, &idx, NULL);
     if (r!=0) return r;
 
     LEAFENTRY le = datav;
@@ -2727,7 +2727,7 @@ static int brt_search_leaf_node(BRT brt, BRTNODE node, brt_search_t *search, DBT
 		break;
 	    }
 	    if (idx>=toku_omt_size(node->u.l.buffer)) continue;
-	    r = toku_omt_fetch(node->u.l.buffer, idx, &datav);
+	    r = toku_omt_fetch(node->u.l.buffer, idx, &datav, NULL);
 	    assert(r==0); // we just validated the index
 	    le = datav;
 	    if (!le_is_provdel(le)) goto got_a_good_value;
@@ -3246,12 +3246,14 @@ static void toku_brt_keyrange_internal (BRT brt, CACHEKEY nodename, DBT *key, u_
 	BRT_CMD_S cmd = { BRT_INSERT, 0, .u.id={key,0}};
 	struct cmd_leafval_bessel_extra be = {brt, &cmd, 0};
 	u_int32_t idx;
-	int r = toku_omt_find_zero(node->u.l.buffer, toku_cmd_leafval_bessel, &be, 0, &idx);
+	int r = toku_omt_find_zero(node->u.l.buffer, toku_cmd_leafval_bessel, &be, 0, &idx, NULL);
+        // TODO: Check for r==ENOMEM if the last argument (cursor) is not NULL
+        // (history: we changed find_zero to support cursor, and now find_zero can fail if the cursor cannot grow)
 	*less += idx;
 	if (r==0 && (brt->flags & TOKU_DB_DUP)) {
 	    // There is something, and so we now want to find the rightmost extent.
 	    u_int32_t idx2;
-	    r = toku_omt_find(node->u.l.buffer, toku_cmd_leafval_bessel, &be, +1, 0, &idx2);
+	    r = toku_omt_find(node->u.l.buffer, toku_cmd_leafval_bessel, &be, +1, 0, &idx2, NULL);
 	    if (r==0) {
 		*greater += toku_omt_size(node->u.l.buffer)-idx2;
 		*equal   += idx2-idx;

newbrt/log.c

@@ -1022,7 +1022,7 @@ static int find_filenum (OMTVALUE v, void *brtv) {
 int toku_txn_note_brt (TOKUTXN txn, BRT brt) {
     OMTVALUE txnv;
     u_int32_t index;
-    int r = toku_omt_find_zero(brt->txns, find_ptr, txn, &txnv, &index);
+    int r = toku_omt_find_zero(brt->txns, find_ptr, txn, &txnv, &index, NULL);
     if (r==0) {
 	// It's already there.
 	assert((TOKUTXN)txnv==txn);
@@ -1039,9 +1039,9 @@ 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;
+    OMTVALUE brtv_again=0;  // TODO BBB This is not entirely safe. Verify initialization needed.
     u_int32_t index;
-    int r = toku_omt_find_zero(txn->open_brts, find_filenum, brt, &brtv_again, &index);
+    int r = toku_omt_find_zero(txn->open_brts, find_filenum, brt, &brtv_again, &index, NULL);
     assert(r==0);
     assert((void*)brtv_again==brtv);
     r = toku_omt_delete_at(txn->open_brts, index);
@@ -1058,9 +1058,9 @@ 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;
+    OMTVALUE txnv_again=0; // TODO BBB This is not entirely safe. Verify initialization needed.
     u_int32_t index;
-    int r = toku_omt_find_zero(brt->txns, find_ptr, txn, &txnv_again, &index);
+    int r = toku_omt_find_zero(brt->txns, find_ptr, txn, &txnv_again, &index, NULL);
     assert(r==0);
     assert((void*)txnv_again==txnv);
     r = toku_omt_delete_at(brt->txns, index);

newbrt/omt-internal.h

+/**
+  \brief OMT implementation header
+*/
+
+#if !defined(OMTI_H)
+#define OMTI_H
+
+#ident "Copyright (c) 2008 Tokutek Inc.  All rights reserved."
+
+#include <stdint.h>
+
+/** Type for the node index */
+typedef u_int32_t node_idx;
+
+
+/** Define a NULL index in the node array */
+#define NODE_NULL UINT32_MAX
+
+/** OMT node */
+typedef struct omt_node *OMT_NODE;
+struct omt_node {
+    u_int32_t weight; /* Size of subtree rooted at this node 
+                         (including this one). */
+    node_idx  left;   /* Index of left  subtree. */
+    node_idx  right;  /* Index of right subtree. */
+    OMTVALUE  value;  /* The value stored in the node. */
+};
+
+/** Order Maintenance Tree */
+struct omt {
+    node_idx   root;
+
+    u_int32_t  node_capacity;
+    OMT_NODE   nodes;
+    node_idx   free_idx;
+
+    u_int32_t  tmparray_size;
+    node_idx*  tmparray;
+};
+
+//Initial max size of root-to-leaf path
+#define TOKU_OMTCURSOR_INITIAL_SIZE 64
+
+// Cursor for order maintenance tree
+struct omtcursor {
+    u_int32_t max_pathlen; //Max (root to leaf) path length;
+    u_int32_t pathlen;     //Length of current path
+    node_idx *path;
+    OMT       omt;         //Associated OMT
+};
+
+
+#endif  /* #ifndef OMTI_H */
+

newbrt/omt.c

@@ -2,36 +2,14 @@
 
 #include <errno.h>
 #include <sys/types.h>
-#include <stdint.h>
 
-typedef void *OMTVALUE;
 #include "omt.h"
+#include "omt-internal.h"
 #include "../newbrt/memory.h"
 #include "../newbrt/toku_assert.h"
 #include "../include/db.h"
 #include "../newbrt/brttypes.h"
 
-typedef u_int32_t node_idx;
-static const node_idx NODE_NULL = UINT32_MAX;
-
-typedef struct omt_node *OMT_NODE;
-struct omt_node {
-    u_int32_t weight; /* Size of subtree rooted at this node (including this one). */
-    node_idx  left;   /* Index of left  subtree. */
-    node_idx  right;  /* Index of right subtree. */
-    OMTVALUE  value;  /* The value stored in the node. */
-};
-
-struct omt {
-    node_idx   root;
-
-    u_int32_t  node_capacity;
-    OMT_NODE   nodes;
-    node_idx   free_idx;
-
-    u_int32_t  tmparray_size;
-    node_idx*  tmparray;
-};
 
 static int omt_create_internal(OMT *omtp, u_int32_t num_starting_nodes) {
     if (num_starting_nodes < 2) num_starting_nodes = 2;
@@ -339,21 +317,60 @@ int toku_omt_delete_at(OMT omt, u_int32_t index) {
     return 0;
 }
 
-static inline void fetch_internal(OMT V, node_idx idx, u_int32_t i, OMTVALUE *v) {
+static int omtcursor_stack_push(OMTCURSOR c, node_idx idx) {
+    if (c->max_pathlen-1<=c->pathlen) {
+        //Increase max_pathlen
+        u_int32_t new_max = c->max_pathlen*2;
+        node_idx *tmp_path = toku_realloc(c->path, new_max*sizeof(*c->path));
+        if (tmp_path==NULL) return errno;
+        c->path        = tmp_path;
+        c->max_pathlen = new_max;
+    }
+    c->path[c->pathlen++] = idx;
+    return 0;
+}
+
+static node_idx omtcursor_stack_peek(OMTCURSOR c) {
+    return c->path[c->pathlen-1];
+}
+
+static node_idx omtcursor_stack_pop(OMTCURSOR c) {
+    assert(c->pathlen);
+    node_idx value = omtcursor_stack_peek(c);;
+    c->pathlen--;
+    return value;
+}
+
+static void omtcursor_associate(OMTCURSOR c, OMT omt) {
+    c->omt     = omt;
+    c->pathlen = 0;
+}
+
+static inline int fetch_internal(OMT V, node_idx idx, u_int32_t i, OMTVALUE *v, OMTCURSOR c) {
+    int r;
+    // Add the current index to the cursor path
+    if (c!=NULL && (r=omtcursor_stack_push(c, idx))) return r;
+
+    /* Find the node corresponding to index idx */
     OMT_NODE n = V->nodes+idx;
+
+    /* Visit recursively the appropriate sub-tree */
     if (i < nweight(V, n->left)) {
-        fetch_internal(V, n->left,  i, v);
+        return fetch_internal(V, n->left,  i, v, c);
     } else if (i == nweight(V, n->left)) {
         *v = n->value;
     } else {
-        fetch_internal(V, n->right, i-nweight(V, n->left)-1, v);
+        return fetch_internal(V, n->right, i-nweight(V, n->left)-1, v, c);
     }
+    return 0;
 }
 
-int toku_omt_fetch(OMT V, u_int32_t i, OMTVALUE *v) {
+int toku_omt_fetch(OMT V, u_int32_t i, OMTVALUE *v, OMTCURSOR c) {
     if (i>=nweight(V, V->root)) return ERANGE;
-    fetch_internal(V, V->root, i, v);
-    return 0;
+    if (c!=NULL) omtcursor_associate(c, V);
+    int r = fetch_internal(V, V->root, i, v, c);
+    if (c!=NULL && r!=0) toku_omt_cursor_invalidate(c);
+    return r;
 }
 
 static inline int iterate_internal(OMT omt, u_int32_t left, u_int32_t right,
@@ -381,7 +398,7 @@ int toku_omt_insert(OMT omt, OMTVALUE value, int(*h)(OMTVALUE, void*v), void *v,
     int r;
     u_int32_t idx;
 
-    r = toku_omt_find_zero(omt, h, v, NULL, &idx);
+    r = toku_omt_find_zero(omt, h, v, NULL, &idx, NULL);
     if (r==0) {
         if (index) *index = idx;
         return DB_KEYEXIST;
@@ -394,80 +411,131 @@ int toku_omt_insert(OMT omt, OMTVALUE value, int(*h)(OMTVALUE, void*v), void *v,
     return 0;
 }
 
-static inline int find_internal_zero(OMT omt, node_idx n_idx, int (*h)(OMTVALUE, void*extra), void*extra, OMTVALUE *value, u_int32_t *index) {
+static inline int find_internal_zero(OMT omt, node_idx n_idx, int (*h)(OMTVALUE, void*extra), void*extra, OMTVALUE *value, u_int32_t *index, OMTCURSOR c) {
+    int r;
     if (n_idx==NODE_NULL) {
-	if (index!=NULL) (*index)=0;
+	*index=0;
 	return DB_NOTFOUND;
     }
+    // Add the current index to the cursor path
+    if (c!=NULL && (r=omtcursor_stack_push(c, n_idx))) return r;
     OMT_NODE n = omt->nodes+n_idx;
     int hv = h(n->value, extra);
     if (hv<0) {
-        int r = find_internal_zero(omt, n->right, h, extra, value, index);
-        if (index!=NULL) (*index) += nweight(omt, n->left)+1;
+        r = find_internal_zero(omt, n->right, h, extra, value, index, c);
+        *index += nweight(omt, n->left)+1;
         return r;
     } else if (hv>0) {
-        return find_internal_zero(omt, n->left, h, extra, value, index);
+        r = find_internal_zero(omt, n->left, h, extra, value, index, c);
+        if (c!=NULL && r==DB_NOTFOUND && *index==nweight(omt, n->left)) {
+            //Truncate the saved cursor path at n_idx.
+            while (omtcursor_stack_peek(c)!=n_idx) omtcursor_stack_pop(c);
+        }
+        return r;
     } else {
-        int r = find_internal_zero(omt, n->left, h, extra, value, index);
+        r = find_internal_zero(omt, n->left, h, extra, value, index, c);
         if (r==DB_NOTFOUND) {
-            if (index!=NULL) *index = nweight(omt, n->left);
-            if (value!=NULL) *value = n->value;
+            *index = nweight(omt, n->left);
+            *value = n->value;
+            if (c!=NULL) {
+                //Truncate the saved cursor path at n_idx.
+                while (omtcursor_stack_peek(c)!=n_idx) omtcursor_stack_pop(c);
+            }
             r = 0;
         }
         return r;
     }
 }
 
-int toku_omt_find_zero(OMT V, int (*h)(OMTVALUE, void*extra), void*extra, OMTVALUE *value, u_int32_t *index) {
-    return find_internal_zero(V, V->root, h, extra, value, index);
+int toku_omt_find_zero(OMT V, int (*h)(OMTVALUE, void*extra), void*extra, OMTVALUE *value, u_int32_t *index, OMTCURSOR c) {
+    if (c!=NULL) omtcursor_associate(c, V);
+    u_int32_t idx_tmp;
+    OMTVALUE  val_tmp; 
+    int r = find_internal_zero(V, V->root, h, extra, &val_tmp, &idx_tmp, c);
+    if (c!=NULL && ( (r!=0 && r!=DB_NOTFOUND) ||
+                      idx_tmp==nweight(V, V->root))) {
+        toku_omt_cursor_invalidate(c);
+    }
+    if (c==NULL || r==0 || r==DB_NOTFOUND) {
+        if (index!=NULL)         *index = idx_tmp;
+        if (value!=NULL && r==0) *value = val_tmp;
+    }
+    return r;
 }
 
 //  If direction <0 then find the largest  i such that h(V_i,extra)<0.
-static inline int find_internal_minus(OMT omt, node_idx n_idx, int (*h)(OMTVALUE, void*extra), void*extra, OMTVALUE *value, u_int32_t *index) {
+static inline int find_internal_minus(OMT omt, node_idx n_idx, int (*h)(OMTVALUE, void*extra), void*extra, OMTVALUE *value, u_int32_t *index, OMTCURSOR c) {
+    int r;
     if (n_idx==NODE_NULL) return DB_NOTFOUND;
+    // Add the current index to the cursor path
+    if (c!=NULL && (r=omtcursor_stack_push(c, n_idx))) return r;
     OMT_NODE n = omt->nodes+n_idx;
     int hv = h(n->value, extra);
     if (hv<0) {
-        int r = find_internal_minus(omt, n->right, h, extra, value, index);
-        if (r==0 && index!=NULL) (*index) += nweight(omt, n->left)+1;
+        r = find_internal_minus(omt, n->right, h, extra, value, index, c);
+        if (r==0) (*index) += nweight(omt, n->left)+1;
         else if (r==DB_NOTFOUND) {
-            if (index!=NULL) *index = nweight(omt, n->left);
-            if (value!=NULL) *value = n->value;
+            *index = nweight(omt, n->left);
+            *value = n->value;
+            if (c!=NULL) {
+                //Truncate the saved cursor path at n_idx.
+                while (omtcursor_stack_peek(c)!=n_idx) omtcursor_stack_pop(c);
+            }
             r = 0;
         }
         return r;
     } else {
-        return find_internal_minus(omt, n->left, h, extra, value, index);
+        return find_internal_minus(omt, n->left, h, extra, value, index, c);
     }
 }
 
 //  If direction >0 then find the smallest i such that h(V_i,extra)>0.
-static inline int find_internal_plus(OMT omt, node_idx n_idx, int (*h)(OMTVALUE, void*extra), void*extra, OMTVALUE *value, u_int32_t *index) {
+static inline int find_internal_plus(OMT omt, node_idx n_idx, int (*h)(OMTVALUE, void*extra), void*extra, OMTVALUE *value, u_int32_t *index, OMTCURSOR c) {
+    int r;
     if (n_idx==NODE_NULL) return DB_NOTFOUND;
+    // Add the current index to the cursor path
+    if (c!=NULL && (r=omtcursor_stack_push(c, n_idx))) return r;
     OMT_NODE n = omt->nodes+n_idx;
     int hv = h(n->value, extra);
     if (hv>0) {
-        int r = find_internal_plus(omt, n->left, h, extra, value, index);
+        r = find_internal_plus(omt, n->left, h, extra, value, index, c);
         if (r==DB_NOTFOUND) {
-            if (index!=NULL) *index = nweight(omt, n->left);
-            if (value!=NULL) *value = n->value;
+            *index = nweight(omt, n->left);
+            *value = n->value;
+            if (c!=NULL) {
+                //Truncate the saved cursor path at n_idx.
+                while (omtcursor_stack_peek(c)!=n_idx) omtcursor_stack_pop(c);
+            }
             r = 0;
         }
         return r;
     } else {
-        int r = find_internal_plus(omt, n->right, h, extra, value, index);
-        if (r==0 && index!=NULL) (*index) += nweight(omt, n->left)+1;
+        r = find_internal_plus(omt, n->right, h, extra, value, index, c);
+        if (r==0) (*index) += nweight(omt, n->left)+1;
         return r;
     }
 }
 
-int toku_omt_find(OMT V, int (*h)(OMTVALUE, void*extra), void*extra, int direction, OMTVALUE *value, u_int32_t *index) {
+int toku_omt_find(OMT V, int (*h)(OMTVALUE, void*extra), void*extra, int direction, OMTVALUE *value, u_int32_t *index, OMTCURSOR c) {
     if (direction==0) {
 	abort();
-    } else if (direction<0) {
-        return find_internal_minus(V, V->root, h, extra, value, index);
+    }
+    else {
+        int r;
+        u_int32_t idx_tmp;
+        OMTVALUE  val_tmp;
+        if (c!=NULL) omtcursor_associate(c, V);
+        if (direction<0) {
+            r = find_internal_minus(V, V->root, h, extra, &val_tmp, &idx_tmp, c);
         } else {
-        return find_internal_plus( V, V->root, h, extra, value, index);
+            r = find_internal_plus( V, V->root, h, extra, &val_tmp, &idx_tmp, c);
+        }
+        if (c!=NULL && r!=0) toku_omt_cursor_invalidate(c);
+        if (r==0) {
+            if (index!=NULL) *index = idx_tmp;
+            if (value!=NULL) *value = val_tmp;
+        }
+        return r;
     }
 }
 
@@ -524,6 +592,114 @@ void toku_omt_clear(OMT omt) {
     omt->root     = NODE_NULL;
 }
 
-unsigned long toku_omt_memory_size (OMT omt) {
+int toku_omt_cursor_create(OMTCURSOR *p) {
+    OMTCURSOR MALLOC(result);
+    if (result==NULL) return errno;
+    result->max_pathlen = TOKU_OMTCURSOR_INITIAL_SIZE;
+    result->pathlen     = 0;
+    MALLOC_N(result->max_pathlen, result->path); 
+    if (result->path==NULL) {
+        toku_free(result);
+        return errno;
+    }
+    result->omt = NULL;
+    *p = result;
+    return 0;
+}
+
+void toku_omt_cursor_destroy(OMTCURSOR *p) {
+    OMTCURSOR c=*p;
+    toku_free(c->path);
+    toku_free(c);
+    *p = NULL;
+}
+
+int toku_omt_cursor_is_valid(OMTCURSOR c) {
+    return c->pathlen>0 && c->omt!=NULL;
+}
+
+void toku_omt_cursor_invalidate(OMTCURSOR c) {
+    c->pathlen = 0;
+    c->omt=NULL;
+}
+
+static void omtcursor_current_internal(OMTCURSOR c, OMTVALUE *v) {
+    *v = c->omt->nodes[omtcursor_stack_peek(c)].value;
+}
+
+int toku_omt_cursor_current(OMTCURSOR c, OMTVALUE *v) {
+    if (!toku_omt_cursor_is_valid(c)) return EINVAL;
+    omtcursor_current_internal(c, v);
+    return 0;
+}
+
+static int omtcursor_next_internal(OMTCURSOR c) {
+    if (!toku_omt_cursor_is_valid(c)) return EINVAL;
+    OMT_NODE current = c->omt->nodes+omtcursor_stack_peek(c);
+    if (current->right!=NODE_NULL) {
+        //Enter into subtree
+        if (omtcursor_stack_push(c, current->right)) goto invalidate;
+        current = c->omt->nodes+current->right;
+        while (current->left!=NODE_NULL) {
+            if (omtcursor_stack_push(c, current->left)) goto invalidate;
+            current = c->omt->nodes+current->left;
+        }
+        return 0;
+    }
+    else {
+        //Pop the stack till we remove a left child.
+        while (c->pathlen>=2) {
+            node_idx child_idx  = omtcursor_stack_pop(c);
+            node_idx parent_idx = omtcursor_stack_peek(c);
+            if (c->omt->nodes[parent_idx].left==child_idx) return 0;
+        }
+        goto invalidate;
+    }
+invalidate:
+    toku_omt_cursor_invalidate(c);
+    return EINVAL;
+}
+
+int toku_omt_cursor_next(OMTCURSOR c, OMTVALUE *v) {
+    if (omtcursor_next_internal(c)) return EINVAL;
+    omtcursor_current_internal(c, v);
+    return 0;
+}
+
+static int omtcursor_prev_internal(OMTCURSOR c) {
+    if (!toku_omt_cursor_is_valid(c)) return EINVAL;
+    OMT_NODE current = c->omt->nodes+omtcursor_stack_peek(c);
+    if (current->left!=NODE_NULL) {
+        //Enter into subtree
+        if (omtcursor_stack_push(c, current->left)) goto invalidate;
+        current = c->omt->nodes+current->left;
+        while (current->right!=NODE_NULL) {
+            if (omtcursor_stack_push(c, current->right)) goto invalidate;
+            current = c->omt->nodes+current->right;
+        }
+        return 0;
+    }
+    else {
+        //Pop the stack till we remove a right child.
+        while (c->pathlen>=2) {
+            node_idx child_idx  = omtcursor_stack_pop(c);
+            node_idx parent_idx = omtcursor_stack_peek(c);
+            if (c->omt->nodes[parent_idx].right==child_idx) return 0;
+        }
+        goto invalidate;
+    }
+invalidate:
+    toku_omt_cursor_invalidate(c);
+    return EINVAL;
+}
+
+int toku_omt_cursor_prev(OMTCURSOR c, OMTVALUE *v) {
+    if (omtcursor_prev_internal(c)) return EINVAL;
+    omtcursor_current_internal(c, v);
+    return 0;
+}
+
+size_t toku_omt_memory_size (OMT omt) {
     return sizeof(*omt)+omt->node_capacity*sizeof(omt->nodes[0]) + omt->tmparray_size*sizeof(omt->tmparray[0]);
 }
+

newbrt/omt.h

@@ -49,11 +49,79 @@
 //  Insertion and deletion should run with $O(\log |V|)$ time and $O(\log |V|)$ calls to the Heaviside function.
 //  The memory required is O(|V|).
 //
+
+//**********************************************************************
+//* OMT Cursors
+//**********************************************************************
+
+// OMTs also support cursors.   An OMTCURSOR is a  mutable
+// An OMTCURSOR is a mutable object that, at any moment in time, is
+// either associated with a single OMT or is not associated with any
+// OMT.  Many different OMTCURSORs can be associated with a single OMT.
+
+// We say that an OMTCURSOR is *valid* if it is currently
+// associated with an OMT and has an abstract offset assigned to it.
+// An OMTCURSOR that is not valid is said to be invalid.
+
+// Abstractly, an OMTCURSOR simply contains an integer offset of a
+// particular OMTVALUE.   We call this abstract integer the *offset*.
+// Note, however, that the implementation may use a more
+// complex representation in order to obtain higher performance.
+// (Note: A first implementation might use the integer.)
+
+// Given a valid OMTCURSOR, one
+//  * obtain the OMTVALUE at which the integer points in O(1) time,
+//  * increment or decrement the abstract integer (usually quickly.)
+//    The requirements are that the cursor is initialized to a
+//    randomly chosen valid integer, then the integer can be
+//    incremented in O(1) expected time.
+
+// The OMTCURSOR may become invalidated under several conditions:
+//  * Incrementing or decrementing the abstract integer out of its
+//    valid range invalidates the OMTCURSOR.
+//  * If the OMT is modified, it may invalidate the cursor.
+//  * The user of the OMTCURSOR may explicitly invalidate the cursor.
+//  * The OMT is destroyed (in which case the OMTCURSOR is
+//    invalidated, but not destroyed.)
+
+// Implementation Hint:  One way to implement the OMTCURSOR is with an
+// integer.  The problem is that obtaining the value at which the integer
+// points takes O(\log n) time, which is not fast enough to meet the
+// specification.    However, this implementation is probably much
+// faster than our current implementation because it is O(\log n)
+// integer comparisons instead of O(\log n) key comparisons.  This
+// simple implementation may be the right thing for a first cut.
+//
+// To actually achieve the performance requirements, here's a better
+// implementation:   The OMTCURSOR contains a path from root to leaf.
+// Fetching the current value is O(1) time since the leaf is
+// immediately accessible.   Modifying the path to find the next or
+// previous item has O(1) expected time at a randomly chosen valid
+// point
+//
+// The path can be implemented as an array.  It probably makes sense
+// for the array to by dynamically resized as needed.  Since the
+// array's size is O(log n), it is not necessary to ever shrink the
+// array.  Also, from the perspective of testing, it's probably best
+// if the array is initialized to a short length (e.g., length 4) so
+// that the doubling code is actually exercised.
+
+// One way to implement invalidation is for each OMT to maintain a
+// doubly linked list of OMTCURSORs.  When destroying an OMT or
+// changing the OMT's shape, one can simply step through the list
+// invalidating all the OMTCURSORs.
+
+// The list of OMTCURSORs should use the list.h abstraction.  If it's
+// not clear how to use it, Rich can explain it.
+
 // The programming API:
 
-//typedef struct value *OMTVALUE; // A slight improvement over using void*.
+typedef void *OMTVALUE;
 typedef struct omt *OMT;
 
+typedef struct omtcursor *OMTCURSOR;
+
+
 int toku_omt_create (OMT *omtp);
 // Effect: Create an empty OMT.  Stores it in *omtp.
 // Requires: omtp != NULL
@@ -175,83 +243,107 @@ int toku_omt_delete_at(OMT omt, u_int32_t index);
 // Rationale: To delete an item, first find its index using toku_omt_find, then delete it.
 // Performance: time=O(\log N) amortized.
 
-
-int toku_omt_fetch (OMT V, u_int32_t i, OMTVALUE *v);
+int toku_omt_fetch (OMT V, u_int32_t i, OMTVALUE *v, OMTCURSOR c);
 // Effect: Set *v=V_i
+//   If c != NULL then set c's abstract offset to i.
 // Requires: v   != NULL
 // Returns
 //    0             success
 //    ERANGE        if index>=toku_omt_size(omt)
-// On nonzero return, *v is unchanged.
+//    ENOMEM        if c!=NULL and we run out of memory
+// On nonzero return, *v is unchanged, and c (if nonnull) is either
+//   invalidated or unchanged.
 // Performance: time=O(\log N)
+// Notes: It is possible that c was previously valid and was
+//   associated with a different OMT.   If c is changed by this
+//   function, the function must remove c's association with the old
+//   OMT, and associate it with the new OMT.
 
-int toku_omt_find_zero(OMT V, int (*h)(OMTVALUE, void*extra), void*extra, OMTVALUE *value, u_int32_t *index);
+int toku_omt_find_zero(OMT V, int (*h)(OMTVALUE, void*extra), void*extra, OMTVALUE *value, u_int32_t *index, OMTCURSOR c);
 // Effect:  Find the smallest i such that h(V_i, extra)>=0
+//   If c != NULL and there is such an i then set c's abstract offset to i.
 //  If there is such an i and h(V_i,extra)==0 then set *index=i and return 0.
 //  If there is such an i and h(V_i,extra)>0  then set *index=i and return DB_NOTFOUND.
-//  If there is no such i then set *index=toku_omt_size(V) and return DB_NOTFOUND.
+//  If there is no such i then set *index=toku_omt_size(V), invalidate the cursor (if not NULL), and return DB_NOTFOUND.
 // Requires: index!=NULL
-
-int toku_omt_find(OMT V, int (*h)(OMTVALUE, void*extra), void*extra, int direction, OMTVALUE *value, u_int32_t *index);
-/* Effect:
-    If direction >0 then find the smallest i such that h(V_i,extra)>0.
-    If direction <0 then find the largest  i such that h(V_i,extra)<0.
-    (Direction may not be equal to zero.)
-    If value!=NULL then store V_i in *value
-    If index!=NULL then store i in *index.
-   Requires: The signum of h is monotically increasing.
-   Returns
-      0             success
-      DB_NOTFOUND   no such value is found.
-   On nonzero return, *value and *index are unchanged.
-   Performance: time=O(\log N)
-   Rationale:
-     Here's how to use the find function to find various things
-       Cases for find:
-        find first value:         ( h(v)=+1, direction=+1 )
-        find last value           ( h(v)=-1, direction=-1 )
-        find first X              ( h(v)=(v< x) ? -1 : 1    direction=+1 )
-        find last X               ( h(v)=(v<=x) ? -1 : 1    direction=-1 )
-        find X or successor to X  ( same as find first X. )
-
-   Rationale: To help understand heaviside functions and behavor of find:
-    There are 7 kinds of heaviside functions.
-    The signus of the h must be monotonically increasing.
-    Given a function of the following form, A is the element
-    returned for direction>0, B is the element returned
-    for direction<0, C is the element returned for
-    direction==0 (see find_zero) (with a return of 0), and D is the element
-    returned for direction==0 (see find_zero) with a return of DB_NOTFOUND.
-    If any of A, B, or C are not found, then asking for the
-    associated direction will return DB_NOTFOUND.
-    See find_zero for more information.
-    
-    Let the following represent the signus of the heaviside function.
-
-    -...-
-        A
-         D
-
-    +...+
-    B
-    D
-
-    0...0
-    C
-
-    -...-0...0
-        AC
-
-    0...0+...+
-    C    B
-
-    -...-+...+
-        AB
-         D
-
-    -...-0...0+...+
-        AC    B
-*/
+// Returns
+//    0             success
+//    ENOMEM        if c!=NULL and we run out of memory
+// Performance: time=O(\log N) (calls to h)
+// Notes: It is possible that c was previously valid and was
+//   associated with a different OMT.   If c is changed by this
+//   function, the function must remove c's association with the old
+//   OMT, and associate it with the new OMT.
+// Future directions: the current implementation can be improved, in some cases, by supporting tail recursion.
+//   This would require an additional parameter that represents the current value of the index where the function is recursing,
+//   so that it becomes similar to the way fetch works.
+
+int toku_omt_find(OMT V, int (*h)(OMTVALUE, void*extra), void*extra, int direction, OMTVALUE *value, u_int32_t *index, OMTCURSOR c);
+// Effect:
+//  If direction >0 then find the smallest i such that h(V_i,extra)>0.
+//  If direction <0 then find the largest  i such that h(V_i,extra)<0.
+//  (Direction may not be equal to zero.)
+//  If value!=NULL then store V_i in *value
+//  If index!=NULL then store i in *index.
+//  If c != NULL and there is such an i then set c's abstract offset to i.
+// Requires: The signum of h is monotically increasing.
+// Performance: time=O(\log N) (calls to h)
+// Returns
+//    0             success
+//    DB_NOTFOUND   no such value is found.
+//    ENOMEM        if c!= NULL and we run out of memory
+// On nonzero return, *value and *index are unchanged, and c (if nonnull) is either
+//   invalidated or unchanged.
+// Notes: It is possible that c was previously valid and was
+//   associated with a different OMT.   If c is changed by this
+//   function, the function must remove c's association with the old
+//   OMT, and associate it with the new OMT.
+// Rationale:
+//   Here's how to use the find function to find various things
+//     Cases for find:
+//      find first value:         ( h(v)=+1, direction=+1 )
+//      find last value           ( h(v)=-1, direction=-1 )
+//      find first X              ( h(v)=(v< x) ? -1 : 1    direction=+1 )
+//      find last X               ( h(v)=(v<=x) ? -1 : 1    direction=-1 )
+//      find X or successor to X  ( same as find first X. )
+//
+// Rationale: To help understand heaviside functions and behavor of find:
+//  There are 7 kinds of heaviside functions.
+//  The signum of the h must be monotonically increasing.
+//  Given a function of the following form, A is the element
+//  returned for direction>0, B is the element returned
+//  for direction<0, C is the element returned for
+//  direction==0 (see find_zero) (with a return of 0), and D is the element
+//  returned for direction==0 (see find_zero) with a return of DB_NOTFOUND.
+//  If any of A, B, or C are not found, then asking for the
+//  associated direction will return DB_NOTFOUND.
+//  See find_zero for more information.
+//  
+//  Let the following represent the signum of the heaviside function.
+//
+//  -...-
+//      A
+//       D
+//
+//  +...+
+//  B
+//  D
+//
+//  0...0
+//  C
+//
+//  -...-0...0
+//      AC
+//
+//  0...0+...+
+//  C    B
+//
+//  -...-+...+
+//      AB
+//       D
+//
+//  -...-0...0+...+
+//      AC    B
 
 
 int toku_omt_split_at(OMT omt, OMT *newomt, u_int32_t index);
@@ -282,7 +374,92 @@ void toku_omt_clear(OMT omt);
 //  Note: Will not resize the array, since void precludes allowing a malloc.
 // Performance: time=O(1)
 
-unsigned long toku_omt_memory_size (OMT omt);
+int toku_omt_cursor_create (OMTCURSOR *p);
+// Effect: Create an OMTCURSOR.  Stores it in *p.  The OMTCURSOR is
+// initially invalid.
+// Requires: p != NULL
+// Returns:
+//   0        success
+//   ENOMEM   out of memory (and doesn't modify *omtp)
+// Performance: constant time.
+
+void toku_omt_cursor_destroy (OMTCURSOR *p);
+// Effect:  Invalidates *p (if it is valid) and frees any memory
+// associated with *p.
+//  Also sets *p=NULL.
+// Requires: *p != NULL
+// Rationale:  The usage is to do something like
+//   toku_omt_cursor_destroy(&c);
+// and now c will have a NULL pointer instead of a dangling freed pointer.
+// Rationale: Returns no values since free() cannot fail.
+// Performance:  time=O(1) x #calls to free
+
+int toku_omt_cursor_is_valid (OMTCURSOR c);
+// Effect:  returns 0 iff c is invalid.
+// Performance:  time=O(1)
+
+int toku_omt_cursor_next (OMTCURSOR c, OMTVALUE *v);
+// Effect: Increment c's abstract offset, and store the corresponding value in v.
+// Requires: v != NULL
+// Returns
+//   0 success
+//   EINVAL if the offset goes out of range or c is invalid.
+// On nonzero return, *v is unchanged and c is invalidated.
+// Performance:  time=O(log N) worst case, expected time=O(1) for a randomly
+//  chosen initial position.
+
+int toku_omt_cursor_current (OMTCURSOR c, OMTVALUE *v);
+// Effect: Store in v the value pointed by c's abstract offset
+// Requires: v != NULL
+// Returns
+//  0 success
+//  EINVAL if c is invalid
+// On non-zero return, *v is unchanged
+// Performance: O(1) time
+
+int toku_omt_cursor_prev (OMTCURSOR c, OMTVALUE *v);
+// Effect: Decrement c's abstract offset, and store the corresponding value in v.
+// Requires: v != NULL
+// Returns
+//   0 success
+//   EINVAL if the offset goes out of range or c is invalid.
+// On nonzero return, *v is unchanged and c is invalidated.
+// Performance:  time=O(log N) worst case, expected time=O(1) for a randomly
+//  chosen initial position.
+
+void toku_omt_cursor_invalidate (OMTCURSOR c);
+// Effect: Invalidate c.  (This does not mean that c is destroyed or
+// that its memory is freed.)
+
+// Usage Hint:   The OMTCURSOR is designed to be used inside the
+// BRTcursor.   A BRTcursor includes a pointer to an OMTCURSOR, which
+// is created when the BRTcursor is created.
+//
+// The brt cursor implements its search by first finding a leaf node,
+// containing an OMT.  The BRT then passes its OMTCURSOR into the lookup
+// method (i.e., one of toku_ebdomt_fetch, toku_omt_find_zero,
+// toku_omt_find).  The lookup method, if successful, sets the
+// OMTCURSOR to refer to that element.
+//
+// As long as the OMTCURSOR remains valid, a BRTCURSOR next or prev
+// operation can be implemented using next or prev on the OMTCURSOR.
+//
+// If the OMTCURSOR becomes invalidated, then the BRT must search
+// again from the root of the tree.   The only error that an OMTCURSOR
+// next operation  can raise is that it is invalid.
+//
+// If an element is inserted into the BRT, it may cause an OMTCURSOR
+// to become invalid.  This is especially true if the element will end
+// up in the OMT associated with the cursor.  A simple implementation
+// is to invalidate all OMTCURSORS any time anything is inserted into
+// into the BRT.  Since the BRT already contains a list of BRT cursors
+// associated with it, it is straightforward to go through that list
+// and invalidate all the cursors.
+//
+// When the BRT closes a cursor, it destroys the OMTCURSOR.
+
+
+size_t toku_omt_memory_size (OMT omt);
 // Effect: Return the size (in bytes) of the omt, as it resides in main memory.  Don't include any of the OMTVALUES.
 
 #endif  /* #ifndef OMT_H */

newbrt/recover.c

@@ -607,7 +607,7 @@ void toku_recover_deleteleafentry (LSN lsn, FILENUM filenum, DISKOFF diskoff, u_
     node->log_lsn = lsn;
     {
 	OMTVALUE data = 0;
-	r=toku_omt_fetch(node->u.l.buffer, idx, &data);
+	r=toku_omt_fetch(node->u.l.buffer, idx, &data, NULL);
 	assert(r==0);
 	LEAFENTRY oldleafentry=data;
 	u_int32_t  len = leafentry_memsize(oldleafentry);

newbrt/roll.c

@@ -55,8 +55,8 @@ 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;
-    r = toku_omt_find_zero(txn->open_brts, find_brt_from_filenum, &filenum, &brtv, NULL);
+    OMTVALUE brtv=NULL; // TODO BBB This is not entirely safe. Verify initialization needed.
+    r = toku_omt_find_zero(txn->open_brts, find_brt_from_filenum, &filenum, &brtv, NULL, NULL);
 
     if (r==DB_NOTFOUND) {
 	r = toku_cachefile_root_put_cmd(cf, &brtcmd, toku_txn_logger(txn));

newbrt/tests/omt-test.c

@@ -3,7 +3,8 @@
 #include <errno.h>
 #include <sys/types.h>
 
-typedef struct value *OMTVALUE;
+// typedef struct value *TESTVALUE;
+typedef struct value *TESTVALUE;
 #include "omt.h"
 #include "../newbrt/memory.h"
 #include "../newbrt/toku_assert.h"
@@ -63,7 +64,7 @@ enum create_type {
 
 /* Globals */
 OMT omt;
-OMTVALUE*       values = NULL;
+TESTVALUE*       values = NULL;
 struct value*   nums   = NULL;
 u_int32_t       length;
 
@@ -97,7 +98,7 @@ void init_identity_values(unsigned int seed, u_int32_t num_elements) {
 
     for (i = 0; i < length; i++) {
         nums[i].number   = i;
-        values[i]        = (OMTVALUE)&nums[i];
+        values[i]        = (TESTVALUE)&nums[i];
     }
 }
 
@@ -111,7 +112,7 @@ void init_distinct_sorted_values(unsigned int seed, u_int32_t num_elements) {
     for (i = 0; i < length; i++) {
         number          += (u_int32_t)(random() % 32) + 1;
         nums[i].number   = number;
-        values[i]        = (OMTVALUE)&nums[i];
+        values[i]        = (TESTVALUE)&nums[i];
     }
 }
 
@@ -220,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, values, length);
+        r = toku_omt_create_from_sorted_array(&omt, (OMTVALUE) values, length);
         CKERR(r);
     }
     else if (create_choice == INSERT_AT) {
@@ -241,40 +242,97 @@ void test_create_from_sorted_array_size(enum create_type create_choice, enum clo
     test_close(close);
 }    
 
-void test_fetch_verify (OMT omtree, OMTVALUE* val, u_int32_t len ) {
+void test_fetch_verify (OMT omtree, TESTVALUE* val, u_int32_t len ) {
     u_int32_t i;
+    int j;
     int r;
-    OMTVALUE v = (OMTVALUE)&i;
-    OMTVALUE oldv = v;
+    TESTVALUE v = (TESTVALUE)&i;
+    TESTVALUE oldv = v;
+
+    OMTCURSOR c;
+    r = toku_omt_cursor_create(&c);
+    CKERR(r);
+
     assert(len == toku_omt_size(omtree));
     for (i = 0; i < len; i++) {
         assert(oldv!=val[i]);
         v = NULL;
-        r = toku_omt_fetch(omtree, i, &v);
+        r = toku_omt_fetch(omtree, i, (OMTVALUE) &v, NULL);
+        CKERR(r);
+        assert(v != NULL);
+        assert(v != oldv);
+        assert(v == val[i]);
+        assert(v->number == val[i]->number);
+
+        v = oldv;
+        r = toku_omt_fetch(omtree, i, (OMTVALUE) &v, c);
         CKERR(r);
         assert(v != NULL);
         assert(v != oldv);
         assert(v == val[i]);
         assert(v->number == val[i]->number);
+        assert(toku_omt_cursor_is_valid(c));
  
         v = oldv;
-        r = toku_omt_fetch(omtree, i, &v);
+        r = toku_omt_cursor_current(c, (OMTVALUE) &v);
+        CKERR(r);
+        assert(v != NULL);
+        assert(v != oldv);
+        assert(v == val[i]);
+        assert(v->number == val[i]->number);
+        assert(toku_omt_cursor_is_valid(c));
+
+        v = oldv;
+        j = i + 1;
+        while ((r = toku_omt_cursor_next(c, (OMTVALUE) &v)) == 0) {
+            assert(toku_omt_cursor_is_valid(c));
+            assert(v != NULL);
+            assert(v != oldv);
+            assert(v == val[j]);
+            assert(v->number == val[j]->number);
+            j++;
+            v = oldv;
+        }
+        CKERR2(r, EINVAL);
+        assert(j == (int) len);
+
+        assert(oldv!=val[i]);
+        v = NULL;
+        r = toku_omt_fetch(omtree, i, (OMTVALUE) &v, c);
         CKERR(r);
         assert(v != NULL);
         assert(v != oldv);
         assert(v == val[i]);
         assert(v->number == val[i]->number);
+
+        v = oldv;
+        j = i - 1;
+        while ((r = toku_omt_cursor_prev(c, (OMTVALUE) &v)) == 0) {
+            assert(toku_omt_cursor_is_valid(c));
+            assert(v != NULL);
+            assert(v != oldv);
+            assert(v == val[j]);
+            assert(v->number == val[j]->number);
+            j--;
+            v = oldv;
+        }
+        CKERR2(r, EINVAL);
+        assert(j == -1);
+
     }
+
     for (i = len; i < len*2; i++) {
         v = oldv;
-        r = toku_omt_fetch(omtree, i, &v);
+        r = toku_omt_fetch(omtree, i, (OMTVALUE) &v, NULL);
         CKERR2(r, ERANGE);
         assert(v == oldv);
         v = NULL;
-        r = toku_omt_fetch(omtree, i, &v);
+        r = toku_omt_fetch(omtree, i, (OMTVALUE) &v, c);
         CKERR2(r, ERANGE);
         assert(v == NULL);
     }
+
+    toku_omt_cursor_destroy(&c);
 }
 
 void test_create_fetch_verify(enum create_type create_choice, enum close_when_done close) {
@@ -285,22 +343,22 @@ void test_create_fetch_verify(enum create_type create_choice, enum close_when_do
 
 static int iterate_helper_error_return = 1;
 
-int iterate_helper(OMTVALUE v, u_int32_t idx, void* extra) {
+int iterate_helper(TESTVALUE v, u_int32_t idx, void* extra) {
     if (extra == NULL) return iterate_helper_error_return;
-    OMTVALUE* vals = (OMTVALUE *)extra;
+    TESTVALUE* vals = (TESTVALUE *)extra;
     assert(v != NULL);
     assert(v == vals[idx]);
     assert(v->number == vals[idx]->number);
     return 0;
 }
 
-void test_iterate_verify(OMT omtree, OMTVALUE* vals, u_int32_t len) {
+void test_iterate_verify(OMT omtree, TESTVALUE* vals, u_int32_t len) {
     int r;
     iterate_helper_error_return = 0;
-    r = toku_omt_iterate(omtree, iterate_helper, (void*)vals);
+    r = toku_omt_iterate(omtree, (OMTVALUE) iterate_helper, (void*)vals);
     CKERR(r);
     iterate_helper_error_return = 0xFEEDABBA;
-    r = toku_omt_iterate(omtree, iterate_helper, NULL);
+    r = toku_omt_iterate(omtree, (OMTVALUE) iterate_helper, NULL);
     if (!len) {
         CKERR2(r, 0);
     }
@@ -346,7 +404,7 @@ void test_create_set_at(enum create_type create_choice, enum close_when_done clo
     MALLOC_N(length, perm);
     assert(perm);
 
-    OMTVALUE* old_values = NULL;
+    TESTVALUE* old_values = NULL;
     MALLOC_N(length, old_values);
     assert(old_values);
     
@@ -387,8 +445,8 @@ void test_create_set_at(enum create_type create_choice, enum close_when_done clo
     test_close(close);
 }
 
-int insert_helper(OMTVALUE value, void* extra_insert) {
-    OMTVALUE to_insert = (OMTVALUE)extra_insert;
+int insert_helper(TESTVALUE value, void* extra_insert) {
+    TESTVALUE to_insert = (OMTVALUE)extra_insert;
     assert(to_insert);
 
     if (value->number < to_insert->number) return -1;
@@ -411,11 +469,11 @@ void test_create_insert(enum close_when_done close) {
     length = 0;
     while (length < size) {
         u_int32_t choice = perm[length];
-        OMTVALUE to_insert = &nums[choice];
+        TESTVALUE to_insert = &nums[choice];
         u_int32_t idx = UINT32_MAX;
 
         assert(length==toku_omt_size(omt));
-        r = toku_omt_insert(omt, to_insert, insert_helper, to_insert, &idx);
+        r = toku_omt_insert(omt, to_insert, (OMTVALUE) insert_helper, to_insert, &idx);
         CKERR(r);
         assert(idx <= length);
         if (idx > 0) {
@@ -435,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, insert_helper, to_insert, &idx);
+        r = toku_omt_insert(omt, to_insert, (OMTVALUE) insert_helper, to_insert, &idx);
         CKERR2(r, DB_KEYEXIST);
         assert(idx < length);
         assert(values[idx]->number == to_insert->number);
@@ -588,7 +646,8 @@ typedef struct {
     u_int32_t first_pos;
 } h_extra;
 
-int test_heaviside(OMTVALUE v, void* x) {
+int test_heaviside(OMTVALUE v_omt, void* x) {
+    TESTVALUE v = (OMTVALUE) v_omt;
     h_extra* extra = (h_extra*)x;
     assert(v && x);
     assert(extra->first_zero <= extra->first_pos);
@@ -606,18 +665,23 @@ void heavy_extra(h_extra* extra, u_int32_t first_zero, u_int32_t first_pos) {
 
 void test_find_dir(int dir, void* extra, int (*h)(OMTVALUE, void*),
                    int r_expect, BOOL idx_will_change, u_int32_t idx_expect,
-                   u_int32_t number_expect) {
+                   u_int32_t number_expect, BOOL cursor_valid) {
     u_int32_t idx     = UINT32_MAX;
     u_int32_t old_idx = idx;
-    OMTVALUE omt_val;
+    TESTVALUE omt_val, omt_val_curs;
+    OMTCURSOR c;
     int r;
+    BOOL found;
+
+    r = toku_omt_cursor_create(&c);
+    CKERR(r);
 
     omt_val = NULL;
     if (dir == 0) {
-        r = toku_omt_find_zero(omt, h, extra,      &omt_val, &idx);
+        r = toku_omt_find_zero(omt, h, extra,      (OMTVALUE) &omt_val, &idx, c);
     }
     else {
-        r = toku_omt_find(     omt, h, extra, dir, &omt_val, &idx);
+        r = toku_omt_find(     omt, h, extra, dir, (OMTVALUE) &omt_val, &idx, c);
     }
     CKERR2(r, r_expect);
     if (idx_will_change) {
@@ -628,19 +692,42 @@ void test_find_dir(int dir, void* extra, int (*h)(OMTVALUE, void*),
     }
     if (r == DB_NOTFOUND) {
         assert(omt_val == NULL);
+        found = FALSE;
     }
     else {
         assert(omt_val->number == number_expect);
+        found = TRUE;
     }
  
+    assert(!cursor_valid == !toku_omt_cursor_is_valid(c));
+    if (cursor_valid) {
+        TESTVALUE tmp;
+        assert(idx_will_change);
+        omt_val_curs = NULL;
+        r = toku_omt_cursor_current(c, (OMTVALUE) &omt_val_curs);
+        CKERR(r);
+        assert(toku_omt_cursor_is_valid(c));
+        r = toku_omt_fetch(omt, idx, (OMTVALUE) &tmp, NULL);
+        CKERR(r);
+        if (found) assert(tmp==omt_val);
+        assert(omt_val_curs != NULL);
+        assert(omt_val_curs == tmp);
+        assert(omt_val_curs->number == tmp->number);
+        if (found) assert(omt_val_curs->number==number_expect);
+    }
+
+    toku_omt_cursor_invalidate(c);
+    assert(!toku_omt_cursor_is_valid(c));
+    toku_omt_cursor_destroy(&c);
+
     /* Verify we can pass NULL value. */
     omt_val = NULL;
     idx      = old_idx;
     if (dir == 0) {
-        r = toku_omt_find_zero(omt, h, extra,      NULL, &idx);
+        r = toku_omt_find_zero(omt, h, extra,      NULL, &idx, NULL);
     }
     else {
-        r = toku_omt_find(     omt, h, extra, dir, NULL, &idx);
+        r = toku_omt_find(     omt, h, extra, dir, NULL, &idx, NULL);
     }
     CKERR2(r, r_expect);
     if (idx_will_change) {
@@ -655,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,      &omt_val, NULL);
+        r = toku_omt_find_zero(omt, h, extra,      (OMTVALUE) &omt_val, 0, NULL);
     }
     else {
-        r = toku_omt_find(     omt, h, extra, dir, &omt_val, NULL);
+        r = toku_omt_find(     omt, h, extra, dir, (OMTVALUE) &omt_val, 0, NULL);
     }
     CKERR2(r, r_expect);
     assert(idx == old_idx);
@@ -673,10 +760,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,      NULL, NULL);
+        r = toku_omt_find_zero(omt, h, extra,      NULL, 0, NULL);
     }
     else {
-        r = toku_omt_find(     omt, h, extra, dir, NULL, NULL);
+        r = toku_omt_find(     omt, h, extra, dir, NULL, 0, NULL);
     }
     CKERR2(r, r_expect);
     assert(idx == old_idx);
@@ -693,9 +780,9 @@ void test_find(enum create_type create_choice, enum close_when_done close) {
         A
 */
     heavy_extra(&extra, length, length);
-    test_find_dir(-1, &extra, test_heaviside, 0,           TRUE,  length-1, length-1);
-    test_find_dir(+1, &extra, test_heaviside, DB_NOTFOUND, FALSE, 0,        0);
-    test_find_dir(0,  &extra, test_heaviside, DB_NOTFOUND, TRUE,  length,   length);
+    test_find_dir(-1, &extra, test_heaviside, 0,           TRUE,  length-1, length-1, TRUE);
+    test_find_dir(+1, &extra, test_heaviside, DB_NOTFOUND, FALSE, 0,        0,        FALSE);
+    test_find_dir(0,  &extra, test_heaviside, DB_NOTFOUND, TRUE,  length,   length,   FALSE);
 
 
 /*
@@ -703,54 +790,54 @@ void test_find(enum create_type create_choice, enum close_when_done close) {
     B
 */
     heavy_extra(&extra, 0, 0);
-    test_find_dir(-1, &extra, test_heaviside, DB_NOTFOUND, FALSE, 0, 0);
-    test_find_dir(+1, &extra, test_heaviside, 0,           TRUE,  0, 0);
-    test_find_dir(0,  &extra, test_heaviside, DB_NOTFOUND, TRUE,  0, 0);
+    test_find_dir(-1, &extra, test_heaviside, DB_NOTFOUND, FALSE, 0, 0, FALSE);
+    test_find_dir(+1, &extra, test_heaviside, 0,           TRUE,  0, 0, TRUE);
+    test_find_dir(0,  &extra, test_heaviside, DB_NOTFOUND, TRUE,  0, 0, TRUE);
 
 /*
     0...0
     C
 */
     heavy_extra(&extra, 0, length);
-    test_find_dir(-1, &extra, test_heaviside, DB_NOTFOUND, FALSE, 0, 0);
-    test_find_dir(+1, &extra, test_heaviside, DB_NOTFOUND, FALSE, 0, 0);
-    test_find_dir(0,  &extra, test_heaviside, 0,           TRUE,  0, 0);
+    test_find_dir(-1, &extra, test_heaviside, DB_NOTFOUND, FALSE, 0, 0, FALSE);
+    test_find_dir(+1, &extra, test_heaviside, DB_NOTFOUND, FALSE, 0, 0, FALSE);
+    test_find_dir(0,  &extra, test_heaviside, 0,           TRUE,  0, 0, TRUE);
 
 /*
     -...-0...0
         AC
 */
     heavy_extra(&extra, length/2, length);
-    test_find_dir(-1, &extra, test_heaviside, 0,           TRUE,  length/2-1, length/2-1);
-    test_find_dir(+1, &extra, test_heaviside, DB_NOTFOUND, FALSE, 0,          0);
-    test_find_dir(0,  &extra, test_heaviside, 0,           TRUE,  length/2,   length/2);
+    test_find_dir(-1, &extra, test_heaviside, 0,           TRUE,  length/2-1, length/2-1, TRUE);
+    test_find_dir(+1, &extra, test_heaviside, DB_NOTFOUND, FALSE, 0,          0,          FALSE);
+    test_find_dir(0,  &extra, test_heaviside, 0,           TRUE,  length/2,   length/2,   TRUE);
 
 /*
     0...0+...+
     C    B
 */
     heavy_extra(&extra, 0, length/2);
-    test_find_dir(-1, &extra, test_heaviside, DB_NOTFOUND, FALSE, 0, 0);
-    test_find_dir(+1, &extra, test_heaviside, 0,           TRUE,  length/2, length/2);
-    test_find_dir(0,  &extra, test_heaviside, 0,           TRUE,  0,        0);
+    test_find_dir(-1, &extra, test_heaviside, DB_NOTFOUND, FALSE, 0,        0,        FALSE);
+    test_find_dir(+1, &extra, test_heaviside, 0,           TRUE,  length/2, length/2, TRUE);
+    test_find_dir(0,  &extra, test_heaviside, 0,           TRUE,  0,        0,        TRUE);
 
 /*
     -...-+...+
         AB
 */
     heavy_extra(&extra, length/2, length/2);
-    test_find_dir(-1, &extra, test_heaviside, 0,           TRUE, length/2-1, length/2-1);
-    test_find_dir(+1, &extra, test_heaviside, 0,           TRUE, length/2,   length/2);
-    test_find_dir(0,  &extra, test_heaviside, DB_NOTFOUND, TRUE, length/2,   length/2);
+    test_find_dir(-1, &extra, test_heaviside, 0,           TRUE, length/2-1, length/2-1, TRUE);
+    test_find_dir(+1, &extra, test_heaviside, 0,           TRUE, length/2,   length/2,   TRUE);
+    test_find_dir(0,  &extra, test_heaviside, DB_NOTFOUND, TRUE, length/2,   length/2,   TRUE);
 
 /*
     -...-0...0+...+
         AC    B
 */    
     heavy_extra(&extra, length/3, 2*length/3);
-    test_find_dir(-1, &extra, test_heaviside, 0, TRUE,   length/3-1,   length/3-1);
-    test_find_dir(+1, &extra, test_heaviside, 0, TRUE, 2*length/3,   2*length/3);
-    test_find_dir(0,  &extra, test_heaviside, 0, TRUE,   length/3,     length/3);
+    test_find_dir(-1, &extra, test_heaviside, 0, TRUE,   length/3-1,   length/3-1, TRUE);
+    test_find_dir(+1, &extra, test_heaviside, 0, TRUE, 2*length/3,   2*length/3,   TRUE);
+    test_find_dir(0,  &extra, test_heaviside, 0, TRUE,   length/3,     length/3,   TRUE);
 
     /* Cleanup */
     test_close(close);