Addresses #1956 refs[t:1956] Add c_getf_set_range_reverse,...

Addresses #1956 refs[t:1956] Add c_getf_set_range_reverse, c_getf_get_both_range_reverse, (and c_get versions)
Equivalent to non-reverse but just changes the search direction

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

buildheader/db.h_4_1

@@ -105,6 +105,8 @@ typedef enum {
 #define DB_SET 30
 #define DB_SET_RANGE 32
 #define DB_CURRENT_BINDING 253
+#define DB_SET_RANGE_REVERSE 252
+#define DB_GET_BOTH_RANGE_REVERSE 251
 #define DB_RMW 1073741824
 #define DB_PRELOCKED 0x00800000
 #define DB_PRELOCKED_WRITE 0x00400000
@@ -300,9 +302,11 @@ struct __toku_dbc {
   int (*c_getf_heaviside)(DBC *, u_int32_t, YDB_HEAVISIDE_CALLBACK_FUNCTION f, void *extra_f, YDB_HEAVISIDE_FUNCTION h, void *extra_h, int direction);
   int (*c_getf_set)(DBC *, u_int32_t, DBT *, YDB_CALLBACK_FUNCTION, void *);
   int (*c_getf_set_range)(DBC *, u_int32_t, DBT *, YDB_CALLBACK_FUNCTION, void *);
+  int (*c_getf_set_range_reverse)(DBC *, u_int32_t, DBT *, YDB_CALLBACK_FUNCTION, void *);
   int (*c_getf_get_both)(DBC *, u_int32_t, DBT *, DBT *, YDB_CALLBACK_FUNCTION, void *);
   int (*c_getf_get_both_range)(DBC *, u_int32_t, DBT *, DBT *, YDB_CALLBACK_FUNCTION, void *);
-  void* __toku_dummy0[4];
+  int (*c_getf_get_both_range_reverse)(DBC *, u_int32_t, DBT *, DBT *, YDB_CALLBACK_FUNCTION, void *);
+  void* __toku_dummy0[2];
   char __toku_dummy1[104];
   int (*c_close) (DBC *); /* 32-bit offset=188 size=4, 64=bit offset=272 size=8 */
   int (*c_count) (DBC *, db_recno_t *, u_int32_t); /* 32-bit offset=192 size=4, 64=bit offset=280 size=8 */

buildheader/db.h_4_3

@@ -106,6 +106,8 @@ typedef enum {
 #define DB_SET 28
 #define DB_SET_RANGE 30
 #define DB_CURRENT_BINDING 253
+#define DB_SET_RANGE_REVERSE 252
+#define DB_GET_BOTH_RANGE_REVERSE 251
 #define DB_RMW 536870912
 #define DB_PRELOCKED 0x00800000
 #define DB_PRELOCKED_WRITE 0x00400000
@@ -317,16 +319,17 @@ struct __toku_dbc {
   int (*c_getf_heaviside)(DBC *, u_int32_t, YDB_HEAVISIDE_CALLBACK_FUNCTION f, void *extra_f, YDB_HEAVISIDE_FUNCTION h, void *extra_h, int direction);
   int (*c_getf_set)(DBC *, u_int32_t, DBT *, YDB_CALLBACK_FUNCTION, void *);
   int (*c_getf_set_range)(DBC *, u_int32_t, DBT *, YDB_CALLBACK_FUNCTION, void *);
+  int (*c_getf_set_range_reverse)(DBC *, u_int32_t, DBT *, YDB_CALLBACK_FUNCTION, void *);
   int (*c_getf_get_both)(DBC *, u_int32_t, DBT *, DBT *, YDB_CALLBACK_FUNCTION, void *);
   int (*c_getf_get_both_range)(DBC *, u_int32_t, DBT *, DBT *, YDB_CALLBACK_FUNCTION, void *);
-  void* __toku_dummy0[2];
-  char __toku_dummy1[112];
+  int (*c_getf_get_both_range_reverse)(DBC *, u_int32_t, DBT *, DBT *, YDB_CALLBACK_FUNCTION, void *);
+  char __toku_dummy0[112];
   int (*c_close) (DBC *); /* 32-bit offset=188 size=4, 64=bit offset=264 size=8 */
   int (*c_count) (DBC *, db_recno_t *, u_int32_t); /* 32-bit offset=192 size=4, 64=bit offset=272 size=8 */
   int (*c_del) (DBC *, u_int32_t); /* 32-bit offset=196 size=4, 64=bit offset=280 size=8 */
-  void* __toku_dummy2[1];
+  void* __toku_dummy1[1];
   int (*c_get) (DBC *, DBT *, DBT *, u_int32_t); /* 32-bit offset=204 size=4, 64=bit offset=296 size=8 */
-  void* __toku_dummy3[10]; /* Padding at the end */ 
+  void* __toku_dummy2[10]; /* Padding at the end */ 
 };
 #ifdef _TOKUDB_WRAP_H
 #define txn_begin txn_begin_tokudb

buildheader/db.h_4_4

@@ -107,6 +107,8 @@ typedef enum {
 #define DB_SET 28
 #define DB_SET_RANGE 30
 #define DB_CURRENT_BINDING 253
+#define DB_SET_RANGE_REVERSE 252
+#define DB_GET_BOTH_RANGE_REVERSE 251
 #define DB_RMW 536870912
 #define DB_PRELOCKED 0x00800000
 #define DB_PRELOCKED_WRITE 0x00400000
@@ -324,9 +326,11 @@ struct __toku_dbc {
   int (*c_getf_heaviside)(DBC *, u_int32_t, YDB_HEAVISIDE_CALLBACK_FUNCTION f, void *extra_f, YDB_HEAVISIDE_FUNCTION h, void *extra_h, int direction);
   int (*c_getf_set)(DBC *, u_int32_t, DBT *, YDB_CALLBACK_FUNCTION, void *);
   int (*c_getf_set_range)(DBC *, u_int32_t, DBT *, YDB_CALLBACK_FUNCTION, void *);
+  int (*c_getf_set_range_reverse)(DBC *, u_int32_t, DBT *, YDB_CALLBACK_FUNCTION, void *);
   int (*c_getf_get_both)(DBC *, u_int32_t, DBT *, DBT *, YDB_CALLBACK_FUNCTION, void *);
   int (*c_getf_get_both_range)(DBC *, u_int32_t, DBT *, DBT *, YDB_CALLBACK_FUNCTION, void *);
-  void* __toku_dummy0[4];
+  int (*c_getf_get_both_range_reverse)(DBC *, u_int32_t, DBT *, DBT *, YDB_CALLBACK_FUNCTION, void *);
+  void* __toku_dummy0[2];
   char __toku_dummy1[104];
   int (*c_close) (DBC *); /* 32-bit offset=188 size=4, 64=bit offset=272 size=8 */
   int (*c_count) (DBC *, db_recno_t *, u_int32_t); /* 32-bit offset=192 size=4, 64=bit offset=280 size=8 */

buildheader/db.h_4_5

@@ -107,6 +107,8 @@ typedef enum {
 #define DB_SET 25
 #define DB_SET_RANGE 27
 #define DB_CURRENT_BINDING 253
+#define DB_SET_RANGE_REVERSE 252
+#define DB_GET_BOTH_RANGE_REVERSE 251
 #define DB_RMW 1073741824
 #define DB_PRELOCKED 0x00800000
 #define DB_PRELOCKED_WRITE 0x00400000
@@ -324,9 +326,11 @@ struct __toku_dbc {
   int (*c_getf_heaviside)(DBC *, u_int32_t, YDB_HEAVISIDE_CALLBACK_FUNCTION f, void *extra_f, YDB_HEAVISIDE_FUNCTION h, void *extra_h, int direction);
   int (*c_getf_set)(DBC *, u_int32_t, DBT *, YDB_CALLBACK_FUNCTION, void *);
   int (*c_getf_set_range)(DBC *, u_int32_t, DBT *, YDB_CALLBACK_FUNCTION, void *);
+  int (*c_getf_set_range_reverse)(DBC *, u_int32_t, DBT *, YDB_CALLBACK_FUNCTION, void *);
   int (*c_getf_get_both)(DBC *, u_int32_t, DBT *, DBT *, YDB_CALLBACK_FUNCTION, void *);
   int (*c_getf_get_both_range)(DBC *, u_int32_t, DBT *, DBT *, YDB_CALLBACK_FUNCTION, void *);
-  void* __toku_dummy0[8];
+  int (*c_getf_get_both_range_reverse)(DBC *, u_int32_t, DBT *, DBT *, YDB_CALLBACK_FUNCTION, void *);
+  void* __toku_dummy0[6];
   char __toku_dummy1[104];
   int (*c_close) (DBC *); /* 32-bit offset=204 size=4, 64=bit offset=304 size=8 */
   int (*c_count) (DBC *, db_recno_t *, u_int32_t); /* 32-bit offset=208 size=4, 64=bit offset=312 size=8 */

buildheader/db.h_4_6

@@ -108,6 +108,8 @@ typedef enum {
 #define DB_SET 26
 #define DB_SET_RANGE 27
 #define DB_CURRENT_BINDING 253
+#define DB_SET_RANGE_REVERSE 252
+#define DB_GET_BOTH_RANGE_REVERSE 251
 #define DB_RMW 1073741824
 #define DB_PRELOCKED 0x00800000
 #define DB_PRELOCKED_WRITE 0x00400000
@@ -329,9 +331,11 @@ struct __toku_dbc {
   int (*c_getf_heaviside)(DBC *, u_int32_t, YDB_HEAVISIDE_CALLBACK_FUNCTION f, void *extra_f, YDB_HEAVISIDE_FUNCTION h, void *extra_h, int direction);
   int (*c_getf_set)(DBC *, u_int32_t, DBT *, YDB_CALLBACK_FUNCTION, void *);
   int (*c_getf_set_range)(DBC *, u_int32_t, DBT *, YDB_CALLBACK_FUNCTION, void *);
+  int (*c_getf_set_range_reverse)(DBC *, u_int32_t, DBT *, YDB_CALLBACK_FUNCTION, void *);
   int (*c_getf_get_both)(DBC *, u_int32_t, DBT *, DBT *, YDB_CALLBACK_FUNCTION, void *);
   int (*c_getf_get_both_range)(DBC *, u_int32_t, DBT *, DBT *, YDB_CALLBACK_FUNCTION, void *);
-  void* __toku_dummy0[18];
+  int (*c_getf_get_both_range_reverse)(DBC *, u_int32_t, DBT *, DBT *, YDB_CALLBACK_FUNCTION, void *);
+  void* __toku_dummy0[16];
   char __toku_dummy1[104];
   int (*c_close) (DBC *); /* 32-bit offset=244 size=4, 64=bit offset=384 size=8 */
   int (*c_count) (DBC *, db_recno_t *, u_int32_t); /* 32-bit offset=248 size=4, 64=bit offset=392 size=8 */

buildheader/make_db_h.c

@@ -119,6 +119,8 @@ void print_defines (void) {
     dodefine(DB_SET);
     dodefine(DB_SET_RANGE);
     printf("#define DB_CURRENT_BINDING 253\n"); // private tokudb
+    printf("#define DB_SET_RANGE_REVERSE 252\n"); // private tokudb
+    printf("#define DB_GET_BOTH_RANGE_REVERSE 251\n"); // private tokudb
     dodefine(DB_RMW);
     printf("#define DB_PRELOCKED 0x00800000\n"); // private tokudb
     printf("#define DB_PRELOCKED_WRITE 0x00400000\n"); // private tokudb
@@ -424,8 +426,10 @@ int main (int argc __attribute__((__unused__)), char *argv[] __attribute__((__un
                                  "YDB_HEAVISIDE_FUNCTION h, void *extra_h, int direction)",
 			     "int (*c_getf_set)(DBC *, u_int32_t, DBT *, YDB_CALLBACK_FUNCTION, void *)",
 			     "int (*c_getf_set_range)(DBC *, u_int32_t, DBT *, YDB_CALLBACK_FUNCTION, void *)",
+			     "int (*c_getf_set_range_reverse)(DBC *, u_int32_t, DBT *, YDB_CALLBACK_FUNCTION, void *)",
 			     "int (*c_getf_get_both)(DBC *, u_int32_t, DBT *, DBT *, YDB_CALLBACK_FUNCTION, void *)",
 			     "int (*c_getf_get_both_range)(DBC *, u_int32_t, DBT *, DBT *, YDB_CALLBACK_FUNCTION, void *)",
+			     "int (*c_getf_get_both_range_reverse)(DBC *, u_int32_t, DBT *, DBT *, YDB_CALLBACK_FUNCTION, void *)",
 			     NULL};
 	assert(sizeof(dbc_fields32)==sizeof(dbc_fields64));
 	print_struct("dbc", 1, dbc_fields32, dbc_fields64, sizeof(dbc_fields32)/sizeof(dbc_fields32[0]), extra);

buildheader/tdb.h

@@ -108,6 +108,8 @@ typedef enum {
 #define DB_SET 26
 #define DB_SET_RANGE 27
 #define DB_CURRENT_BINDING 253
+#define DB_SET_RANGE_REVERSE 252
+#define DB_GET_BOTH_RANGE_REVERSE 251
 #define DB_RMW 1073741824
 #define DB_PRELOCKED 0x00800000
 #define DB_PRELOCKED_WRITE 0x00400000
@@ -263,8 +265,10 @@ struct __toku_dbc {
   int (*c_getf_heaviside)(DBC *, u_int32_t, YDB_HEAVISIDE_CALLBACK_FUNCTION f, void *extra_f, YDB_HEAVISIDE_FUNCTION h, void *extra_h, int direction);
   int (*c_getf_set)(DBC *, u_int32_t, DBT *, YDB_CALLBACK_FUNCTION, void *);
   int (*c_getf_set_range)(DBC *, u_int32_t, DBT *, YDB_CALLBACK_FUNCTION, void *);
+  int (*c_getf_set_range_reverse)(DBC *, u_int32_t, DBT *, YDB_CALLBACK_FUNCTION, void *);
   int (*c_getf_get_both)(DBC *, u_int32_t, DBT *, DBT *, YDB_CALLBACK_FUNCTION, void *);
   int (*c_getf_get_both_range)(DBC *, u_int32_t, DBT *, DBT *, YDB_CALLBACK_FUNCTION, void *);
+  int (*c_getf_get_both_range_reverse)(DBC *, u_int32_t, DBT *, DBT *, YDB_CALLBACK_FUNCTION, void *);
   int (*c_close) (DBC *);
   int (*c_count) (DBC *, db_recno_t *, u_int32_t);
   int (*c_del) (DBC *, u_int32_t);

include/db.h

@@ -108,6 +108,8 @@ typedef enum {
 #define DB_SET 26
 #define DB_SET_RANGE 27
 #define DB_CURRENT_BINDING 253
+#define DB_SET_RANGE_REVERSE 252
+#define DB_GET_BOTH_RANGE_REVERSE 251
 #define DB_RMW 1073741824
 #define DB_PRELOCKED 0x00800000
 #define DB_PRELOCKED_WRITE 0x00400000
@@ -263,8 +265,10 @@ struct __toku_dbc {
   int (*c_getf_heaviside)(DBC *, u_int32_t, YDB_HEAVISIDE_CALLBACK_FUNCTION f, void *extra_f, YDB_HEAVISIDE_FUNCTION h, void *extra_h, int direction);
   int (*c_getf_set)(DBC *, u_int32_t, DBT *, YDB_CALLBACK_FUNCTION, void *);
   int (*c_getf_set_range)(DBC *, u_int32_t, DBT *, YDB_CALLBACK_FUNCTION, void *);
+  int (*c_getf_set_range_reverse)(DBC *, u_int32_t, DBT *, YDB_CALLBACK_FUNCTION, void *);
   int (*c_getf_get_both)(DBC *, u_int32_t, DBT *, DBT *, YDB_CALLBACK_FUNCTION, void *);
   int (*c_getf_get_both_range)(DBC *, u_int32_t, DBT *, DBT *, YDB_CALLBACK_FUNCTION, void *);
+  int (*c_getf_get_both_range_reverse)(DBC *, u_int32_t, DBT *, DBT *, YDB_CALLBACK_FUNCTION, void *);
   int (*c_close) (DBC *);
   int (*c_count) (DBC *, db_recno_t *, u_int32_t);
   int (*c_del) (DBC *, u_int32_t);

newbrt/brt.c

@@ -4078,6 +4078,21 @@ toku_brt_cursor_get_both_range(BRT_CURSOR cursor, DBT *key, DBT *val, BRT_GET_CA
     return brt_cursor_search_eq_k_x(cursor, &search, getf, getf_v, logger);
 }
 
+static int brt_cursor_compare_get_both_range_reverse(brt_search_t *search, DBT *x, DBT *y) {
+    BRT brt = search->context;
+    int keycmp = compare_k_x(brt, search->k, x);
+    if (keycmp > 0)
+        return 1;
+    else
+        return keycmp == 0 && (y == 0 || compare_v_y(brt, search->v, y) >= 0); /* return min xy: k >= x && v >= y */
+}
+
+int
+toku_brt_cursor_get_both_range_reverse(BRT_CURSOR cursor, DBT *key, DBT *val, BRT_GET_CALLBACK_FUNCTION getf, void *getf_v, TOKULOGGER logger)
+{
+    brt_search_t search; brt_search_init(&search, brt_cursor_compare_get_both_range_reverse, BRT_SEARCH_RIGHT, key, val, cursor->brt);
+    return brt_cursor_search_eq_k_x(cursor, &search, getf, getf_v, logger);
+}
 
 static int
 brt_cursor_prev_shortcut (BRT_CURSOR cursor, BRT_GET_CALLBACK_FUNCTION getf, void *getf_v)
@@ -4167,6 +4182,19 @@ toku_brt_cursor_set_range(BRT_CURSOR cursor, DBT *key, BRT_GET_CALLBACK_FUNCTION
     return brt_cursor_search(cursor, &search, getf, getf_v, logger);
 }
 
+static int brt_cursor_compare_set_range_reverse(brt_search_t *search, DBT *x, DBT *y) {
+    BRT brt = search->context;
+    return compare_kv_xy(brt, search->k, search->v, x, y) >= 0; /* return kv >= xy */
+}
+
+int
+toku_brt_cursor_set_range_reverse(BRT_CURSOR cursor, DBT *key, BRT_GET_CALLBACK_FUNCTION getf, void *getf_v, TOKULOGGER logger)
+{
+    brt_search_t search; brt_search_init(&search, brt_cursor_compare_set_range_reverse, BRT_SEARCH_RIGHT, key, NULL, cursor->brt);
+    return brt_cursor_search(cursor, &search, getf, getf_v, logger);
+}
+
+
 //TODO: When tests have been rewritten, get rid of this function.
 //Only used by tests.
 int

newbrt/brt.h

@@ -100,7 +100,9 @@ int toku_brt_cursor_prev_dup(BRT_CURSOR cursor, BRT_GET_CALLBACK_FUNCTION getf,
 int toku_brt_cursor_current(BRT_CURSOR cursor, int op, BRT_GET_CALLBACK_FUNCTION getf, void *getf_v, TOKULOGGER logger);
 int toku_brt_cursor_set(BRT_CURSOR cursor, DBT *key, DBT *val, BRT_GET_CALLBACK_FUNCTION getf, void *getf_v, TOKULOGGER logger);
 int toku_brt_cursor_set_range(BRT_CURSOR cursor, DBT *key, BRT_GET_CALLBACK_FUNCTION getf, void *getf_v, TOKULOGGER logger);
+int toku_brt_cursor_set_range_reverse(BRT_CURSOR cursor, DBT *key, BRT_GET_CALLBACK_FUNCTION getf, void *getf_v, TOKULOGGER logger);
 int toku_brt_cursor_get_both_range(BRT_CURSOR cursor, DBT *key, DBT *val, BRT_GET_CALLBACK_FUNCTION getf, void *getf_v, TOKULOGGER logger);
+int toku_brt_cursor_get_both_range_reverse(BRT_CURSOR cursor, DBT *key, DBT *val, BRT_GET_CALLBACK_FUNCTION getf, void *getf_v, TOKULOGGER logger);
 
 
 typedef struct {

src/tests/test_db_txn_locks_nonheaviside.c

@@ -348,103 +348,103 @@ test_first (u_int32_t dup_flags) {
 }
 
 static void
-test_set_range (u_int32_t dup_flags) {
+test_set_range (u_int32_t dup_flags, u_int32_t flag, int i) {
     /* ********************************************************************** */
     setup_dbs(dup_flags);
-    cget(TRUE, FALSE, 'a', 1, 1, 0, 0, DB_SET_RANGE);
+    cget(TRUE, FALSE, 'a', i*1, i*1, 0, 0, flag);
     close_dbs();
     /* ********************************************************************** */
     setup_dbs(dup_flags);
-    cget(TRUE, FALSE, 'a', 1, 1, 0, 0, DB_SET_RANGE);
-    cget(TRUE, FALSE, 'a', 2, 1, 0, 0, DB_SET_RANGE);
+    cget(TRUE, FALSE, 'a', i*1, i*1, 0, 0, flag);
+    cget(TRUE, FALSE, 'a', i*2, i*1, 0, 0, flag);
     close_dbs();
     /* ********************************************************************** */
     setup_dbs(dup_flags);
-    cget(TRUE, FALSE, 'a', 1, 1, 0, 0, DB_SET_RANGE);
-    cget(TRUE, FALSE, 'a', 1, 1, 0, 0, DB_SET_RANGE);
+    cget(TRUE, FALSE, 'a', i*1, i*1, 0, 0, flag);
+    cget(TRUE, FALSE, 'a', i*1, i*1, 0, 0, flag);
     close_dbs();
     /* ********************************************************************** */
     setup_dbs(dup_flags);
-    cget(TRUE, FALSE, 'a', 1, 1, 0, 0, DB_SET_RANGE);
-    cget(TRUE, FALSE, 'b', 2, 1, 0, 0, DB_SET_RANGE);
+    cget(TRUE, FALSE, 'a', i*1, i*1, 0, 0, flag);
+    cget(TRUE, FALSE, 'b', i*2, i*1, 0, 0, flag);
     close_dbs();
     /* ********************************************************************** */
     setup_dbs(dup_flags);
-    cget(TRUE, FALSE, 'a', 1, 1, 0, 0, DB_SET_RANGE);
-    cget(TRUE, FALSE, 'b', 1, 1, 0, 0, DB_SET_RANGE);
+    cget(TRUE, FALSE, 'a', i*1, i*1, 0, 0, flag);
+    cget(TRUE, FALSE, 'b', i*1, i*1, 0, 0, flag);
     close_dbs();
     /* ********************************************************************** */
     setup_dbs(dup_flags);
-    cget(TRUE, FALSE, 'a', 1, 1, 0, 0, DB_SET_RANGE);
-    cget(TRUE, FALSE, 'b', 5, 5, 0, 0, DB_SET_RANGE);
-    put(FALSE, 'a', 7, 6);
-    put(FALSE, 'a', 5, 5);
-    put(TRUE,  'a', 4, 4);
-    put(TRUE,  'b', -1, 4);
-    put(FALSE,  'b', 2, 4);
-    put(FALSE, 'a', 5, 4);
+    cget(TRUE, FALSE, 'a', i*1, i*1, 0, 0, flag);
+    cget(TRUE, FALSE, 'b', i*5, i*5, 0, 0, flag);
+    put(FALSE, 'a', i*7, i*6);
+    put(FALSE, 'a', i*5, i*5);
+    put(TRUE,  'a', i*4, i*4);
+    put(TRUE,  'b', -i*1, i*4);
+    put(FALSE,  'b', i*2, i*4);
+    put(FALSE, 'a', i*5, i*4);
     early_commit('b');
-    put(TRUE, 'a', 7, 6);
-    put(TRUE, 'a', 5, 5);
-    put(TRUE,  'a', 4, 4);
-    put(TRUE, 'a', 5, 4);
-    cget(TRUE, TRUE, 'a', 1, 1, 4, 4, DB_SET_RANGE);
-    cget(TRUE, TRUE, 'a', 2, 1, 4, 4, DB_SET_RANGE);
-    cget(FALSE, TRUE, 'c', 6, 6, 7, 6, DB_SET_RANGE);
+    put(TRUE, 'a', i*7, i*6);
+    put(TRUE, 'a', i*5, i*5);
+    put(TRUE,  'a', i*4, i*4);
+    put(TRUE, 'a', i*5, i*4);
+    cget(TRUE, TRUE, 'a', i*1, i*1, i*4, i*4, flag);
+    cget(TRUE, TRUE, 'a', i*2, i*1, i*4, i*4, flag);
+    cget(FALSE, TRUE, 'c', i*6, i*6, i*7, i*6, flag);
     early_commit('a');
-    cget(TRUE, TRUE, 'c', 6, 6, 7, 6, DB_SET_RANGE);
+    cget(TRUE, TRUE, 'c', i*6, i*6, i*7, i*6, flag);
     close_dbs();
 }
 
 static void
-test_both_range (u_int32_t dup_flags) {
+test_both_range (u_int32_t dup_flags, u_int32_t flag, int i) {
     if (dup_flags == 0) {
-      test_both(dup_flags, DB_GET_BOTH_RANGE);
+      test_both(dup_flags, flag);
       return;
     }
     /* ********************************************************************** */
     setup_dbs(dup_flags);
-    cget(TRUE, FALSE, 'a', 1, 1, 0, 0, DB_GET_BOTH_RANGE);
+    cget(TRUE, FALSE, 'a', i*1, i*1, 0, 0, flag);
     close_dbs();
     /* ********************************************************************** */
     setup_dbs(dup_flags);
-    cget(TRUE, FALSE, 'a', 1, 1, 0, 0, DB_GET_BOTH_RANGE);
-    cget(TRUE, FALSE, 'a', 2, 1, 0, 0, DB_GET_BOTH_RANGE);
+    cget(TRUE, FALSE, 'a', i*1, i*1, 0, 0, flag);
+    cget(TRUE, FALSE, 'a', i*2, i*1, 0, 0, flag);
     close_dbs();
     /* ********************************************************************** */
     setup_dbs(dup_flags);
-    cget(TRUE, FALSE, 'a', 1, 1, 0, 0, DB_GET_BOTH_RANGE);
-    cget(TRUE, FALSE, 'a', 1, 1, 0, 0, DB_GET_BOTH_RANGE);
+    cget(TRUE, FALSE, 'a', i*1, i*1, 0, 0, flag);
+    cget(TRUE, FALSE, 'a', i*1, i*1, 0, 0, flag);
     close_dbs();
     /* ********************************************************************** */
     setup_dbs(dup_flags);
-    cget(TRUE, FALSE, 'a', 1, 1, 0, 0, DB_GET_BOTH_RANGE);
-    cget(TRUE, FALSE, 'b', 2, 1, 0, 0, DB_GET_BOTH_RANGE);
+    cget(TRUE, FALSE, 'a', i*1, i*1, 0, 0, flag);
+    cget(TRUE, FALSE, 'b', i*2, i*1, 0, 0, flag);
     close_dbs();
     /* ********************************************************************** */
     setup_dbs(dup_flags);
-    cget(TRUE, FALSE, 'a', 1, 1, 0, 0, DB_GET_BOTH_RANGE);
-    cget(TRUE, FALSE, 'b', 1, 1, 0, 0, DB_GET_BOTH_RANGE);
+    cget(TRUE, FALSE, 'a', i*1, i*1, 0, 0, flag);
+    cget(TRUE, FALSE, 'b', i*1, i*1, 0, 0, flag);
     close_dbs();
     /* ********************************************************************** */
     setup_dbs(dup_flags);
-    cget(TRUE, FALSE, 'a', 1, 1, 0, 0, DB_GET_BOTH_RANGE);
-    cget(TRUE, FALSE, 'b', 5, 5, 0, 0, DB_GET_BOTH_RANGE);
-    put(TRUE, 'a', 5, 0);
-    put(FALSE, 'a', 5, 5);
-    put(FALSE, 'a', 5, 6);
-    put(TRUE,  'a', 6, 0);
-    put(TRUE,  'b', 1, 0);
+    cget(TRUE, FALSE, 'a', i*1, i*1, 0, 0, flag);
+    cget(TRUE, FALSE, 'b', i*5, i*5, 0, 0, flag);
+    put(TRUE, 'a', i*5, 0);
+    put(FALSE, 'a', i*5, i*5);
+    put(FALSE, 'a', i*5, i*6);
+    put(TRUE,  'a', i*6, 0);
+    put(TRUE,  'b', i*1, 0);
     early_commit('b');
-    put(TRUE, 'a', 5, 0);
-    put(TRUE, 'a', 5, 5);
-    put(TRUE, 'a', 5, 6);
-    put(TRUE,  'a', 6, 0);
-    cget(TRUE, FALSE, 'a', 1, 1, 4, 4, DB_GET_BOTH_RANGE);
-    cget(TRUE,  TRUE, 'a', 1, 0, 1, 0, DB_GET_BOTH_RANGE);
-    cget(FALSE, TRUE, 'c', 5, 5, 5, 5, DB_GET_BOTH_RANGE);
+    put(TRUE, 'a', i*5, 0);
+    put(TRUE, 'a', i*5, i*5);
+    put(TRUE, 'a', i*5, i*6);
+    put(TRUE,  'a', i*6, 0);
+    cget(TRUE, FALSE, 'a', i*1, i*1, i*4, i*4, flag);
+    cget(TRUE,  TRUE, 'a', i*1, 0, i*1, 0, flag);
+    cget(FALSE, TRUE, 'c', i*5, i*5, i*5, i*5, flag);
     early_commit('a');
-    cget(TRUE, TRUE, 'c', 5, 5, 5, 5, DB_GET_BOTH_RANGE);
+    cget(TRUE, TRUE, 'c', i*5, i*5, i*5, i*5, flag);
     close_dbs();
 }
 
@@ -657,9 +657,15 @@ test (u_int32_t dup_flags) {
     /* ********************************************************************** */
     test_last(dup_flags);
     /* ********************************************************************** */
-    test_set_range(dup_flags);
+    test_set_range(dup_flags, DB_SET_RANGE, 1);
+#ifdef DB_SET_RANGE_REVERSE
+    test_set_range(dup_flags, DB_SET_RANGE_REVERSE, -1);
+#endif
     /* ********************************************************************** */
-    test_both_range(dup_flags);
+    test_both_range(dup_flags, DB_GET_BOTH_RANGE, 1);
+#ifdef DB_GET_BOTH_RANGE_REVERSE
+    test_both_range(dup_flags, DB_GET_BOTH_RANGE_REVERSE, -1);
+#endif
     /* ********************************************************************** */
     test_next(dup_flags, DB_NEXT);
     test_next(dup_flags, DB_NEXT_NODUP);

src/ydb.c

@@ -142,8 +142,10 @@ static int toku_c_getf_current_binding(DBC *c, u_int32_t flag, YDB_CALLBACK_FUNC
 
 static int toku_c_getf_set(DBC *c, u_int32_t flag, DBT *key, YDB_CALLBACK_FUNCTION f, void *extra);
 static int toku_c_getf_set_range(DBC *c, u_int32_t flag, DBT *key, YDB_CALLBACK_FUNCTION f, void *extra);
+static int toku_c_getf_set_range_reverse(DBC *c, u_int32_t flag, DBT *key, YDB_CALLBACK_FUNCTION f, void *extra);
 static int toku_c_getf_get_both(DBC *c, u_int32_t flag, DBT *key, DBT *val, YDB_CALLBACK_FUNCTION f, void *extra);
 static int toku_c_getf_get_both_range(DBC *c, u_int32_t flag, DBT *key, DBT *val, YDB_CALLBACK_FUNCTION f, void *extra);
+static int toku_c_getf_get_both_range_reverse(DBC *c, u_int32_t flag, DBT *key, DBT *val, YDB_CALLBACK_FUNCTION f, void *extra);
 
 static int toku_c_getf_heaviside(DBC *c, u_int32_t flags,
                                  YDB_HEAVISIDE_CALLBACK_FUNCTION f, void *extra_f,
@@ -1469,6 +1471,10 @@ toku_c_get(DBC* c, DBT* key, DBT* val, u_int32_t flag) {
             query_context_wrapped_init(&context, c, key,  val);
             r = toku_c_getf_set_range(c, remaining_flags, key, c_get_wrapper_callback, &context);
             break;
+        case (DB_SET_RANGE_REVERSE):
+            query_context_wrapped_init(&context, c, key,  val);
+            r = toku_c_getf_set_range_reverse(c, remaining_flags, key, c_get_wrapper_callback, &context);
+            break;
         case (DB_GET_BOTH):
             query_context_wrapped_init(&context, c, NULL, NULL);
             r = toku_c_getf_get_both(c, remaining_flags, key, val, c_get_wrapper_callback, &context);
@@ -1480,6 +1486,13 @@ toku_c_get(DBC* c, DBT* key, DBT* val, u_int32_t flag) {
             else                  query_context_wrapped_init(&context, c, NULL, val);
             r = toku_c_getf_get_both_range(c, remaining_flags, key, val, c_get_wrapper_callback, &context);
             break;
+        case (DB_GET_BOTH_RANGE_REVERSE):
+            //For a nodup database, DB_GET_BOTH_RANGE_REVERSE is an alias for DB_GET_BOTH.
+            //DB_GET_BOTH(_RANGE_REVERSE) require different contexts (see case(DB_GET_BOTH)).
+            if (c_db_is_nodup(c)) query_context_wrapped_init(&context, c, NULL, NULL);
+            else                  query_context_wrapped_init(&context, c, NULL, val);
+            r = toku_c_getf_get_both_range_reverse(c, remaining_flags, key, val, c_get_wrapper_callback, &context);
+            break;
         default:
             r = EINVAL;
             break;
@@ -1535,6 +1548,10 @@ static int locked_c_getf_set_range(DBC *c, u_int32_t flag, DBT * key, YDB_CALLBA
     toku_ydb_lock();  int r = toku_c_getf_set_range(c, flag, key, f, extra); toku_ydb_unlock(); return r;
 }
 
+static int locked_c_getf_set_range_reverse(DBC *c, u_int32_t flag, DBT * key, YDB_CALLBACK_FUNCTION f, void *extra) {
+    toku_ydb_lock();  int r = toku_c_getf_set_range_reverse(c, flag, key, f, extra); toku_ydb_unlock(); return r;
+}
+
 static int locked_c_getf_get_both(DBC *c, u_int32_t flag, DBT * key, DBT *val, YDB_CALLBACK_FUNCTION f, void *extra) {
     toku_ydb_lock();  int r = toku_c_getf_get_both(c, flag, key, val, f, extra); toku_ydb_unlock(); return r;
 }
@@ -1543,6 +1560,10 @@ static int locked_c_getf_get_both_range(DBC *c, u_int32_t flag, DBT * key, DBT *
     toku_ydb_lock();  int r = toku_c_getf_get_both_range(c, flag, key, val, f, extra); toku_ydb_unlock(); return r;
 }
 
+static int locked_c_getf_get_both_range_reverse(DBC *c, u_int32_t flag, DBT * key, DBT *val, YDB_CALLBACK_FUNCTION f, void *extra) {
+    toku_ydb_lock();  int r = toku_c_getf_get_both_range_reverse(c, flag, key, val, f, extra); toku_ydb_unlock(); return r;
+}
+
 typedef struct {
     BOOL            is_read_lock;
     DB_TXN         *txn;
@@ -2267,6 +2288,65 @@ c_getf_set_range_callback(ITEMLEN keylen, bytevec key, ITEMLEN vallen, bytevec v
     return r;
 }
 
+static int c_getf_set_range_reverse_callback(ITEMLEN keylen, bytevec key, ITEMLEN vallen, bytevec val, void *extra);
+
+static int
+toku_c_getf_set_range_reverse(DBC *c, u_int32_t flag, DBT *key, YDB_CALLBACK_FUNCTION f, void *extra) {
+    HANDLE_PANICKED_DB(c->dbp);
+    HANDLE_CURSOR_ILLEGAL_WORKING_PARENT_TXN(c);
+
+    QUERY_CONTEXT_WITH_INPUT_S context; //Describes the context of this query.
+    query_context_with_input_init(&context, c, flag, key, NULL, f, extra); 
+    TOKULOGGER logger = c_get_logger(c);
+    //toku_brt_cursor_set_range_reverse will call c_getf_set_range_reverse_callback(..., context) (if query is successful)
+    int r = toku_brt_cursor_set_range_reverse(dbc_struct_i(c)->c, key, c_getf_set_range_reverse_callback, &context, logger);
+    if (r == TOKUDB_USER_CALLBACK_ERROR) r = context.base.r_user_callback;
+    return r;
+}
+
+//result is the result of the query (i.e. 0 means found, DB_NOTFOUND, etc..)
+static int
+c_getf_set_range_reverse_callback(ITEMLEN keylen, bytevec key, ITEMLEN vallen, bytevec val, void *extra) {
+    QUERY_CONTEXT_WITH_INPUT super_context = extra;
+    QUERY_CONTEXT_BASE       context       = &super_context->base;
+
+    int r;
+
+    DBT found_key;
+    DBT found_val;
+    toku_fill_dbt(&found_key, key, keylen);
+    toku_fill_dbt(&found_val, val, vallen);
+
+    //Lock:
+    //  left(key) = found ? found_key : -infinity
+    //  left(val) = found ? found_val : -infinity
+    //  right(key,val)  = (input_key, infinity)
+    if (context->do_locking) {
+        RANGE_LOCK_REQUEST_S request;
+        if (key!=NULL) {
+            read_lock_request_init(&request, context->txn, context->db,
+                                   &found_key,               &found_val,
+                                   super_context->input_key, toku_lt_infinity);
+        }
+        else {
+            read_lock_request_init(&request, context->txn, context->db,
+                                   toku_lt_neg_infinity,     toku_lt_neg_infinity,
+                                   super_context->input_key, toku_lt_infinity);
+        }
+        r = grab_range_lock(&request);
+    }
+    else r = 0;
+
+    //Call application-layer callback if found and locks were successfully obtained.
+    if (r==0 && key!=NULL) {
+        context->r_user_callback = super_context->f(&found_key, &found_val, context->f_extra);
+        if (context->r_user_callback) r = TOKUDB_USER_CALLBACK_ERROR;
+    }
+
+    //Give brt-layer an error (if any) to return from toku_brt_cursor_set_range_reverse
+    return r;
+}
+
 static int c_getf_get_both_callback(ITEMLEN keylen, bytevec key, ITEMLEN vallen, bytevec val, void *extra);
 
 static int
@@ -2379,6 +2459,66 @@ c_getf_get_both_range_callback(ITEMLEN keylen, bytevec key, ITEMLEN vallen, byte
     return r;
 }
 
+static int c_getf_get_both_range_reverse_callback(ITEMLEN keylen, bytevec key, ITEMLEN vallen, bytevec val, void *extra);
+
+static int
+toku_c_getf_get_both_range_reverse(DBC *c, u_int32_t flag, DBT *key, DBT *val, YDB_CALLBACK_FUNCTION f, void *extra) {
+    HANDLE_PANICKED_DB(c->dbp);
+    HANDLE_CURSOR_ILLEGAL_WORKING_PARENT_TXN(c);
+    int r;
+    if (c_db_is_nodup(c)) r = toku_c_getf_get_both(c, flag, key, val, f, extra);
+    else {
+        QUERY_CONTEXT_WITH_INPUT_S context; //Describes the context of this query.
+        query_context_with_input_init(&context, c, flag, key, val, f, extra); 
+        TOKULOGGER logger = c_get_logger(c);
+        //toku_brt_cursor_get_both_range_reverse will call c_getf_get_both_range_reverse_callback(..., context) (if query is successful)
+        r = toku_brt_cursor_get_both_range_reverse(dbc_struct_i(c)->c, key, val, c_getf_get_both_range_reverse_callback, &context, logger);
+        if (r == TOKUDB_USER_CALLBACK_ERROR) r = context.base.r_user_callback;
+    }
+    return r;
+}
+
+//result is the result of the query (i.e. 0 means found, DB_NOTFOUND, etc..)
+static int
+c_getf_get_both_range_reverse_callback(ITEMLEN keylen, bytevec key, ITEMLEN vallen, bytevec val, void *extra) {
+    QUERY_CONTEXT_WITH_INPUT super_context = extra;
+    QUERY_CONTEXT_BASE       context       = &super_context->base;
+
+    int r;
+
+    DBT found_key;
+    DBT found_val;
+    toku_fill_dbt(&found_key, key, keylen);
+    toku_fill_dbt(&found_val, val, vallen);
+
+    //Lock:
+    //  left(key,val)  = (input_key, found ? found_val : -infinity)
+    //  right(key,val) = (input_key, input_val)
+    if (context->do_locking) {
+        RANGE_LOCK_REQUEST_S request;
+        if (key!=NULL) {
+            read_lock_request_init(&request, context->txn, context->db,
+                                   super_context->input_key, &found_val,
+                                   super_context->input_key, super_context->input_val);
+        }
+        else {
+            read_lock_request_init(&request, context->txn, context->db,
+                                   super_context->input_key, toku_lt_neg_infinity,
+                                   super_context->input_key, super_context->input_val);
+        }
+        r = grab_range_lock(&request);
+    }
+    else r = 0;
+
+    //Call application-layer callback if found and locks were successfully obtained.
+    if (r==0 && key!=NULL) {
+        context->r_user_callback = super_context->f(&found_key, &found_val, context->f_extra);
+        if (context->r_user_callback) r = TOKUDB_USER_CALLBACK_ERROR;
+    }
+
+    //Give brt-layer an error (if any) to return from toku_brt_cursor_get_both_range_reverse
+    return r;
+}
 
 static int locked_c_getf_heaviside(DBC *c, u_int32_t flags,
                                YDB_HEAVISIDE_CALLBACK_FUNCTION f, void *extra_f,
@@ -2737,8 +2877,10 @@ static int toku_db_cursor(DB * db, DB_TXN * txn, DBC ** c, u_int32_t flags, int
     SCRS(c_getf_heaviside);
     SCRS(c_getf_set);
     SCRS(c_getf_set_range);
+    SCRS(c_getf_set_range_reverse);
     SCRS(c_getf_get_both);
     SCRS(c_getf_get_both_range);
+    SCRS(c_getf_get_both_range_reverse);
 #undef SCRS
 
 #if !TOKUDB_NATIVE_H

toku_include/Makefile.include

@@ -98,7 +98,7 @@ WERROR    = -Werror
 endif
 WALL      = -Wall -Wextra -Wcast-align -Wbad-function-cast -Wno-missing-noreturn
 
-FORMAT    = -Wmissing-format-attribute
+FORMAT    = -Wmissing-format-attribute #-Wformat=2 #Stronger printf warnings once logger.c cleaned up
 ifeq ($(SYSTEM),sunos)
 VISIBILITY=
 else
@@ -338,6 +338,7 @@ endif
 tags: cscope.out TAGS
 
 TAGS: $(TOKUROOT)*/*.[ch] $(TOKUROOT)*/*/*.[ch] $(TOKUROOT)*/*/*/*.[ch]
+	rm -f `find $(TOKUROOT) -type f -name TAGS` #Delete all other tag files
 	etags $(TOKUROOT)*/*.[ch] $(TOKUROOT)*/*/*.[ch] $(TOKUROOT)*/*/*/*.[ch]
 
 cscope.files: $(TOKUROOT)*/*.[ch] $(TOKUROOT)*/*/*.[ch] $(TOKUROOT)*/*/*/*.[ch]