Fix the DBT problem for duplicate keys. Fixes #101.

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

newbrt/brt.c

@@ -2449,7 +2449,7 @@ static int brtcurs_set_position_last (BRT_CURSOR cursor, DISKOFF off, DBT *key,
         cursor->path_len--;
         goto died0;
     } else {
-	r=toku_pma_cursor(node->u.l.buffer, &cursor->pmacurs);
+	r=toku_pma_cursor(node->u.l.buffer, &cursor->pmacurs, &cursor->brt->skey, &cursor->brt->sval);
 	if (r!=0) {
 	    if (0) { died10: toku_pma_cursor_free(&cursor->pmacurs); }
 	    cursor->path_len--;
@@ -2512,7 +2512,7 @@ static int brtcurs_set_position_first (BRT_CURSOR cursor, DISKOFF off, DBT *key,
         cursor->path_len--;
         goto died0;
     } else {
-	r=toku_pma_cursor(node->u.l.buffer, &cursor->pmacurs);
+	r=toku_pma_cursor(node->u.l.buffer, &cursor->pmacurs, &cursor->brt->skey, &cursor->brt->sval);
 	if (r!=0) {
 	    if (0) { died10: toku_pma_cursor_free(&cursor->pmacurs); }
 	    cursor->path_len--;
@@ -2681,7 +2681,7 @@ static int brtcurs_set_key(BRT_CURSOR cursor, DISKOFF off, DBT *key, DBT *val, i
     } else {
         cursor->path_len += 1;
         cursor->path[cursor->path_len-1] = node;
-        r = toku_pma_cursor(node->u.l.buffer, &cursor->pmacurs);
+        r = toku_pma_cursor(node->u.l.buffer, &cursor->pmacurs, &cursor->brt->skey, &cursor->brt->sval);
         if (r == 0) {
             if (flag == DB_SET)
                 r = toku_pma_cursor_set_key(cursor->pmacurs, key);
@@ -2752,7 +2752,7 @@ static int brtcurs_set_range(BRT_CURSOR cursor, DISKOFF off, DBT *key, TOKUTXN t
     } else {
         cursor->path_len += 1;
         cursor->path[cursor->path_len-1] = node;
-        r = toku_pma_cursor(node->u.l.buffer, &cursor->pmacurs);
+        r = toku_pma_cursor(node->u.l.buffer, &cursor->pmacurs, &cursor->brt->skey, &cursor->brt->sval);
         if (r == 0) {
             r = toku_pma_cursor_set_range(cursor->pmacurs, key);
             if (r != 0) {

newbrt/pma-internal.h

@@ -7,7 +7,7 @@ struct pma_cursor {
     PMA pma;
     int position; /* -1 if the position is undefined. */
     struct list next;
-    void *skey, *sval; /* used in dbts. */ 
+    void **sskey, *ssval; /* Used in dbts.   When a cursor is created, you must provide a void** to return results in. */
 };
 
 struct pma {

newbrt/pma.c

@@ -665,14 +665,14 @@ int toku_pma_set_dup_compare(PMA pma, pma_compare_fun_t dup_compare_fun) {
     return 0;
 }
 
-int toku_pma_cursor (PMA pma, PMA_CURSOR *cursp) {
+int toku_pma_cursor (PMA pma, PMA_CURSOR *cursp, void **sskey, void **ssval) {
     PMA_CURSOR MALLOC(curs);
     assert(curs!=0);
     if (errno!=0) return errno;
     curs->position=-1; /* undefined */
     curs->pma = pma;
-    curs->skey = 0;
-    curs->sval=0;
+    curs->sskey = sskey;
+    curs->ssval = ssval;
     list_push(&pma->cursors, &curs->next);
     *cursp=curs;
     return 0;
@@ -758,7 +758,7 @@ int toku_pma_cursor_get_current_data(PMA_CURSOR c, DBT *data) {
     struct kv_pair *pair = pma->pairs[c->position];
     if (!kv_pair_valid(pair)) 
         return BRT_KEYEMPTY;
-    toku_dbt_set_value(data, kv_pair_val(pair), kv_pair_vallen(pair), &c->sval);
+    toku_dbt_set_value(data, kv_pair_val(pair), kv_pair_vallen(pair), c->ssval);
     return 0;
 }
 
@@ -769,8 +769,8 @@ int toku_pma_cursor_get_current(PMA_CURSOR c, DBT *key, DBT *val) {
     struct kv_pair *pair = pma->pairs[c->position];
     if (!kv_pair_valid(pair)) 
         return BRT_KEYEMPTY;
-    toku_dbt_set_value(key, kv_pair_key(pair), kv_pair_keylen(pair), &c->skey);
-    toku_dbt_set_value(val, kv_pair_val(pair), kv_pair_vallen(pair), &c->sval);
+    toku_dbt_set_value(key, kv_pair_key(pair), kv_pair_keylen(pair), c->sskey);
+    toku_dbt_set_value(val, kv_pair_val(pair), kv_pair_vallen(pair), c->ssval);
     return 0;
 }
 
@@ -863,8 +863,7 @@ int toku_pma_cursor_free (PMA_CURSOR *cursp) {
         __pma_count_cursor_refs(pma, curs->position) == 0) {
         __pma_delete_finish(pma, curs->position);
     }
-    if (curs->skey) toku_free(curs->skey);
-    if (curs->sval) toku_free(curs->sval);
+    // It's not our job to free the sskey and ssval blocks.
     toku_free(curs);
     *cursp=0;
     return 0;

newbrt/pma.h

@@ -91,7 +91,7 @@ int toku_pma_split(PMA origpma,  unsigned int *origpma_size, DBT *splitk,
 int toku_pma_bulk_insert(PMA pma, DBT *keys, DBT *vals, int n_newpairs, u_int32_t rand4sem, u_int32_t *fingerprint);
 
 /* Move the cursor to the beginning or the end or to a key */
-int toku_pma_cursor (PMA, PMA_CURSOR *);
+int toku_pma_cursor (PMA, PMA_CURSOR *, void** /*sskey*/, void ** /*ssval*/); // the sskey and ssval point to variables that hold blocks that can be used to return values for zero'd DBTS.
 int toku_pma_cursor_free (PMA_CURSOR*);
 
 /* get the pma that a pma cursor is bound to */

src/tests/test_env_close_flags.c

@@ -24,8 +24,8 @@ int main (int argc __attribute__((__unused__)), char *argv[]  __attribute__((__u
     r=mkdir(DIR, 0777);        assert(r==0);
     r=db_env_create(&env, 0);  assert(r==0);
     r=env->close   (env, 1);  
-    //BDB does not check this.
-#ifdef USE_TDB    
+    //BDB does not check this in some versions
+#if defined(USE_TDB) || 
     assert(r==EINVAL);
 #else
     assert(r==0);

src/ydb.c

@@ -811,7 +811,7 @@ static int toku_db_put(DB * db, DB_TXN * txn, DBT * key, DBT * data, u_int32_t f
     if (r!=0) return r;
 
     // For each secondary add the relevant records.
-    if (db->i->associate_callback) {
+    if (db->i->primary==0) { // Only do it if it is a primary.   This loop would run an unknown number of times if we tried it on a secondary.
 	struct list *h;
 	for (h=list_head(&db->i->associated); h!=&db->i->associated; h=h->next) {
 	    struct __toku_db_internal *dbi=list_struct(h, struct __toku_db_internal, associated);