#3129 fix recover of insert multiple and delete multiple with a valid source filenum closes[t:3129]

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

newbrt/recover.c

@@ -839,38 +839,42 @@ static int toku_recover_enq_insert_multiple (struct logtype_enq_insert_multiple
             if (r == 0)
                 src_db = tuple->brt->db;
             else
-                do_inserts = FALSE; // src file was probably deleted
+                do_inserts = FALSE; // src file was probably deleted, #3129
         }
     }
-    DBT src_key, src_val, dest_key, dest_val;
-    toku_fill_dbt(&src_key, l->src_key.data, l->src_key.len);
-    toku_fill_dbt(&src_val, l->src_val.data, l->src_val.len);
-    toku_init_dbt_flags(&dest_key, DB_DBT_REALLOC);
-    toku_init_dbt_flags(&dest_val, DB_DBT_REALLOC);
-
-    for (uint32_t file = 0; do_inserts && file < l->dest_filenums.num; file++) {
-        struct file_map_tuple *tuple = NULL;
-        r = file_map_find(&renv->fmap, l->dest_filenums.filenums[file], &tuple);
-        if (r==0) {
-            // We found the cachefile.  (maybe) Do the insert.
-            DB *db = tuple->brt->db;
-            r = renv->generate_row_for_put(db, src_db, &dest_key, &dest_val, &src_key, &src_val);
-            assert(r==0);
-            r = toku_brt_maybe_insert(tuple->brt, &dest_key, &dest_val, txn, TRUE, l->lsn, FALSE, BRT_INSERT);
-            assert(r == 0);
+    
+    if (do_inserts) {
+        DBT src_key, src_val, dest_key, dest_val;
+        toku_fill_dbt(&src_key, l->src_key.data, l->src_key.len);
+        toku_fill_dbt(&src_val, l->src_val.data, l->src_val.len);
+        toku_init_dbt_flags(&dest_key, DB_DBT_REALLOC);
+        toku_init_dbt_flags(&dest_val, DB_DBT_REALLOC);
+
+        for (uint32_t file = 0; file < l->dest_filenums.num; file++) {
+            struct file_map_tuple *tuple = NULL;
+            r = file_map_find(&renv->fmap, l->dest_filenums.filenums[file], &tuple);
+            if (r==0) {
+                // We found the cachefile.  (maybe) Do the insert.
+                DB *db = tuple->brt->db;
+                r = renv->generate_row_for_put(db, src_db, &dest_key, &dest_val, &src_key, &src_val);
+                assert(r==0);
+                r = toku_brt_maybe_insert(tuple->brt, &dest_key, &dest_val, txn, TRUE, l->lsn, FALSE, BRT_INSERT);
+                assert(r == 0);
 
-            //flags==0 means generate_row_for_put callback changed it
-            //(and freed any memory necessary to do so) so that values are now stored
-            //in temporary memory that does not need to be freed.  We need to continue
-            //using DB_DBT_REALLOC however.
-            if (dest_key.flags == 0) 
-                toku_init_dbt_flags(&dest_key, DB_DBT_REALLOC);
-            if (dest_val.flags == 0)
-                toku_init_dbt_flags(&dest_val, DB_DBT_REALLOC);
+                //flags==0 means generate_row_for_put callback changed it
+                //(and freed any memory necessary to do so) so that values are now stored
+                //in temporary memory that does not need to be freed.  We need to continue
+                //using DB_DBT_REALLOC however.
+                if (dest_key.flags == 0) 
+                    toku_init_dbt_flags(&dest_key, DB_DBT_REALLOC);
+                if (dest_val.flags == 0)
+                    toku_init_dbt_flags(&dest_val, DB_DBT_REALLOC);
+            }
         }
+
+        if (dest_key.data) toku_free(dest_key.data); //TODO: #2321 May need windows hack
+        if (dest_val.data) toku_free(dest_val.data); //TODO: #2321 May need windows hack
     }
-    if (dest_key.data) toku_free(dest_key.data); //TODO: #2321 May need windows hack
-    if (dest_val.data) toku_free(dest_val.data); //TODO: #2321 May need windows hack
 
     return 0;
 }
@@ -897,33 +901,36 @@ static int toku_recover_enq_delete_multiple (struct logtype_enq_delete_multiple
             if (r == 0)
                 src_db = tuple->brt->db;
             else
-                do_deletes = FALSE; // src file was probably deleted
+                do_deletes = FALSE; // src file was probably deleted, #3129
         }
     }
 
-    DBT src_key, src_val, dest_key;
-    toku_fill_dbt(&src_key, l->src_key.data, l->src_key.len);
-    toku_fill_dbt(&src_val, l->src_val.data, l->src_val.len);
-    toku_init_dbt_flags(&dest_key, DB_DBT_REALLOC);
-
-    for (uint32_t file = 0; do_deletes && file < l->dest_filenums.num; file++) {
-        struct file_map_tuple *tuple = NULL;
-        r = file_map_find(&renv->fmap, l->dest_filenums.filenums[file], &tuple);
-        if (r==0) {
-            // We found the cachefile.  (maybe) Do the delete.
-            DB *db = tuple->brt->db;
-            r = renv->generate_row_for_del(db, src_db, &dest_key, &src_key, &src_val);
-            assert(r==0);
-            r = toku_brt_maybe_delete(tuple->brt, &dest_key, txn, TRUE, l->lsn, FALSE);
-            assert(r == 0);
+    if (do_deletes) {
+        DBT src_key, src_val, dest_key;
+        toku_fill_dbt(&src_key, l->src_key.data, l->src_key.len);
+        toku_fill_dbt(&src_val, l->src_val.data, l->src_val.len);
+        toku_init_dbt_flags(&dest_key, DB_DBT_REALLOC);
+
+        for (uint32_t file = 0; file < l->dest_filenums.num; file++) {
+            struct file_map_tuple *tuple = NULL;
+            r = file_map_find(&renv->fmap, l->dest_filenums.filenums[file], &tuple);
+            if (r==0) {
+                // We found the cachefile.  (maybe) Do the delete.
+                DB *db = tuple->brt->db;
+                r = renv->generate_row_for_del(db, src_db, &dest_key, &src_key, &src_val);
+                assert(r==0);
+                r = toku_brt_maybe_delete(tuple->brt, &dest_key, txn, TRUE, l->lsn, FALSE);
+                assert(r == 0);
 
-            //flags==0 indicates the return values are stored in temporary memory that does
-            //not need to be freed.  We need to continue using DB_DBT_REALLOC however.
-            if (dest_key.flags == 0)
-                toku_init_dbt_flags(&dest_key, DB_DBT_REALLOC);
+                //flags==0 indicates the return values are stored in temporary memory that does
+                //not need to be freed.  We need to continue using DB_DBT_REALLOC however.
+                if (dest_key.flags == 0)
+                    toku_init_dbt_flags(&dest_key, DB_DBT_REALLOC);
+            }
         }
+        
+        if (dest_key.flags & DB_DBT_REALLOC && dest_key.data) toku_free(dest_key.data); //TODO: #2321 May need windows hack
     }
-    if (dest_key.flags & DB_DBT_REALLOC && dest_key.data) toku_free(dest_key.data); //TODO: #2321 May need windows hack
 
     return 0;
 }