Closes #1055

Implement DBC->c_getf_prev

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

src/ydb.c

@@ -91,6 +91,9 @@ static int toku_db_cursor(DB *db, DB_TXN * txn, DBC **c, u_int32_t flags, int is
 
 /* lightweight cursor methods. */
 static int toku_c_getf_next(DBC *c, u_int32_t flag, void(*f)(DBT const *key, DBT const *data, void *extra), void *extra);
+
+static int toku_c_getf_prev(DBC *c, u_int32_t flag, void(*f)(DBT const *key, DBT const *data, void *extra), void *extra);
+
 static int toku_c_getf_next_dup(DBC *c, u_int32_t flag, void(*f)(DBT const *key, DBT const *data, void *extra), void *extra);
 
 static int toku_c_getf_heavi(DBC *c, u_int32_t flags,
@@ -1766,6 +1769,10 @@ static int locked_c_getf_next(DBC *c, u_int32_t flag, void(*f)(DBT const *key, D
     toku_ydb_lock();  int r = toku_c_getf_next(c, flag, f, extra); toku_ydb_unlock(); return r;
 }
 
+static int locked_c_getf_prev(DBC *c, u_int32_t flag, void(*f)(DBT const *key, DBT const *data, void *extra), void *extra) {
+    toku_ydb_lock();  int r = toku_c_getf_prev(c, flag, f, extra); toku_ydb_unlock(); return r;
+}
+
 static int locked_c_getf_next_dup(DBC *c, u_int32_t flag, void(*f)(DBT const *key, DBT const *data, void *extra), void *extra) {
     toku_ydb_lock();  int r = toku_c_getf_next_dup(c, flag, f, extra); toku_ydb_unlock(); return r;
 }
@@ -1828,6 +1835,64 @@ static int toku_c_getf_next(DBC *c, u_int32_t flag, void(*f)(DBT const *key, DBT
     return r;
 }
 
+static int toku_c_getf_prev_old(DBC *c, u_int32_t flag, void(*f)(DBT const *key, DBT const *data, void *extra), void *extra) {
+    DBT key,val;
+    memset(&key, 0, sizeof(key));
+    memset(&val, 0, sizeof(val));
+    
+    u_int32_t lock_flags = get_prelocked_flags(flag);
+    flag &= ~lock_flags;
+    assert(flag==0);
+    int r = toku_c_get_noassociate(c, &key, &val, DB_PREV | flag | lock_flags);
+    if (r==0) f(&key, &val, extra);
+    return r;
+}
+
+static int toku_c_getf_prev(DBC *c, u_int32_t flag, void(*f)(DBT const *key, DBT const *data, void *extra), void *extra) {
+    HANDLE_PANICKED_DB(c->dbp);
+    if (toku_c_uninitialized(c)) return toku_c_getf_prev_old(c, flag, f, extra); //return toku_c_getf_last(c, flag, f, extra);
+    u_int32_t lock_flags = get_prelocked_flags(flag);
+    flag &= ~lock_flags;
+    assert(flag==0);
+    TOKUTXN txn = c->i->txn ? c->i->txn->i->tokutxn : NULL;
+    const DBT *pkey, *pval;
+    pkey = pval = toku_lt_neg_infinity;
+    int r;
+
+    DB *db=c->dbp;
+    toku_lock_tree* lt = db->i->lt;
+    BOOL do_locking = lt!=NULL && !lock_flags;
+
+    unsigned int brtflags;
+    toku_brt_get_flags(db->i->brt, &brtflags);
+
+    int c_get_result = toku_brt_cursor_get(c->i->c, NULL, NULL,
+                                          (brtflags & TOKU_DB_DUPSORT) ? DB_PREV : DB_PREV_NODUP,
+                                           txn);
+    if (c_get_result!=0 && c_get_result!=DB_NOTFOUND) { r = c_get_result; goto cleanup; }
+    int found = c_get_result==0;
+    if (found) brt_cursor_peek_current(c->i->c, &pkey, &pval);
+    if (do_locking) {
+
+	DBT *prevkey = found ? brt_cursor_peek_prev_key(c->i->c) : brt_cursor_peek_current_key(c->i->c);
+	DBT *prevval = found ? brt_cursor_peek_prev_val(c->i->c) : brt_cursor_peek_current_key(c->i->c);
+
+        DB_TXN *txn_anc = toku_txn_ancestor(c->i->txn);
+        r = toku_txn_add_lt(txn_anc, lt);
+        if (r!=0) goto cleanup;
+        r = toku_lt_acquire_range_read_lock(lt, db, toku_txn_get_txnid(txn_anc->i->tokutxn),
+                                            pkey, pval,
+                                            prevkey, prevval);
+        if (r!=0) goto cleanup;
+    }
+    if (found) {
+	f(pkey, pval, extra);
+    }
+    r = c_get_result;
+ cleanup:
+    return r;
+}
+
 static int toku_c_getf_next_dup(DBC *c, u_int32_t flag, void(*f)(DBT const *key, DBT const *data, void *extra), void *extra) {
     HANDLE_PANICKED_DB(c->dbp);
     if (toku_c_uninitialized(c)) return EINVAL;
@@ -2303,6 +2368,7 @@ static int toku_db_cursor(DB * db, DB_TXN * txn, DBC ** c, u_int32_t flags, int
     SCRS(c_del);
     SCRS(c_count);
     SCRS(c_getf_next);
+    SCRS(c_getf_prev);
     SCRS(c_getf_next_dup);
     SCRS(c_getf_heavi);
 #undef SCRS