#2946 merge le-cursor's to main

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

newbrt/Makefile

@@ -59,6 +59,7 @@ BRT_SOURCES = \
   key \
   leafentry \
   leaflock \
+  le-cursor \
   logfilemgr \
   logger \
   log_code \

newbrt/brt-internal.h

@@ -313,6 +313,7 @@ struct brt_cursor {
     u_int64_t  root_put_counter; // what was the count on the BRT when we validated the cursor?
     TXNID      oldest_living_xid;// what was the oldest live txnid when we created the cursor?
     BOOL is_snapshot_read; // true if query is read_committed, false otherwise
+    BOOL is_leaf_mode;
     TOKUTXN ttxn;
     struct brt_cursor_leaf_info  leaf_info;
 };

newbrt/brt.c

@@ -4055,7 +4055,11 @@ static inline int brt_cursor_extract_key_and_val(
                    u_int32_t *vallen,
                    void     **val) {
     int r = 0;
-    if (cursor->is_snapshot_read) {
+    if (toku_brt_cursor_is_leaf_mode(cursor)) {
+        *key = le_key_and_len(le, keylen);
+        *val = le;
+        *vallen = leafentry_memsize(le);
+    } else if (cursor->is_snapshot_read) {
         le_iterate_val(
             le, 
             does_txn_read_entry, 
@@ -4063,9 +4067,8 @@ static inline int brt_cursor_extract_key_and_val(
             vallen, 
             cursor->ttxn
             );
-        *key = le_key_and_len(le,keylen);
-    }
-    else {
+        *key = le_key_and_len(le, keylen);
+    } else {
         *key = le_key_and_len(le, keylen);
         *val = le_latest_val_and_len(le, vallen);
     }
@@ -4147,6 +4150,7 @@ int toku_brt_cursor (
     cursor->prefetching = FALSE;
     cursor->oldest_living_xid = ttxn ? toku_logger_get_oldest_living_xid(ttxn->logger) : TXNID_NONE;
     cursor->is_snapshot_read = is_snapshot_read;
+    cursor->is_leaf_mode = FALSE;
     cursor->ttxn = ttxn;
     toku_list_push(&brt->cursors, &cursor->cursors_link);
     int r = toku_omt_cursor_create(&cursor->omtcursor);
@@ -4158,6 +4162,16 @@ int toku_brt_cursor (
     return 0;
 }
 
+void
+toku_brt_cursor_set_leaf_mode(BRT_CURSOR brtcursor) {
+    brtcursor->is_leaf_mode = TRUE;
+}
+
+int
+toku_brt_cursor_is_leaf_mode(BRT_CURSOR brtcursor) {
+    return brtcursor->is_leaf_mode;
+}
+
 // Called during cursor destruction
 // It is the same as brt_cursor_invalidate, except that
 // we make sure the callback function is never called.
@@ -4299,7 +4313,7 @@ brt_search_leaf_node(BRTNODE node, brt_search_t *search, BRT_GET_CALLBACK_FUNCTI
     if (r!=0) return r;
 
     LEAFENTRY le = datav;
-    if (is_le_val_empty(le,brtcursor)) {
+    if (!toku_brt_cursor_is_leaf_mode(brtcursor) && is_le_val_empty(le,brtcursor)) {
         // Provisionally deleted stuff is gone.
         // So we need to scan in the direction to see if we can find something
         while (1) {
@@ -4708,7 +4722,7 @@ brt_cursor_shortcut (BRT_CURSOR cursor, int direction, u_int32_t limit, BRT_GET_
             r = toku_omt_fetch(omt, index, &le, NULL);
             assert(r==0);
 
-            if (!is_le_val_empty(le,cursor)) {
+            if (toku_brt_cursor_is_leaf_mode(cursor) || !is_le_val_empty(le, cursor)) {
                 maybe_do_implicit_promotion_on_query(cursor, le);
                 u_int32_t keylen;
                 void     *key;

newbrt/brt.h

@@ -131,6 +131,8 @@ int toku_verify_brt (BRT brt)  __attribute__ ((warn_unused_result));
 
 typedef struct brt_cursor *BRT_CURSOR;
 int toku_brt_cursor (BRT, BRT_CURSOR*, TOKUTXN, BOOL)  __attribute__ ((warn_unused_result));
+void toku_brt_cursor_set_leaf_mode(BRT_CURSOR);
+int toku_brt_cursor_is_leaf_mode(BRT_CURSOR);
 
 // get is deprecated in favor of the individual functions below
 int toku_brt_cursor_get (BRT_CURSOR cursor, DBT *key, BRT_GET_CALLBACK_FUNCTION getf, void *getf_v, int get_flags)  __attribute__ ((warn_unused_result));

newbrt/le-cursor.c

+/* -*- mode: C; c-basic-offset: 4 -*- */
+#ident "Copyright (c) 2010 Tokutek Inc.  All rights reserved."
+#ident "The technology is licensed by the Massachusetts Institute of Technology, Rutgers State University of New Jersey, and the Research Foundation of State University of New York at Stony Brook under United States of America Serial No. 11/760379 and to the patents and/or patent applications resulting from it."
+
+#include "includes.h"
+#include "le-cursor.h"
+
+// A LE_CURSOR is a special type of BRT_CURSOR that retrieves all of the leaf entries in a BRT.
+// It caches the key that is was last positioned over to speed up key comparisions.
+
+struct le_cursor {
+    BRT_CURSOR brt_cursor;
+    DBT key;
+    BOOL neg_infinity, pos_infinity;
+};
+
+int 
+le_cursor_create(LE_CURSOR *le_cursor_result, BRT brt, TOKUTXN txn) {
+    int result = 0;
+    LE_CURSOR le_cursor = (LE_CURSOR) toku_malloc(sizeof (struct le_cursor));
+    if (le_cursor == NULL)
+        result = errno;
+    else {
+        result = toku_brt_cursor(brt, &le_cursor->brt_cursor, txn, FALSE);
+        if (result == 0) {
+            toku_brt_cursor_set_leaf_mode(le_cursor->brt_cursor);
+            toku_init_dbt(&le_cursor->key); le_cursor->key.flags = DB_DBT_REALLOC;
+            le_cursor->neg_infinity = TRUE;
+            le_cursor->pos_infinity = FALSE;
+        }
+    }
+
+    if (result == 0)
+        *le_cursor_result = le_cursor;
+    else
+        toku_free(le_cursor);
+
+    return result;
+}
+
+int 
+le_cursor_close(LE_CURSOR le_cursor) {
+    int result = 0;
+    result = toku_brt_cursor_close(le_cursor->brt_cursor);
+    toku_destroy_dbt(&le_cursor->key);
+    toku_free(le_cursor);
+    return result;
+}
+
+struct le_cursor_callback_arg {
+    DBT *key, *val;
+};
+
+static int
+le_cursor_callback(ITEMLEN keylen, bytevec key, ITEMLEN vallen, bytevec val, void *v) {
+    struct le_cursor_callback_arg *arg = (struct le_cursor_callback_arg *) v;
+    toku_dbt_set(keylen, key, arg->key, NULL);
+    toku_dbt_set(vallen, val, arg->val, NULL);
+    return 0;
+}
+
+int 
+le_cursor_next(LE_CURSOR le_cursor, DBT *key, DBT *val) {
+    le_cursor->neg_infinity = FALSE;
+    struct le_cursor_callback_arg arg = { &le_cursor->key, val };
+    int error = toku_brt_cursor_get(le_cursor->brt_cursor, NULL, le_cursor_callback, &arg, DB_NEXT);
+    if (error == 0 && key != NULL)
+        toku_dbt_set(le_cursor->key.size, le_cursor->key.data, key, NULL);
+    else if (error == DB_NOTFOUND)
+        le_cursor->pos_infinity = TRUE;
+    return error;
+}
+
+int
+is_key_right_of_le_cursor(LE_CURSOR le_cursor, DBT *key, int (*keycompare)(DB *, const DBT *, const DBT *), DB *db) {
+    int result;
+    if (le_cursor->neg_infinity)
+        result = TRUE;
+    else if (le_cursor->pos_infinity)
+        return FALSE;
+    else {
+        int r = keycompare(db, &le_cursor->key, key);
+        if (r < 0)
+            result = TRUE;
+        else
+            result = FALSE;
+    }
+    return result;
+}
+    
+

newbrt/le-cursor.h

+/* -*- mode: C; c-basic-offset: 4 -*- */
+#ident "Copyright (c) 2010 Tokutek Inc.  All rights reserved."
+#ident "The technology is licensed by the Massachusetts Institute of Technology, Rutgers State University of New Jersey, and the Research Foundation of State University of New York at Stony Brook under United States of America Serial No. 11/760379 and to the patents and/or patent applications resulting from it."
+
+// A leaf entry cursor (LE_CURSOR) is a special type of BRT_CURSOR that visits all of the leaf entries in a tree
+// and returns the leaf entry to the caller.  It maintains a copy of the key that it was last positioned over to
+// speed up key comparisions with a given key.  For example, the hot indexing could use the _key_right_of_cursor
+// function to determine where a given key sits relative to the LE_CURSOR position.
+
+// When _next and _key_right_of_cursor functions are run on multiple threads, they must be protected by a lock.  This
+// lock is assumed to exist outside of the LE_CURSOR.
+
+typedef struct le_cursor *LE_CURSOR;
+
+// Create a leaf cursor for a tree (brt) within a transaction (txn)
+// Success: returns 0, stores the LE_CURSOR in the le_cursor_result
+// Failure: returns a non-zero error number
+int le_cursor_create(LE_CURSOR *le_cursor_result, BRT brt, TOKUTXN txn);
+
+// Close and free the LE_CURSOR
+// Success: returns 0
+// Failure: returns a non-zero error number
+int le_cursor_close(LE_CURSOR le_cursor);
+
+// Retrieve the next leaf entry under the LE_CURSOR
+// Success: returns zero, stores the leaf entry key into the key dbt, and the leaf entry into the val dbt
+// Failure: returns a non-zero error number
+int le_cursor_next(LE_CURSOR le_cursor, DBT *key, DBT *val);
+
+// Return TRUE if the key is to the right of the LE_CURSOR position
+// Otherwise returns FALSE
+// The LE_CURSOR position is intialized to -infinity. Any key comparision with -infinity returns TRUE.
+// When the cursor runs off the right edge of the tree, the LE_CURSOR position is set to +infinity.  Any key comparision with +infinity
+// returns FALSE.
+int is_key_right_of_le_cursor(LE_CURSOR le_cursor, DBT *key, int (*keycompare)(DB *extra, const DBT *, const DBT *), DB *extra);
+

newbrt/tests/le-cursor-provdel.c

+/* -*- mode: C; c-basic-offset: 4 -*- */
+#ident "Copyright (c) 2010 Tokutek Inc.  All rights reserved."
+
+// test the LE_CURSOR next function with provisionally deleted rows
+
+#include "includes.h"
+#include "checkpoint.h"
+#include "le-cursor.h"
+#include "test.h"
+
+static TOKUTXN const null_txn = 0;
+static DB * const null_db = 0;
+
+static int 
+test_brt_cursor_keycompare(DB *UU(db), const DBT *a, const DBT *b) {
+    return toku_keycompare(a->data, a->size, b->data, b->size);
+}
+
+static void
+txn_yield(voidfp UU(f), void *UU(fv), void *UU(v)) {
+    if (f)
+        f(fv);
+}
+
+// create a tree and populate it with n rows
+static void
+create_populate_tree(const char *logdir, const char *fname, int n) {
+    if (verbose) fprintf(stderr, "%s %s %s %d\n", __FUNCTION__, logdir, fname, n);
+    int error;
+
+    TOKULOGGER logger = NULL;
+    error = toku_logger_create(&logger);
+    assert(error == 0);
+    error = toku_logger_open(logdir, logger);
+    assert(error == 0);
+    CACHETABLE ct = NULL;
+    error = toku_brt_create_cachetable(&ct, 0, ZERO_LSN, logger);
+    assert(error == 0);
+    toku_logger_set_cachetable(logger, ct);
+    error = toku_logger_open_rollback(logger, ct, TRUE);
+    assert(error == 0);
+
+    TOKUTXN txn = NULL;
+    error = toku_txn_begin_txn(NULL, &txn, logger, TXN_SNAPSHOT_NONE);
+    assert(error == 0);
+
+    BRT brt = NULL;
+    error = toku_open_brt(fname, 1, &brt, 1<<12, ct, txn, test_brt_cursor_keycompare, null_db);
+    assert(error == 0);
+
+    error = toku_txn_commit_txn(txn, TRUE, txn_yield, NULL, NULL, NULL);
+    assert(error == 0);
+    toku_txn_close_txn(txn);
+
+    txn = NULL;
+    error = toku_txn_begin_txn(NULL, &txn, logger, TXN_SNAPSHOT_NONE);
+    assert(error == 0);
+
+    // insert keys 0, 1, 2, .. (n-1)
+    for (int i = 0; i < n; i++) {
+        int k = toku_htonl(i);
+        int v = i;
+        DBT key;
+        toku_fill_dbt(&key, &k, sizeof k);
+        DBT val;
+        toku_fill_dbt(&val, &v, sizeof v);
+        error = toku_brt_insert(brt, &key, &val, txn);
+        assert(error == 0);
+    }
+
+    error = toku_txn_commit_txn(txn, TRUE, txn_yield, NULL, NULL, NULL);
+    assert(error == 0);
+    toku_txn_close_txn(txn);
+
+    error = toku_close_brt(brt, NULL);
+    assert(error == 0);
+
+    error = toku_checkpoint(ct, logger, NULL, NULL, NULL, NULL);
+    assert(error == 0);
+
+    error = toku_logger_close_rollback(logger, FALSE);
+    assert(error == 0);
+    error = toku_logger_close(&logger);
+    assert(error == 0);
+
+    error = toku_cachetable_close(&ct);
+    assert(error == 0);
+}
+
+// provionally delete all of the even keys
+// the LE_CURSOR should see all of the leaf entries
+static void 
+test_provdel(const char *logdir, const char *fname, int n) {
+    if (verbose) fprintf(stderr, "%s %s %s %d\n", __FUNCTION__, logdir, fname, n);
+    int error;
+
+    TOKULOGGER logger = NULL;
+    error = toku_logger_create(&logger);
+    assert(error == 0);
+    error = toku_logger_open(logdir, logger);
+    assert(error == 0);
+    CACHETABLE ct = NULL;
+    error = toku_brt_create_cachetable(&ct, 0, ZERO_LSN, logger);
+    assert(error == 0);
+    toku_logger_set_cachetable(logger, ct);
+    error = toku_logger_open_rollback(logger, ct, FALSE);
+    assert(error == 0);
+
+    TOKUTXN txn = NULL;
+    error = toku_txn_begin_txn(NULL, &txn, logger, TXN_SNAPSHOT_NONE);
+    assert(error == 0);
+
+    BRT brt = NULL;
+    error = toku_open_brt(fname, 1, &brt, 1<<12, ct, txn, test_brt_cursor_keycompare, null_db);
+    assert(error == 0);
+
+    error = toku_txn_commit_txn(txn, TRUE, txn_yield, NULL, NULL, NULL);
+    assert(error == 0);
+    toku_txn_close_txn(txn);
+
+    txn = NULL;
+    error = toku_txn_begin_txn(NULL, &txn, logger, TXN_SNAPSHOT_NONE);
+    assert(error == 0);
+
+    // del keys 0, 2, 4, ...
+    for (int i = 0; i < n; i += 2) {
+        int k = toku_htonl(i);
+        DBT key;
+        toku_fill_dbt(&key, &k, sizeof k);
+        error = toku_brt_delete(brt, &key, txn);
+        assert(error == 0);
+    }
+
+    TOKUTXN cursortxn = NULL;
+    error = toku_txn_begin_txn(NULL, &cursortxn, logger, TXN_SNAPSHOT_NONE);
+    assert(error == 0);
+
+    LE_CURSOR cursor = NULL;
+    error = le_cursor_create(&cursor, brt, cursortxn);
+    assert(error == 0);
+
+    DBT key;
+    toku_init_dbt(&key); key.flags = DB_DBT_REALLOC;
+    DBT val;
+    toku_init_dbt(&val); val.flags = DB_DBT_REALLOC;
+
+    int i;
+    for (i=0; ; i++) {
+        error = le_cursor_next(cursor, &key, &val);
+        if (error != 0) 
+            break;
+        assert(key.size == sizeof (int));
+        int ii;
+        memcpy(&ii, key.data, key.size);
+        assert((int) toku_htonl(i) == ii);
+        
+        LEAFENTRY le = (LEAFENTRY) val.data;
+        assert(le->attributes == LE_MVCC);
+    }
+    assert(i == n);
+
+    toku_destroy_dbt(&key);
+    toku_destroy_dbt(&val);
+
+    error = le_cursor_close(cursor);
+    assert(error == 0);
+
+    error = toku_txn_commit_txn(cursortxn, TRUE, txn_yield, NULL, NULL, NULL);
+    assert(error == 0);
+    toku_txn_close_txn(cursortxn);
+
+    error = toku_txn_commit_txn(txn, TRUE, txn_yield, NULL, NULL, NULL);
+    assert(error == 0);
+    toku_txn_close_txn(txn);
+
+    error = toku_close_brt(brt, NULL);
+    assert(error == 0);
+
+    error = toku_checkpoint(ct, logger, NULL, NULL, NULL, NULL);
+    assert(error == 0);
+
+    error = toku_logger_close_rollback(logger, FALSE);
+    assert(error == 0);
+    error = toku_logger_close(&logger);
+    assert(error == 0);
+
+    error = toku_cachetable_close(&ct);
+    assert(error == 0);
+}
+
+static void
+init_logdir(const char *logdir) {
+    int error;
+
+    char cmd[32+strlen(logdir)];
+    sprintf(cmd, "rm -rf %s", logdir);
+    error = system(cmd);
+    assert(error == 0);
+
+    error = toku_os_mkdir(logdir, 0777);
+    assert(error == 0);
+}
+
+int
+test_main (int argc , const char *argv[]) {
+    default_parse_args(argc, argv);
+
+    const char *logdir = "dir." __FILE__;
+    init_logdir(logdir);
+    int error = chdir(logdir);
+    assert(error == 0);
+
+    const int n = 10;
+    const char *brtfile =  __FILE__ ".brt";
+    create_populate_tree(".", brtfile, n);
+    test_provdel(".", brtfile, n);
+
+    return 0;
+}

newbrt/tests/le-cursor-right.c

+/* -*- mode: C; c-basic-offset: 4 -*- */
+#ident "Copyright (c) 2010 Tokutek Inc.  All rights reserved."
+
+// test the LE_CURSOR is_key_right_of_le_cursor function
+// - LE_CURSOR at neg infinity
+// - LE_CURSOR at pos infinity
+// - LE_CURSOR somewhere else
+
+#include "includes.h"
+#include "checkpoint.h"
+#include "le-cursor.h"
+#include "test.h"
+
+static TOKUTXN const null_txn = 0;
+static DB * const null_db = 0;
+
+static int 
+test_keycompare(DB *UU(db), const DBT *a, const DBT *b) {
+    return toku_keycompare(a->data, a->size, b->data, b->size);
+}
+
+static void
+txn_yield(voidfp UU(f), void *UU(fv), void *UU(v)) {
+    if (f)
+        f(fv);
+}
+
+// create a tree and populate it with n rows
+static void
+create_populate_tree(const char *logdir, const char *fname, int n) {
+    if (verbose) fprintf(stderr, "%s %s %s %d\n", __FUNCTION__, logdir, fname, n);
+    int error;
+
+    TOKULOGGER logger = NULL;
+    error = toku_logger_create(&logger);
+    assert(error == 0);
+    error = toku_logger_open(logdir, logger);
+    assert(error == 0);
+    CACHETABLE ct = NULL;
+    error = toku_brt_create_cachetable(&ct, 0, ZERO_LSN, logger);
+    assert(error == 0);
+    toku_logger_set_cachetable(logger, ct);
+    error = toku_logger_open_rollback(logger, ct, TRUE);
+    assert(error == 0);
+
+    TOKUTXN txn = NULL;
+    error = toku_txn_begin_txn(NULL, &txn, logger, TXN_SNAPSHOT_NONE);
+    assert(error == 0);
+
+    BRT brt = NULL;
+    error = toku_open_brt(fname, 1, &brt, 1<<12, ct, txn, test_keycompare, null_db);
+    assert(error == 0);
+
+    error = toku_txn_commit_txn(txn, TRUE, txn_yield, NULL, NULL, NULL);
+    assert(error == 0);
+    toku_txn_close_txn(txn);
+
+    txn = NULL;
+    error = toku_txn_begin_txn(NULL, &txn, logger, TXN_SNAPSHOT_NONE);
+    assert(error == 0);
+
+    // insert keys 0, 1, 2, .. (n-1)
+    for (int i = 0; i < n; i++) {
+        int k = toku_htonl(i);
+        int v = i;
+        DBT key;
+        toku_fill_dbt(&key, &k, sizeof k);
+        DBT val;
+        toku_fill_dbt(&val, &v, sizeof v);
+        error = toku_brt_insert(brt, &key, &val, txn);
+        assert(error == 0);
+    }
+
+    error = toku_txn_commit_txn(txn, TRUE, txn_yield, NULL, NULL, NULL);
+    assert(error == 0);
+    toku_txn_close_txn(txn);
+
+    error = toku_close_brt(brt, NULL);
+    assert(error == 0);
+
+    error = toku_checkpoint(ct, logger, NULL, NULL, NULL, NULL);
+    assert(error == 0);
+    error = toku_logger_close_rollback(logger, FALSE);
+    assert(error == 0);
+    error = toku_logger_close(&logger);
+    assert(error == 0);
+
+    error = toku_cachetable_close(&ct);
+    assert(error == 0);
+}
+
+// test is_key_right_of_le_cursor when the LE_CURSOR is positioned at -infinity
+static void 
+test_neg_infinity(const char *fname, int n) {
+    if (verbose) fprintf(stderr, "%s %s %d\n", __FUNCTION__, fname, n);
+    int error;
+
+    CACHETABLE ct = NULL;
+    error = toku_brt_create_cachetable(&ct, 0, ZERO_LSN, NULL_LOGGER);
+    assert(error == 0);
+
+    BRT brt = NULL;
+    error = toku_open_brt(fname, 1, &brt, 1<<12, ct, null_txn, test_keycompare, null_db);
+    assert(error == 0);
+
+    // position the cursor at -infinity
+    LE_CURSOR cursor = NULL;
+    error = le_cursor_create(&cursor, brt, NULL);
+    assert(error == 0);
+
+    for (int i = 0; i < 2*n; i++) {
+        int k = toku_htonl(i);
+        DBT key;
+        toku_fill_dbt(&key, &k, sizeof k);
+        int right = is_key_right_of_le_cursor(cursor, &key, test_keycompare, null_db);
+        assert(right == TRUE);
+    }
+        
+    error = le_cursor_close(cursor);
+    assert(error == 0);
+
+    error = toku_close_brt(brt, 0);
+    assert(error == 0);
+
+    error = toku_cachetable_close(&ct);
+    assert(error == 0);
+}
+
+// test is_key_right_of_le_cursor when the LE_CURSOR is positioned at +infinity
+static void 
+test_pos_infinity(const char *fname, int n) {
+    if (verbose) fprintf(stderr, "%s %s %d\n", __FUNCTION__, fname, n);
+    int error;
+
+    CACHETABLE ct = NULL;
+    error = toku_brt_create_cachetable(&ct, 0, ZERO_LSN, NULL_LOGGER);
+    assert(error == 0);
+
+    BRT brt = NULL;
+    error = toku_open_brt(fname, 1, &brt, 1<<12, ct, null_txn, test_keycompare, null_db);
+    assert(error == 0);
+
+    // position the LE_CURSOR at +infinity
+    LE_CURSOR cursor = NULL;
+    error = le_cursor_create(&cursor, brt, NULL);
+    assert(error == 0);
+
+    DBT key;
+    toku_init_dbt(&key); key.flags = DB_DBT_REALLOC;
+    DBT val;
+    toku_init_dbt(&val); val.flags = DB_DBT_REALLOC;
+
+    int i;
+    for (i = 0; ; i++) {
+        error = le_cursor_next(cursor, &key, &val);
+        if (error != 0) 
+            break;
+        assert(key.size == sizeof (int));
+        int ii;
+        memcpy(&ii, key.data, key.size);
+        assert((int) toku_htonl(i) == ii);
+        
+        LEAFENTRY le = (LEAFENTRY) val.data;
+        assert(le->attributes == LE_MVCC);
+    }
+    assert(i == n);
+
+    toku_destroy_dbt(&key);
+    toku_destroy_dbt(&val);
+
+    for (i = 0; i < 2*n; i++) {
+        int k = toku_htonl(i);
+        DBT key2;
+        toku_fill_dbt(&key2, &k, sizeof k);
+        int right = is_key_right_of_le_cursor(cursor, &key2, test_keycompare, null_db);
+        assert(right == FALSE);
+    }
+
+    error = le_cursor_close(cursor);
+    assert(error == 0);
+
+    error = toku_close_brt(brt, 0);
+    assert(error == 0);
+
+    error = toku_cachetable_close(&ct);
+    assert(error == 0);
+}
+
+// test is_key_right_of_le_cursor when the LE_CURSOR is positioned in between -infinity and +infinity
+static void 
+test_between(const char *fname, int n) {
+    if (verbose) fprintf(stderr, "%s %s %d\n", __FUNCTION__, fname, n);
+    int error;
+
+    CACHETABLE ct = NULL;
+    error = toku_brt_create_cachetable(&ct, 0, ZERO_LSN, NULL_LOGGER);
+    assert(error == 0);
+
+    BRT brt = NULL;
+    error = toku_open_brt(fname, 1, &brt, 1<<12, ct, null_txn, test_keycompare, null_db);
+    assert(error == 0);
+
+    // position the LE_CURSOR at +infinity
+    LE_CURSOR cursor = NULL;
+    error = le_cursor_create(&cursor, brt, NULL);
+    assert(error == 0);
+
+    DBT key;
+    toku_init_dbt(&key); key.flags = DB_DBT_REALLOC;
+    DBT val;
+    toku_init_dbt(&val); val.flags = DB_DBT_REALLOC;
+
+    int i;
+    for (i = 0; ; i++) {
+        // move the LE_CURSOR forward
+        error = le_cursor_next(cursor, &key, &val);
+        if (error != 0) 
+            break;
+        assert(key.size == sizeof (int));
+        int ii;
+        memcpy(&ii, key.data, key.size);
+        assert((int) toku_htonl(i) == ii);
+        
+        LEAFENTRY le = (LEAFENTRY) val.data;
+        assert(le->attributes == LE_MVCC);
+
+        // test that 0 .. i is not right of the cursor
+        for (int j = 0; j <= i; j++) {
+            int k = toku_htonl(j);
+            DBT key2;
+            toku_fill_dbt(&key2, &k, sizeof k);
+            int right = is_key_right_of_le_cursor(cursor, &key2, test_keycompare, null_db);
+            assert(right == FALSE);
+        }
+
+        // test that i+1 .. n is left of the cursor
+        for (int j = i + 1; j <= n; j++) {
+            int k = toku_htonl(j);
+            DBT key2;
+            toku_fill_dbt(&key2, &k, sizeof k);
+            int right = is_key_right_of_le_cursor(cursor, &key2, test_keycompare, null_db);
+            assert(right == TRUE);
+        }
+
+    }
+    assert(i == n);
+
+    toku_destroy_dbt(&key);
+    toku_destroy_dbt(&val);
+
+    error = le_cursor_close(cursor);
+    assert(error == 0);
+
+    error = toku_close_brt(brt, 0);
+    assert(error == 0);
+
+    error = toku_cachetable_close(&ct);
+    assert(error == 0);
+}
+
+static void
+init_logdir(const char *logdir) {
+    int error;
+
+    char cmd[32+strlen(logdir)];
+    sprintf(cmd, "rm -rf %s", logdir);
+    error = system(cmd);
+    assert(error == 0);
+
+    error = toku_os_mkdir(logdir, 0777);
+    assert(error == 0);
+}
+
+int
+test_main (int argc , const char *argv[]) {
+    default_parse_args(argc, argv);
+
+    const char *logdir = "dir." __FILE__;
+    init_logdir(logdir);
+    int error = chdir(logdir);
+    assert(error == 0);
+
+    const int n = 10;
+    const char *brtfile =  __FILE__ ".brt";
+    create_populate_tree(".", brtfile, n);
+    test_neg_infinity(brtfile, n);
+    test_pos_infinity(brtfile, n);
+    test_between(brtfile, n);
+
+    return 0;
+}

newbrt/tests/le-cursor-walk.c

+/* -*- mode: C; c-basic-offset: 4 -*- */
+#ident "Copyright (c) 2010 Tokutek Inc.  All rights reserved."
+
+// test the LE_CURSOR next function
+
+#include "includes.h"
+#include "checkpoint.h"
+#include "le-cursor.h"
+#include "test.h"
+
+static TOKUTXN const null_txn = 0;
+static DB * const null_db = 0;
+
+static int 
+test_brt_cursor_keycompare(DB *UU(db), const DBT *a, const DBT *b) {
+    return toku_keycompare(a->data, a->size, b->data, b->size);
+}
+
+static void
+txn_yield(voidfp UU(f), void *UU(fv), void *UU(v)) {
+    if (f)
+        f(fv);
+}
+
+// create a tree and populate it with n rows
+static void
+create_populate_tree(const char *logdir, const char *fname, int n) {
+    if (verbose) fprintf(stderr, "%s %s %s %d\n", __FUNCTION__, logdir, fname, n);
+    int error;
+
+    TOKULOGGER logger = NULL;
+    error = toku_logger_create(&logger);
+    assert(error == 0);
+    error = toku_logger_open(logdir, logger);
+    assert(error == 0);
+    CACHETABLE ct = NULL;
+    error = toku_brt_create_cachetable(&ct, 0, ZERO_LSN, logger);
+    assert(error == 0);
+    toku_logger_set_cachetable(logger, ct);
+    error = toku_logger_open_rollback(logger, ct, TRUE);
+    assert(error == 0);
+
+    TOKUTXN txn = NULL;
+    error = toku_txn_begin_txn(NULL, &txn, logger, TXN_SNAPSHOT_NONE);
+    assert(error == 0);
+
+    BRT brt = NULL;
+    error = toku_open_brt(fname, 1, &brt, 1<<12, ct, txn, test_brt_cursor_keycompare, null_db);
+    assert(error == 0);
+
+    error = toku_txn_commit_txn(txn, TRUE, txn_yield, NULL, NULL, NULL);
+    assert(error == 0);
+    toku_txn_close_txn(txn);
+
+    txn = NULL;
+    error = toku_txn_begin_txn(NULL, &txn, logger, TXN_SNAPSHOT_NONE);
+    assert(error == 0);
+
+    // insert keys 0, 1, 2, .. (n-1)
+    for (int i = 0; i < n; i++) {
+        int k = toku_htonl(i);
+        int v = i;
+        DBT key;
+        toku_fill_dbt(&key, &k, sizeof k);
+        DBT val;
+        toku_fill_dbt(&val, &v, sizeof v);
+        error = toku_brt_insert(brt, &key, &val, txn);
+        assert(error == 0);
+    }
+
+    error = toku_txn_commit_txn(txn, TRUE, txn_yield, NULL, NULL, NULL);
+    assert(error == 0);
+    toku_txn_close_txn(txn);
+
+    error = toku_close_brt(brt, NULL);
+    assert(error == 0);
+
+    error = toku_checkpoint(ct, logger, NULL, NULL, NULL, NULL);
+    assert(error == 0);
+    error = toku_logger_close_rollback(logger, FALSE);
+    assert(error == 0);
+    error = toku_logger_close(&logger);
+    assert(error == 0);
+
+    error = toku_cachetable_close(&ct);
+    assert(error == 0);
+}
+
+// retrieve all of the leaf entries in the the tree and verify the key associated with each one.
+// there should be n leaf entires in the tree.
+static void 
+walk_tree(const char *fname, int n) {
+    if (verbose) fprintf(stderr, "%s %s %d\n", __FUNCTION__, fname, n);
+    int error;
+
+    CACHETABLE ct = NULL;
+    error = toku_brt_create_cachetable(&ct, 0, ZERO_LSN, NULL_LOGGER);
+    assert(error == 0);
+
+    BRT brt = NULL;
+    error = toku_open_brt(fname, 1, &brt, 1<<12, ct, null_txn, test_brt_cursor_keycompare, null_db);
+    assert(error == 0);
+
+    LE_CURSOR cursor = NULL;
+    error = le_cursor_create(&cursor, brt, NULL);
+    assert(error == 0);
+
+    DBT key;
+    toku_init_dbt(&key); key.flags = DB_DBT_REALLOC;
+    DBT val;
+    toku_init_dbt(&val); val.flags = DB_DBT_REALLOC;
+
+    int i;
+    for (i = 0; ; i++) {
+        error = le_cursor_next(cursor, &key, &val);
+        if (error != 0) 
+            break;
+        assert(key.size == sizeof (int));
+        int ii;
+        memcpy(&ii, key.data, key.size);
+        assert((int) toku_htonl(i) == ii);
+        
+        LEAFENTRY le = (LEAFENTRY) val.data;
+        assert(le->attributes == LE_MVCC);
+    }
+    assert(i == n);
+
+    toku_destroy_dbt(&key);
+    toku_destroy_dbt(&val);
+
+    error = le_cursor_close(cursor);
+    assert(error == 0);
+
+    error = toku_close_brt(brt, 0);
+    assert(error == 0);
+
+    error = toku_cachetable_close(&ct);
+    assert(error == 0);
+}
+
+static void
+init_logdir(const char *logdir) {
+    int error;
+
+    char cmd[32+strlen(logdir)];
+    sprintf(cmd, "rm -rf %s", logdir);
+    error = system(cmd);
+    assert(error == 0);
+
+    error = toku_os_mkdir(logdir, 0777);
+    assert(error == 0);
+}
+
+int
+test_main (int argc , const char *argv[]) {
+    default_parse_args(argc, argv);
+
+    const char *logdir = "dir." __FILE__;
+    init_logdir(logdir);
+    int error = chdir(logdir);
+    assert(error == 0);
+
+    const int n = 1000;
+    const char *brtfile =  __FILE__ ".brt";
+    create_populate_tree(".", brtfile, n);
+    walk_tree(brtfile, n);
+
+    return 0;
+}

src/ydb-le-cursor.h

+#include "le_cursor.h"
+
+static inline int le_cursor_create_db_txn(LE_CURSOR *le_cursor_result, DB *db, DB_TXN *txn) {
+    return le_cursor_create(le_cursor_result, db_struct_i(db)->brt, db_txn_struct_i(txn)->tokutxn);
+}
+