closes #5773, merge to main from tokudb.5773, Merging r51027 through r51078 into '.'

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

ft/ft-internal.h

@@ -916,7 +916,7 @@ int toku_cmd_leafval_heaviside (OMTVALUE leafentry, void *extra)
     __attribute__((__warn_unused_result__));
 
 // toku_ft_root_put_cmd() accepts non-constant cmd because this is where we set the msn
-void toku_ft_root_put_cmd(FT h, FT_MSG_S * cmd, TXNID oldest_referenced_xid);
+void toku_ft_root_put_cmd(FT h, FT_MSG_S * cmd, TXNID oldest_referenced_xid, GC_INFO gc_info);
 
 void *mempool_malloc_from_omt(OMT *omtp, struct mempool *mp, size_t size, void **maybe_free);
 // Effect: Allocate a new object of size SIZE in MP.  If MP runs out of space, allocate new a new mempool space, and copy all the items
@@ -1077,6 +1077,7 @@ toku_ft_bn_apply_cmd_once (
     uint32_t idx,
     LEAFENTRY le,
     TXNID oldest_referenced_xid,
+    GC_INFO gc_info,
     uint64_t *workdonep,
     STAT64INFO stats_to_update
     );
@@ -1089,6 +1090,7 @@ toku_ft_bn_apply_cmd (
     BASEMENTNODE bn,
     FT_MSG cmd,
     TXNID oldest_referenced_xid,
+    GC_INFO gc_info,
     uint64_t *workdone,
     STAT64INFO stats_to_update
     );
@@ -1101,6 +1103,7 @@ toku_ft_leaf_apply_cmd (
     FTNODE node,
     int target_childnum,
     FT_MSG cmd,
+    GC_INFO gc_info,
     uint64_t *workdone,
     STAT64INFO stats_to_update
     );
@@ -1114,6 +1117,7 @@ toku_ft_node_put_cmd (
     int target_childnum,
     FT_MSG cmd,
     bool is_fresh,
+    GC_INFO gc_info,
     size_t flow_deltas[],
     STAT64INFO stats_to_update
     );

ft/ft-ops.cc

@@ -1539,6 +1539,7 @@ toku_ft_bn_apply_cmd_once (
     uint32_t idx,
     LEAFENTRY le,
     TXNID oldest_referenced_xid,
+    GC_INFO gc_info,
     uint64_t *workdone,
     STAT64INFO stats_to_update
     )
@@ -1559,7 +1560,7 @@ toku_ft_bn_apply_cmd_once (
     // toku_le_apply_msg() may call mempool_malloc_from_omt() to allocate more space.
     // That means le is guaranteed to not cause a sigsegv but it may point to a mempool that is
     // no longer in use.  We'll have to release the old mempool later.
-    toku_le_apply_msg(cmd, le, oldest_referenced_xid, &newsize, &new_le, &bn->buffer, &bn->buffer_mempool, &maybe_free, &numbytes_delta);
+    toku_le_apply_msg(cmd, le, oldest_referenced_xid, gc_info, &newsize, &new_le, &bn->buffer, &bn->buffer_mempool, &maybe_free, &numbytes_delta);
 
     if (new_le) {
         paranoid_invariant(newsize == leafentry_disksize(new_le));
@@ -1629,6 +1630,7 @@ struct setval_extra_s {
     uint32_t idx;
     LEAFENTRY le;
     TXNID oldest_referenced_xid;
+    GC_INFO gc_info;
     uint64_t * workdone;  // set by toku_ft_bn_apply_cmd_once()
     STAT64INFO stats_to_update;
 };
@@ -1661,7 +1663,7 @@ static void setval_fun (const DBT *new_val, void *svextra_v) {
         }
         toku_ft_bn_apply_cmd_once(svextra->bn, &msg,
                                   svextra->idx, svextra->le,
-                                  svextra->oldest_referenced_xid,
+                                  svextra->oldest_referenced_xid, svextra->gc_info,
                                   svextra->workdone, svextra->stats_to_update);
         svextra->setval_r = 0;
     }
@@ -1674,6 +1676,7 @@ static void setval_fun (const DBT *new_val, void *svextra_v) {
 static int do_update(ft_update_func update_fun, DESCRIPTOR desc, BASEMENTNODE bn, FT_MSG cmd, uint32_t idx,
                      LEAFENTRY le,
                      TXNID oldest_referenced_xid,
+                     GC_INFO gc_info,
                      uint64_t * workdone,
                      STAT64INFO stats_to_update) {
     LEAFENTRY le_for_update;
@@ -1716,7 +1719,7 @@ static int do_update(ft_update_func update_fun, DESCRIPTOR desc, BASEMENTNODE bn
     le_for_update = le;
 
     struct setval_extra_s setval_extra = {setval_tag, false, 0, bn, cmd->msn, cmd->xids,
-                                          keyp, idx, le_for_update, oldest_referenced_xid, workdone, stats_to_update};
+                                          keyp, idx, le_for_update, oldest_referenced_xid, gc_info, workdone, stats_to_update};
     // call handlerton's brt->update_fun(), which passes setval_extra to setval_fun()
     FAKE_DB(db, desc);
     int r = update_fun(
@@ -1740,6 +1743,7 @@ toku_ft_bn_apply_cmd (
     BASEMENTNODE bn,
     FT_MSG cmd,
     TXNID oldest_known_referenced_xid,
+    GC_INFO gc_info, 
     uint64_t *workdone,
     STAT64INFO stats_to_update
     )
@@ -1781,7 +1785,7 @@ toku_ft_bn_apply_cmd (
             assert_zero(r);
             CAST_FROM_VOIDP(storeddata, storeddatav);
         }
-        toku_ft_bn_apply_cmd_once(bn, cmd, idx, storeddata, oldest_known_referenced_xid, workdone, stats_to_update);
+        toku_ft_bn_apply_cmd_once(bn, cmd, idx, storeddata, oldest_known_referenced_xid, gc_info, workdone, stats_to_update);
 
         // if the insertion point is within a window of the right edge of
         // the leaf then it is sequential
@@ -1809,7 +1813,7 @@ toku_ft_bn_apply_cmd (
         if (r == DB_NOTFOUND) break;
         assert_zero(r);
         CAST_FROM_VOIDP(storeddata, storeddatav);
-        toku_ft_bn_apply_cmd_once(bn, cmd, idx, storeddata, oldest_known_referenced_xid, workdone, stats_to_update);
+        toku_ft_bn_apply_cmd_once(bn, cmd, idx, storeddata, oldest_known_referenced_xid, gc_info, workdone, stats_to_update);
 
         break;
     }
@@ -1825,7 +1829,7 @@ toku_ft_bn_apply_cmd (
             CAST_FROM_VOIDP(storeddata, storeddatav);
             int deleted = 0;
             if (!le_is_clean(storeddata)) { //If already clean, nothing to do.
-                toku_ft_bn_apply_cmd_once(bn, cmd, idx, storeddata, oldest_known_referenced_xid, workdone, stats_to_update);
+                toku_ft_bn_apply_cmd_once(bn, cmd, idx, storeddata, oldest_known_referenced_xid, gc_info, workdone, stats_to_update);
                 uint32_t new_omt_size = toku_omt_size(bn->buffer);
                 if (new_omt_size != omt_size) {
                     paranoid_invariant(new_omt_size+1 == omt_size);
@@ -1851,7 +1855,7 @@ toku_ft_bn_apply_cmd (
             CAST_FROM_VOIDP(storeddata, storeddatav);
             int deleted = 0;
             if (le_has_xids(storeddata, cmd->xids)) {
-                toku_ft_bn_apply_cmd_once(bn, cmd, idx, storeddata, oldest_known_referenced_xid, workdone, stats_to_update);
+                toku_ft_bn_apply_cmd_once(bn, cmd, idx, storeddata, oldest_known_referenced_xid, gc_info, workdone, stats_to_update);
                 uint32_t new_omt_size = toku_omt_size(bn->buffer);
                 if (new_omt_size != omt_size) {
                     paranoid_invariant(new_omt_size+1 == omt_size);
@@ -1872,10 +1876,10 @@ toku_ft_bn_apply_cmd (
         r = toku_omt_find_zero(bn->buffer, toku_cmd_leafval_heaviside, &be,
                                &storeddatav, &idx);
         if (r==DB_NOTFOUND) {
-            r = do_update(update_fun, desc, bn, cmd, idx, NULL, oldest_known_referenced_xid, workdone, stats_to_update);
+            r = do_update(update_fun, desc, bn, cmd, idx, NULL, oldest_known_referenced_xid, gc_info, workdone, stats_to_update);
         } else if (r==0) {
             CAST_FROM_VOIDP(storeddata, storeddatav);
-            r = do_update(update_fun, desc, bn, cmd, idx, storeddata, oldest_known_referenced_xid, workdone, stats_to_update);
+            r = do_update(update_fun, desc, bn, cmd, idx, storeddata, oldest_known_referenced_xid, gc_info, workdone, stats_to_update);
         } // otherwise, a worse error, just return it
         break;
     }
@@ -1887,7 +1891,7 @@ toku_ft_bn_apply_cmd (
             r = toku_omt_fetch(bn->buffer, idx, &storeddatav);
             assert_zero(r);
             CAST_FROM_VOIDP(storeddata, storeddatav);
-            r = do_update(update_fun, desc, bn, cmd, idx, storeddata, oldest_known_referenced_xid, workdone, stats_to_update);
+            r = do_update(update_fun, desc, bn, cmd, idx, storeddata, oldest_known_referenced_xid, gc_info, workdone, stats_to_update);
             assert_zero(r);
 
             if (num_leafentries_before == toku_omt_size(bn->buffer)) {
@@ -2344,6 +2348,7 @@ void toku_bnc_flush_to_child(
                 -1,
                 &ftcmd,
                 is_fresh,
+                make_gc_info(true), // mvcc_needed
                 flow_deltas,
                 &stats_delta
                 );
@@ -2379,6 +2384,7 @@ toku_ft_node_put_cmd (
     int target_childnum,
     FT_MSG cmd,
     bool is_fresh,
+    GC_INFO gc_info,
     size_t flow_deltas[],
     STAT64INFO stats_to_update
     )
@@ -2396,7 +2402,7 @@ toku_ft_node_put_cmd (
     // and instead defer to these functions
     //
     if (node->height==0) {
-        toku_ft_leaf_apply_cmd(compare_fun, update_fun, desc, node, target_childnum, cmd, nullptr, stats_to_update);
+        toku_ft_leaf_apply_cmd(compare_fun, update_fun, desc, node, target_childnum, cmd, gc_info, nullptr, stats_to_update);
     } else {
         ft_nonleaf_put_cmd(compare_fun, desc, node, target_childnum, cmd, is_fresh, flow_deltas);
     }
@@ -2416,6 +2422,7 @@ void toku_ft_leaf_apply_cmd(
     FTNODE node,
     int target_childnum,  // which child to inject to, or -1 if unknown
     FT_MSG cmd,
+    GC_INFO gc_info,
     uint64_t *workdone,
     STAT64INFO stats_to_update
     )
@@ -2465,6 +2472,7 @@ void toku_ft_leaf_apply_cmd(
                                  bn,
                                  cmd,
                                  oldest_known_referenced_xid,
+                                 gc_info,
                                  workdone,
                                  stats_to_update);
         } else {
@@ -2481,6 +2489,7 @@ void toku_ft_leaf_apply_cmd(
                                       BLB(node, childnum),
                                       cmd,
                                       oldest_known_referenced_xid,
+                                      gc_info,
                                       workdone,
                                       stats_to_update);
             } else {
@@ -2500,7 +2509,8 @@ static void inject_message_in_locked_node(
     int childnum, 
     FT_MSG_S *cmd, 
     size_t flow_deltas[],
-    TXNID oldest_referenced_xid
+    TXNID oldest_referenced_xid,
+    GC_INFO gc_info
     ) 
 {
     // No guarantee that we're the writer, but oh well.
@@ -2531,6 +2541,7 @@ static void inject_message_in_locked_node(
         childnum,
         cmd,
         true,
+        gc_info,
         flow_deltas,
         &stats_delta
         );
@@ -2690,7 +2701,7 @@ static bool process_maybe_reactive_child(FT ft, FTNODE parent, FTNODE child, int
     abort();
 }
 
-static void inject_message_at_this_blocknum(FT ft, CACHEKEY cachekey, uint32_t fullhash, FT_MSG_S *cmd, size_t flow_deltas[], TXNID oldest_referenced_xid)
+static void inject_message_at_this_blocknum(FT ft, CACHEKEY cachekey, uint32_t fullhash, FT_MSG_S *cmd, size_t flow_deltas[], TXNID oldest_referenced_xid, GC_INFO gc_info)
 // Effect:
 //  Inject cmd into the node at this blocknum (cachekey).
 //  Gets a write lock on the node for you.
@@ -2702,7 +2713,7 @@ static void inject_message_at_this_blocknum(FT ft, CACHEKEY cachekey, uint32_t f
     toku_assert_entire_node_in_memory(node);
     paranoid_invariant(node->fullhash==fullhash);
     ft_verify_flags(ft, node);
-    inject_message_in_locked_node(ft, node, -1, cmd, flow_deltas, oldest_referenced_xid);
+    inject_message_in_locked_node(ft, node, -1, cmd, flow_deltas, oldest_referenced_xid, gc_info);
 }
 
 __attribute__((const))
@@ -2722,6 +2733,7 @@ static void push_something_in_subtree(
     FT_MSG_S *cmd, 
     size_t flow_deltas[], 
     TXNID oldest_referenced_xid, 
+    GC_INFO gc_info,
     int depth, 
     seqinsert_loc loc, 
     bool just_did_split_or_merge
@@ -2762,7 +2774,7 @@ static void push_something_in_subtree(
         default:
             STATUS_INC(FT_PRO_NUM_INJECT_DEPTH_GT3, 1); break;
         }
-        inject_message_in_locked_node(ft, subtree_root, target_childnum, cmd, flow_deltas, oldest_referenced_xid);
+        inject_message_in_locked_node(ft, subtree_root, target_childnum, cmd, flow_deltas, oldest_referenced_xid, gc_info);
     } else {
         int r;
         int childnum;
@@ -2851,13 +2863,13 @@ static void push_something_in_subtree(
                     struct ftnode_fetch_extra bfe;
                     fill_bfe_for_full_read(&bfe, ft); // should be fully in memory, we just split it
                     toku_pin_ftnode_off_client_thread_batched(ft, subtree_root_blocknum, subtree_root_fullhash, &bfe, PL_READ, 0, nullptr, &newparent);
-                    push_something_in_subtree(ft, newparent, -1, cmd, flow_deltas, oldest_referenced_xid, depth, loc, true);
+                    push_something_in_subtree(ft, newparent, -1, cmd, flow_deltas, oldest_referenced_xid, gc_info, depth, loc, true);
                     return;
                 }
             }
 
             if (next_loc != NEITHER_EXTREME || child->dirty || toku_bnc_should_promote(ft, bnc)) {
-                push_something_in_subtree(ft, child, -1, cmd, flow_deltas, oldest_referenced_xid, depth + 1, next_loc, false);
+                push_something_in_subtree(ft, child, -1, cmd, flow_deltas, oldest_referenced_xid, gc_info, depth + 1, next_loc, false);
                 toku_sync_fetch_and_add(&bnc->flow[0], flow_deltas[0]);
                 // The recursive call unpinned the child, but
                 // we're responsible for unpinning subtree_root.
@@ -2893,12 +2905,17 @@ static void push_something_in_subtree(
             default:
                 STATUS_INC(FT_PRO_NUM_INJECT_DEPTH_GT3, 1); break;
             }
-            inject_message_at_this_blocknum(ft, subtree_root_blocknum, subtree_root_fullhash, cmd, flow_deltas, oldest_referenced_xid);
+            inject_message_at_this_blocknum(ft, subtree_root_blocknum, subtree_root_fullhash, cmd, flow_deltas, oldest_referenced_xid, gc_info);
         }
     }
 }
 
-void toku_ft_root_put_cmd(FT ft, FT_MSG_S *cmd, TXNID oldest_referenced_xid)
+void toku_ft_root_put_cmd(
+    FT ft, 
+    FT_MSG_S *cmd, 
+    TXNID oldest_referenced_xid,
+    GC_INFO gc_info
+    )
 // Effect:
 //  - assign msn to cmd and update msn in the header
 //  - push the cmd into the ft
@@ -2998,22 +3015,22 @@ void toku_ft_root_put_cmd(FT ft, FT_MSG_S *cmd, TXNID oldest_referenced_xid)
         // If the root's a leaf or we're injecting a broadcast, drop the read lock and inject here.
         toku_unpin_ftnode_read_only(ft, node);
         STATUS_INC(FT_PRO_NUM_ROOT_H0_INJECT, 1);
-        inject_message_at_this_blocknum(ft, root_key, fullhash, cmd, flow_deltas, oldest_referenced_xid);
+        inject_message_at_this_blocknum(ft, root_key, fullhash, cmd, flow_deltas, oldest_referenced_xid, gc_info);
     } else if (node->height > 1) {
         // If the root's above height 1, we are definitely eligible for promotion.
-        push_something_in_subtree(ft, node, -1, cmd, flow_deltas, oldest_referenced_xid, 0, LEFT_EXTREME | RIGHT_EXTREME, false);
+        push_something_in_subtree(ft, node, -1, cmd, flow_deltas, oldest_referenced_xid, gc_info, 0, LEFT_EXTREME | RIGHT_EXTREME, false);
     } else {
         // The root's height 1.  We may be eligible for promotion here.
         // On the extremes, we want to promote, in the middle, we don't.
         int childnum = toku_ftnode_which_child(node, cmd->u.id.key, &ft->cmp_descriptor, ft->compare_fun);
         if (childnum == 0 || childnum == node->n_children - 1) {
             // On the extremes, promote.  We know which childnum we're going to, so pass that down too.
-            push_something_in_subtree(ft, node, childnum, cmd, flow_deltas, oldest_referenced_xid, 0, LEFT_EXTREME | RIGHT_EXTREME, false);
+            push_something_in_subtree(ft, node, childnum, cmd, flow_deltas, oldest_referenced_xid, gc_info, 0, LEFT_EXTREME | RIGHT_EXTREME, false);
         } else {
             // At height 1 in the middle, don't promote, drop the read lock and inject here.
             toku_unpin_ftnode_read_only(ft, node);
             STATUS_INC(FT_PRO_NUM_ROOT_H1_INJECT, 1);
-            inject_message_at_this_blocknum(ft, root_key, fullhash, cmd, flow_deltas, oldest_referenced_xid);
+            inject_message_at_this_blocknum(ft, root_key, fullhash, cmd, flow_deltas, oldest_referenced_xid, gc_info);
         }
     }
 }
@@ -3076,7 +3093,7 @@ void toku_ft_optimize (FT_HANDLE brt) {
         toku_init_dbt(&key);
         toku_init_dbt(&val);
         FT_MSG_S ftcmd = { FT_OPTIMIZE, ZERO_MSN, message_xids, .u = { .id = {&key,&val} } };
-        toku_ft_root_put_cmd(brt->ft, &ftcmd, TXNID_NONE);
+        toku_ft_root_put_cmd(brt->ft, &ftcmd, TXNID_NONE, make_gc_info(true));
         xids_destroy(&message_xids);
     }
 }
@@ -3151,7 +3168,7 @@ void toku_ft_maybe_insert (FT_HANDLE ft_h, DBT *key, DBT *val, TOKUTXN txn, bool
         // do nothing
     } else {
         TXNID oldest_referenced_xid = (txn) ? txn->oldest_referenced_xid : TXNID_NONE;
-        toku_ft_send_insert(ft_h, key, val, message_xids, type, oldest_referenced_xid);
+        toku_ft_send_insert(ft_h, key, val, message_xids, type, oldest_referenced_xid, make_gc_info(!txn->recovered_from_checkpoint));
     }
 }
 
@@ -3162,7 +3179,7 @@ ft_send_update_msg(FT_HANDLE brt, FT_MSG_S *msg, TOKUTXN txn) {
                  : xids_get_root_xids());
     
     TXNID oldest_referenced_xid = (txn) ? txn->oldest_referenced_xid : TXNID_NONE;
-    toku_ft_root_put_cmd(brt->ft, msg, oldest_referenced_xid);
+    toku_ft_root_put_cmd(brt->ft, msg, oldest_referenced_xid, make_gc_info(!txn->recovered_from_checkpoint));
 }
 
 void toku_ft_maybe_update(FT_HANDLE ft_h, const DBT *key, const DBT *update_function_extra,
@@ -3231,15 +3248,15 @@ void toku_ft_maybe_update_broadcast(FT_HANDLE ft_h, const DBT *update_function_e
     }
 }
 
-void toku_ft_send_insert(FT_HANDLE brt, DBT *key, DBT *val, XIDS xids, enum ft_msg_type type, TXNID oldest_referenced_xid) {
+void toku_ft_send_insert(FT_HANDLE brt, DBT *key, DBT *val, XIDS xids, enum ft_msg_type type, TXNID oldest_referenced_xid, GC_INFO gc_info) {
     FT_MSG_S ftcmd = { type, ZERO_MSN, xids, .u = { .id = { key, val } } };
-    toku_ft_root_put_cmd(brt->ft, &ftcmd, oldest_referenced_xid);
+    toku_ft_root_put_cmd(brt->ft, &ftcmd, oldest_referenced_xid, gc_info);
 }
 
-void toku_ft_send_commit_any(FT_HANDLE brt, DBT *key, XIDS xids, TXNID oldest_referenced_xid) {
+void toku_ft_send_commit_any(FT_HANDLE brt, DBT *key, XIDS xids, TXNID oldest_referenced_xid, GC_INFO gc_info) {
     DBT val;
     FT_MSG_S ftcmd = { FT_COMMIT_ANY, ZERO_MSN, xids, .u = { .id = { key, toku_init_dbt(&val) } } };
-    toku_ft_root_put_cmd(brt->ft, &ftcmd, oldest_referenced_xid);
+    toku_ft_root_put_cmd(brt->ft, &ftcmd, oldest_referenced_xid, gc_info);
 }
 
 void toku_ft_delete(FT_HANDLE brt, DBT *key, TOKUTXN txn) {
@@ -3296,14 +3313,14 @@ void toku_ft_maybe_delete(FT_HANDLE ft_h, DBT *key, TOKUTXN txn, bool oplsn_vali
         // do nothing
     } else {
         TXNID oldest_referenced_xid = (txn) ? txn->oldest_referenced_xid : TXNID_NONE;
-        toku_ft_send_delete(ft_h, key, message_xids, oldest_referenced_xid);
+        toku_ft_send_delete(ft_h, key, message_xids, oldest_referenced_xid, make_gc_info(!txn->recovered_from_checkpoint));
     }
 }
 
-void toku_ft_send_delete(FT_HANDLE brt, DBT *key, XIDS xids, TXNID oldest_referenced_xid) {
+void toku_ft_send_delete(FT_HANDLE brt, DBT *key, XIDS xids, TXNID oldest_referenced_xid, GC_INFO gc_info) {
     DBT val; toku_init_dbt(&val);
     FT_MSG_S ftcmd = { FT_DELETE_ANY, ZERO_MSN, xids, .u = { .id = { key, &val } } };
-    toku_ft_root_put_cmd(brt->ft, &ftcmd, oldest_referenced_xid);
+    toku_ft_root_put_cmd(brt->ft, &ftcmd, oldest_referenced_xid, gc_info);
 }
 
 /* mempool support */
@@ -4145,7 +4162,7 @@ int fifo_offset_msn_cmp(FIFO &fifo, const int32_t &ao, const int32_t &bo)
  * basement node.
  */
 static void
-do_bn_apply_cmd(FT_HANDLE t, BASEMENTNODE bn, struct fifo_entry *entry, uint64_t *workdone, STAT64INFO stats_to_update)
+do_bn_apply_cmd(FT_HANDLE t, BASEMENTNODE bn, struct fifo_entry *entry, TXNID oldest_referenced_xid, uint64_t *workdone, STAT64INFO stats_to_update)
 {
     // The messages are being iterated over in (key,msn) order or just in
     // msn order, so all the messages for one key, from one buffer, are in
@@ -4170,7 +4187,8 @@ do_bn_apply_cmd(FT_HANDLE t, BASEMENTNODE bn, struct fifo_entry *entry, uint64_t
             &t->ft->cmp_descriptor,
             bn,
             &ftcmd,
-            TXNID_NONE,
+            oldest_referenced_xid,
+            make_gc_info(true), //mvcc is needed
             workdone,
             stats_to_update
             );
@@ -4188,6 +4206,7 @@ struct iterate_do_bn_apply_cmd_extra {
     FT_HANDLE t;
     BASEMENTNODE bn;
     NONLEAF_CHILDINFO bnc;
+    TXNID oldest_referenced_xid;
     uint64_t *workdone;
     STAT64INFO stats_to_update;
 };
@@ -4196,7 +4215,7 @@ int iterate_do_bn_apply_cmd(const int32_t &offset, const uint32_t UU(idx), struc
 int iterate_do_bn_apply_cmd(const int32_t &offset, const uint32_t UU(idx), struct iterate_do_bn_apply_cmd_extra *const e)
 {
     struct fifo_entry *entry = toku_fifo_get_entry(e->bnc->buffer, offset);
-    do_bn_apply_cmd(e->t, e->bn, entry, e->workdone, e->stats_to_update);
+    do_bn_apply_cmd(e->t, e->bn, entry, e->oldest_referenced_xid, e->workdone, e->stats_to_update);
     return 0;
 }
 
@@ -4318,6 +4337,7 @@ bnc_apply_messages_to_basement_node(
     FTNODE ancestor,  // the ancestor node where we can find messages to apply
     int childnum,      // which child buffer of ancestor contains messages we want
     struct pivot_bounds const * const bounds,  // contains pivot key bounds of this basement node
+    TXNID oldest_referenced_xid, // may be younger than what's in ancestor, we should grab the value from the highest node we have
     bool* msgs_applied
     )
 {
@@ -4376,13 +4396,13 @@ bnc_apply_messages_to_basement_node(
         for (int i = 0; i < buffer_size; ++i) {
             *msgs_applied = true;
             struct fifo_entry *entry = toku_fifo_get_entry(bnc->buffer, offsets[i]);
-            do_bn_apply_cmd(t, bn, entry, &workdone_this_ancestor, &stats_delta);
+            do_bn_apply_cmd(t, bn, entry, oldest_referenced_xid, &workdone_this_ancestor, &stats_delta);
         }
 
         toku_free(offsets);
     } else if (stale_lbi == stale_ube) {
         // No stale messages to apply, we just apply fresh messages, and mark them to be moved to stale later.
-        struct iterate_do_bn_apply_cmd_extra iter_extra = { .t = t, .bn = bn, .bnc = bnc, .workdone = &workdone_this_ancestor, .stats_to_update = &stats_delta };
+        struct iterate_do_bn_apply_cmd_extra iter_extra = { .t = t, .bn = bn, .bnc = bnc, .oldest_referenced_xid = oldest_referenced_xid, .workdone = &workdone_this_ancestor, .stats_to_update = &stats_delta };
         if (fresh_ube - fresh_lbi > 0) *msgs_applied = true;
         r = bnc->fresh_message_tree.iterate_and_mark_range<struct iterate_do_bn_apply_cmd_extra, iterate_do_bn_apply_cmd>(fresh_lbi, fresh_ube, &iter_extra);
         assert_zero(r);
@@ -4391,7 +4411,7 @@ bnc_apply_messages_to_basement_node(
         // No fresh messages to apply, we just apply stale messages.
 
         if (stale_ube - stale_lbi > 0) *msgs_applied = true;
-        struct iterate_do_bn_apply_cmd_extra iter_extra = { .t = t, .bn = bn, .bnc = bnc, .workdone = &workdone_this_ancestor, .stats_to_update = &stats_delta };
+        struct iterate_do_bn_apply_cmd_extra iter_extra = { .t = t, .bn = bn, .bnc = bnc, .oldest_referenced_xid = oldest_referenced_xid, .workdone = &workdone_this_ancestor, .stats_to_update = &stats_delta };
 
         r = bnc->stale_message_tree.iterate_on_range<struct iterate_do_bn_apply_cmd_extra, iterate_do_bn_apply_cmd>(stale_lbi, stale_ube, &iter_extra);
         assert_zero(r);
@@ -4420,6 +4440,14 @@ toku_apply_ancestors_messages_to_node (FT_HANDLE t, FTNODE node, ANCESTORS ances
 {
     VERIFY_NODE(t, node);
     invariant(node->height == 0);
+
+    TXNID oldest_referenced_xid = ancestors->node->oldest_known_referenced_xid;
+    for (ANCESTORS curr_ancestors = ancestors; curr_ancestors; curr_ancestors = curr_ancestors->next) {
+        if (curr_ancestors->node->oldest_known_referenced_xid > oldest_referenced_xid) {
+            oldest_referenced_xid = curr_ancestors->node->oldest_known_referenced_xid;
+        }
+    }
+
     // know we are a leaf node
     // An important invariant:
     // We MUST bring every available basement node up to date.
@@ -4439,6 +4467,7 @@ toku_apply_ancestors_messages_to_node (FT_HANDLE t, FTNODE node, ANCESTORS ances
                     curr_ancestors->node,
                     curr_ancestors->childnum,
                     &curr_bounds,
+                    oldest_referenced_xid,
                     msgs_applied
                     );
                 // We don't want to check this ancestor node again if the

ft/ft-ops.h

@@ -154,9 +154,9 @@ void toku_ft_delete (FT_HANDLE brt, DBT *k, TOKUTXN txn);
 // Effect: Delete a key from a brt if the oplsn is newer than the brt lsn.  This function is called during recovery.
 void toku_ft_maybe_delete (FT_HANDLE brt, DBT *k, TOKUTXN txn, bool oplsn_valid, LSN oplsn, bool do_logging);
 
-void toku_ft_send_insert(FT_HANDLE brt, DBT *key, DBT *val, XIDS xids, enum ft_msg_type type, TXNID oldest_referenced_xid);
-void toku_ft_send_delete(FT_HANDLE brt, DBT *key, XIDS xids, TXNID oldest_referenced_xid);
-void toku_ft_send_commit_any(FT_HANDLE brt, DBT *key, XIDS xids, TXNID oldest_referenced_xids);
+void toku_ft_send_insert(FT_HANDLE brt, DBT *key, DBT *val, XIDS xids, enum ft_msg_type type, TXNID oldest_referenced_xid, GC_INFO gc_info);
+void toku_ft_send_delete(FT_HANDLE brt, DBT *key, XIDS xids, TXNID oldest_referenced_xid, GC_INFO gc_info);
+void toku_ft_send_commit_any(FT_HANDLE brt, DBT *key, XIDS xids, TXNID oldest_referenced_xids, GC_INFO gc_info);
 
 int toku_close_ft_handle_nolsn (FT_HANDLE, char **error_string)  __attribute__ ((warn_unused_result));
 

ft/ft-test-helpers.cc

@@ -145,6 +145,7 @@ int toku_testsetup_insert_to_leaf (FT_HANDLE brt, BLOCKNUM blocknum, const char
         -1,
         &cmd,
         true,
+        make_gc_info(true),
         zero_flow_deltas,
         NULL
         );

ft/ftloader.cc

@@ -2705,7 +2705,7 @@ static void add_pair_to_leafnode (struct leaf_buf *lbuf, unsigned char *key, int
                      .xids = lbuf->xids,
                      .u = { .id = { &thekey, &theval } } };
     uint64_t workdone=0;
-    toku_ft_bn_apply_cmd_once(BLB(leafnode,0), &cmd, idx, NULL, TXNID_NONE, &workdone, stats_to_update);
+    toku_ft_bn_apply_cmd_once(BLB(leafnode,0), &cmd, idx, NULL, TXNID_NONE, make_gc_info(true), &workdone, stats_to_update);
 }
 
 static int write_literal(struct dbout *out, void*data,  size_t len) {

ft/fttypes.h

@@ -48,6 +48,10 @@ typedef TOKU_XA_XID *XIDP; // this is the type that's passed to the logger code
 
 static inline BLOCKNUM make_blocknum(int64_t b) { BLOCKNUM result={b}; return result; }
 
+typedef struct gc_info_s { bool mvcc_needed; } GC_INFO;
+
+static inline GC_INFO make_gc_info(bool mvcc_needed) { GC_INFO result = {mvcc_needed}; return result; }
+
 // This struct hold information about values stored in the cachetable.
 // As one can tell from the names, we are probably violating an
 // abstraction layer by placing names.

ft/leafentry.h

@@ -170,6 +170,7 @@ toku_le_upgrade_13_14(LEAFENTRY_13 old_leafentry, // NULL if there was no stored
 void toku_le_apply_msg(FT_MSG   msg,
                        LEAFENTRY old_leafentry, // NULL if there was no stored data.
                        TXNID oldest_referenced_xid,
+                       GC_INFO gc_info,
                        size_t *new_leafentry_memorysize,
                        LEAFENTRY *new_leafentry_p,
                        OMT *omtp,

ft/roll.cc

@@ -177,7 +177,7 @@ static int do_insertion (enum ft_msg_type type, FILENUM filenum, BYTESTRING key,
                                           ? toku_fill_dbt(&data_dbt, data->data, data->len)
                                           : toku_init_dbt(&data_dbt) } } };
 
-        toku_ft_root_put_cmd(h, &ftcmd, txn->oldest_referenced_xid);
+        toku_ft_root_put_cmd(h, &ftcmd, txn->oldest_referenced_xid, make_gc_info(!txn->recovered_from_checkpoint));
         if (reset_root_xid_that_created) {
             TXNID new_root_xid_that_created = xids_get_outermost_xid(xids);
             toku_reset_root_xid_that_created(h, new_root_xid_that_created);

ft/tests/make-tree.cc

@@ -40,7 +40,7 @@ append_leaf(FTNODE leafnode, void *key, size_t keylen, void *val, size_t vallen)
 
     // apply an insert to the leaf node
     FT_MSG_S cmd = { FT_INSERT, msn, xids_get_root_xids(), .u = {.id = { &thekey, &theval }} };
-    toku_ft_bn_apply_cmd_once(BLB(leafnode,0), &cmd, idx, NULL, TXNID_NONE, NULL, NULL);
+    toku_ft_bn_apply_cmd_once(BLB(leafnode,0), &cmd, idx, NULL, TXNID_NONE, make_gc_info(false), NULL, NULL);
 
     leafnode->max_msn_applied_to_node_on_disk = msn;
 

ft/tests/msnfilter.cc

@@ -48,7 +48,7 @@ append_leaf(FT_HANDLE brt, FTNODE leafnode, void *key, uint32_t keylen, void *va
     brt->ft->h->max_msn_in_ft = msn;
     FT_MSG_S cmd = { FT_INSERT, msn, xids_get_root_xids(), .u={.id = { &thekey, &theval }} };
 
-    toku_ft_leaf_apply_cmd(brt->ft->compare_fun, brt->ft->update_fun, &brt->ft->cmp_descriptor, leafnode, -1, &cmd, nullptr, nullptr);
+    toku_ft_leaf_apply_cmd(brt->ft->compare_fun, brt->ft->update_fun, &brt->ft->cmp_descriptor, leafnode, -1, &cmd, make_gc_info(false), nullptr, nullptr);
     {
 	int r = toku_ft_lookup(brt, &thekey, lookup_checkf, &pair);
 	assert(r==0);
@@ -56,7 +56,7 @@ append_leaf(FT_HANDLE brt, FTNODE leafnode, void *key, uint32_t keylen, void *va
     }
 
     FT_MSG_S badcmd = { FT_INSERT, msn, xids_get_root_xids(), .u={.id = { &thekey, &badval }} };
-    toku_ft_leaf_apply_cmd(brt->ft->compare_fun, brt->ft->update_fun, &brt->ft->cmp_descriptor, leafnode, -1, &badcmd, nullptr, nullptr);
+    toku_ft_leaf_apply_cmd(brt->ft->compare_fun, brt->ft->update_fun, &brt->ft->cmp_descriptor, leafnode, -1, &badcmd, make_gc_info(false), nullptr, nullptr);
 
     // message should be rejected for duplicate msn, row should still have original val
     {
@@ -69,7 +69,7 @@ append_leaf(FT_HANDLE brt, FTNODE leafnode, void *key, uint32_t keylen, void *va
     msn = next_dummymsn();
     brt->ft->h->max_msn_in_ft = msn;
     FT_MSG_S cmd2 = { FT_INSERT, msn, xids_get_root_xids(), .u={.id = { &thekey, &val2 }} };
-    toku_ft_leaf_apply_cmd(brt->ft->compare_fun, brt->ft->update_fun, &brt->ft->cmp_descriptor, leafnode, -1, &cmd2, nullptr, nullptr);
+    toku_ft_leaf_apply_cmd(brt->ft->compare_fun, brt->ft->update_fun, &brt->ft->cmp_descriptor, leafnode, -1, &cmd2, make_gc_info(false), nullptr, nullptr);
 
     // message should be accepted, val should have new value
     {
@@ -81,7 +81,7 @@ append_leaf(FT_HANDLE brt, FTNODE leafnode, void *key, uint32_t keylen, void *va
     // now verify that message with lesser (older) msn is rejected
     msn.msn = msn.msn - 10;
     FT_MSG_S cmd3 = { FT_INSERT, msn, xids_get_root_xids(), .u={.id = { &thekey, &badval } }};
-    toku_ft_leaf_apply_cmd(brt->ft->compare_fun, brt->ft->update_fun, &brt->ft->cmp_descriptor, leafnode, -1, &cmd3, nullptr, nullptr);
+    toku_ft_leaf_apply_cmd(brt->ft->compare_fun, brt->ft->update_fun, &brt->ft->cmp_descriptor, leafnode, -1, &cmd3, make_gc_info(false), nullptr, nullptr);
 
     // message should be rejected, val should still have value in pair2
     {

ft/tests/orthopush-flush.cc

@@ -123,8 +123,8 @@ insert_random_message_to_bn(FT_HANDLE t, BASEMENTNODE blb, LEAFENTRY *save, XIDS
     msg.u.id.val = valdbt;
     size_t memsize;
     int64_t numbytes;
-    toku_le_apply_msg(&msg, NULL, TXNID_NONE, &memsize, save, NULL, NULL, NULL, &numbytes);
-    toku_ft_bn_apply_cmd(t->ft->compare_fun, t->ft->update_fun, NULL, blb, &msg, TXNID_NONE, NULL, NULL);
+    toku_le_apply_msg(&msg, NULL, TXNID_NONE, make_gc_info(false), &memsize, save, NULL, NULL, NULL, &numbytes);
+    toku_ft_bn_apply_cmd(t->ft->compare_fun, t->ft->update_fun, NULL, blb, &msg, TXNID_NONE, make_gc_info(false), NULL, NULL);
     if (msn.msn > blb->max_msn_applied.msn) {
         blb->max_msn_applied = msn;
     }
@@ -163,12 +163,12 @@ insert_same_message_to_bns(FT_HANDLE t, BASEMENTNODE blb1, BASEMENTNODE blb2, LE
     msg.u.id.val = valdbt;
     size_t memsize;
     int64_t numbytes;
-    toku_le_apply_msg(&msg, NULL, TXNID_NONE, &memsize, save, NULL, NULL, NULL, &numbytes);
-    toku_ft_bn_apply_cmd(t->ft->compare_fun, t->ft->update_fun, NULL, blb1, &msg, TXNID_NONE, NULL, NULL);
+    toku_le_apply_msg(&msg, NULL, TXNID_NONE, make_gc_info(false), &memsize, save, NULL, NULL, NULL, &numbytes);
+    toku_ft_bn_apply_cmd(t->ft->compare_fun, t->ft->update_fun, NULL, blb1, &msg, TXNID_NONE, make_gc_info(false), NULL, NULL);
     if (msn.msn > blb1->max_msn_applied.msn) {
         blb1->max_msn_applied = msn;
     }
-    toku_ft_bn_apply_cmd(t->ft->compare_fun, t->ft->update_fun, NULL, blb2, &msg, TXNID_NONE, NULL, NULL);
+    toku_ft_bn_apply_cmd(t->ft->compare_fun, t->ft->update_fun, NULL, blb2, &msg, TXNID_NONE, make_gc_info(false), NULL, NULL);
     if (msn.msn > blb2->max_msn_applied.msn) {
         blb2->max_msn_applied = msn;
     }
@@ -578,7 +578,7 @@ flush_to_leaf(FT_HANDLE t, bool make_leaf_up_to_date, bool use_flush) {
     if (make_leaf_up_to_date) {
         for (i = 0; i < num_parent_messages; ++i) {
             if (!parent_messages_is_fresh[i]) {
-                toku_ft_leaf_apply_cmd(t->ft->compare_fun, t->ft->update_fun, &t->ft->descriptor, child, -1, parent_messages[i], NULL, NULL);
+                toku_ft_leaf_apply_cmd(t->ft->compare_fun, t->ft->update_fun, &t->ft->descriptor, child, -1, parent_messages[i], make_gc_info(false), NULL, NULL);
             }
         }
         for (i = 0; i < 8; ++i) {
@@ -801,7 +801,7 @@ flush_to_leaf_with_keyrange(FT_HANDLE t, bool make_leaf_up_to_date) {
         for (i = 0; i < num_parent_messages; ++i) {
             if (dummy_cmp(NULL, parent_messages[i]->u.id.key, &childkeys[7]) <= 0 &&
                 !parent_messages_is_fresh[i]) {
-                toku_ft_leaf_apply_cmd(t->ft->compare_fun, t->ft->update_fun, &t->ft->descriptor, child, -1, parent_messages[i], NULL, NULL);
+                toku_ft_leaf_apply_cmd(t->ft->compare_fun, t->ft->update_fun, &t->ft->descriptor, child, -1, parent_messages[i], make_gc_info(false), NULL, NULL);
             }
         }
         for (i = 0; i < 8; ++i) {
@@ -993,8 +993,8 @@ compare_apply_and_flush(FT_HANDLE t, bool make_leaf_up_to_date) {
     if (make_leaf_up_to_date) {
         for (i = 0; i < num_parent_messages; ++i) {
             if (!parent_messages_is_fresh[i]) {
-                toku_ft_leaf_apply_cmd(t->ft->compare_fun, t->ft->update_fun, &t->ft->descriptor, child1, -1, parent_messages[i], NULL, NULL);
-                toku_ft_leaf_apply_cmd(t->ft->compare_fun, t->ft->update_fun, &t->ft->descriptor, child2, -1, parent_messages[i], NULL, NULL);
+                toku_ft_leaf_apply_cmd(t->ft->compare_fun, t->ft->update_fun, &t->ft->descriptor, child1, -1, parent_messages[i], make_gc_info(false), NULL, NULL);
+                toku_ft_leaf_apply_cmd(t->ft->compare_fun, t->ft->update_fun, &t->ft->descriptor, child2, -1, parent_messages[i], make_gc_info(false), NULL, NULL);
             }
         }
         for (i = 0; i < 8; ++i) {

ft/tests/test-leafentry-nested.cc

@@ -399,6 +399,7 @@ test_le_apply(ULE ule_initial, FT_MSG msg, ULE ule_expected) {
     toku_le_apply_msg(msg,
                       le_initial,
                       TXNID_NONE,
+                      make_gc_info(true),
                       &result_memsize,
                       &le_result,
                       NULL, 

ft/tests/verify-bad-msn.cc

@@ -43,7 +43,7 @@ append_leaf(FTNODE leafnode, void *key, size_t keylen, void *val, size_t vallen)
 
     // apply an insert to the leaf node
     FT_MSG_S cmd = { FT_INSERT, msn, xids_get_root_xids(), .u={.id = { &thekey, &theval }} };
-    toku_ft_bn_apply_cmd_once(BLB(leafnode, 0), &cmd, idx, NULL, TXNID_NONE, NULL, NULL);
+    toku_ft_bn_apply_cmd_once(BLB(leafnode, 0), &cmd, idx, NULL, TXNID_NONE, make_gc_info(false), NULL, NULL);
 
     // Create bad tree (don't do following):
     // leafnode->max_msn_applied_to_node = msn;

ft/tests/verify-bad-pivots.cc

@@ -31,7 +31,7 @@ append_leaf(FTNODE leafnode, void *key, size_t keylen, void *val, size_t vallen)
     // apply an insert to the leaf node
     MSN msn = next_dummymsn();
     FT_MSG_S cmd = { FT_INSERT, msn, xids_get_root_xids(), .u={.id = { &thekey, &theval }} };
-    toku_ft_bn_apply_cmd_once(BLB(leafnode, 0), &cmd, idx, NULL, TXNID_NONE, NULL, NULL);
+    toku_ft_bn_apply_cmd_once(BLB(leafnode, 0), &cmd, idx, NULL, TXNID_NONE, make_gc_info(false), NULL, NULL);
 
     // dont forget to dirty the node
     leafnode->dirty = 1;

ft/tests/verify-dup-in-leaf.cc

@@ -32,7 +32,7 @@ append_leaf(FTNODE leafnode, void *key, size_t keylen, void *val, size_t vallen)
     // apply an insert to the leaf node
     MSN msn = next_dummymsn();
     FT_MSG_S cmd = { FT_INSERT, msn, xids_get_root_xids(), .u={.id = { &thekey, &theval }} };
-    toku_ft_bn_apply_cmd_once(BLB(leafnode, 0), &cmd, idx, NULL, TXNID_NONE, NULL, NULL);
+    toku_ft_bn_apply_cmd_once(BLB(leafnode, 0), &cmd, idx, NULL, TXNID_NONE, make_gc_info(false), NULL, NULL);
 
     // dont forget to dirty the node
     leafnode->dirty = 1;

ft/tests/verify-dup-pivots.cc

@@ -31,7 +31,7 @@ append_leaf(FTNODE leafnode, void *key, size_t keylen, void *val, size_t vallen)
     // apply an insert to the leaf node
     MSN msn = next_dummymsn();
     FT_MSG_S cmd = { FT_INSERT, msn, xids_get_root_xids(), .u={.id = { &thekey, &theval }} };
-    toku_ft_bn_apply_cmd_once(BLB(leafnode, 0), &cmd, idx, NULL, TXNID_NONE, NULL, NULL);
+    toku_ft_bn_apply_cmd_once(BLB(leafnode, 0), &cmd, idx, NULL, TXNID_NONE, make_gc_info(false), NULL, NULL);
 
     // dont forget to dirty the node
     leafnode->dirty = 1;

ft/tests/verify-misrouted-msgs.cc

@@ -32,7 +32,7 @@ append_leaf(FTNODE leafnode, void *key, size_t keylen, void *val, size_t vallen)
     // apply an insert to the leaf node
     MSN msn = next_dummymsn();
     FT_MSG_S cmd = { FT_INSERT, msn, xids_get_root_xids(), .u={.id = { &thekey, &theval }} };
-    toku_ft_bn_apply_cmd_once(BLB(leafnode,0), &cmd, idx, NULL, TXNID_NONE, NULL, NULL);
+    toku_ft_bn_apply_cmd_once(BLB(leafnode,0), &cmd, idx, NULL, TXNID_NONE, make_gc_info(false), NULL, NULL);
 
     // dont forget to dirty the node
     leafnode->dirty = 1;

ft/tests/verify-unsorted-leaf.cc

@@ -34,7 +34,7 @@ append_leaf(FTNODE leafnode, void *key, size_t keylen, void *val, size_t vallen)
     // apply an insert to the leaf node
     MSN msn = next_dummymsn();
     FT_MSG_S cmd = { FT_INSERT, msn, xids_get_root_xids(), .u={.id = { &thekey, &theval }} };
-    toku_ft_bn_apply_cmd_once(BLB(leafnode, 0), &cmd, idx, NULL, TXNID_NONE, NULL, NULL);
+    toku_ft_bn_apply_cmd_once(BLB(leafnode, 0), &cmd, idx, NULL, TXNID_NONE, make_gc_info(false), NULL, NULL);
 
     // dont forget to dirty the node
     leafnode->dirty = 1;

ft/tests/verify-unsorted-pivots.cc

@@ -31,7 +31,7 @@ append_leaf(FTNODE leafnode, void *key, size_t keylen, void *val, size_t vallen)
     // apply an insert to the leaf node
     MSN msn = next_dummymsn();
     FT_MSG_S cmd = { FT_INSERT, msn, xids_get_root_xids(), .u={.id = { &thekey, &theval }} };
-    toku_ft_bn_apply_cmd_once(BLB(leafnode, 0), &cmd, idx, NULL, TXNID_NONE, NULL, NULL);
+    toku_ft_bn_apply_cmd_once(BLB(leafnode, 0), &cmd, idx, NULL, TXNID_NONE, make_gc_info(false), NULL, NULL);
 
     // dont forget to dirty the node
     leafnode->dirty = 1;

ft/ule.cc

@@ -212,12 +212,13 @@ xid_reads_committed_xid(TXNID tl1, TXNID xc, const xid_omt_t &snapshot_txnids, c
 // so we get rid of them.
 //
 static void
-ule_simple_garbage_collection(ULE ule, TXNID oldest_referenced_xid) {
+ule_simple_garbage_collection(ULE ule, TXNID oldest_referenced_xid, GC_INFO gc_info) {
     uint32_t curr_index = 0;
     uint32_t num_entries;
-    if (ule->num_cuxrs == 1 || oldest_referenced_xid == TXNID_NONE) {
+    if (ule->num_cuxrs == 1) {
         goto done;
     }
+    if (gc_info.mvcc_needed) {
         // starting at the top of the committed stack, find the first
         // uxr with a txnid that is less than oldest_referenced_xid
         for (uint32_t i = 0; i < ule->num_cuxrs; i++) {
@@ -226,6 +227,12 @@ ule_simple_garbage_collection(ULE ule, TXNID oldest_referenced_xid) {
                 break;
             }
         }
+    }
+    else {
+        // if mvcc is not needed, we can need the top committed
+        // value and nothing else
+        curr_index = ule->num_cuxrs - 1;
+    }
     // curr_index is now set to the youngest uxr older than oldest_referenced_xid
     if (curr_index == 0) {
         goto done;
@@ -353,6 +360,7 @@ void
 toku_le_apply_msg(FT_MSG   msg,		// message to apply to leafentry
                   LEAFENTRY old_leafentry, // NULL if there was no stored data.
                   TXNID oldest_referenced_xid,
+                  GC_INFO gc_info,
                   size_t *new_leafentry_memorysize,
                   LEAFENTRY *new_leafentry_p,
                   OMT *omtp,
@@ -370,7 +378,7 @@ toku_le_apply_msg(FT_MSG   msg,		// message to apply to leafentry
 	oldnumbytes = ule_get_innermost_numbytes(&ule);
     }
     msg_modify_ule(&ule, msg);          // modify unpacked leafentry
-    ule_simple_garbage_collection(&ule, oldest_referenced_xid);
+    ule_simple_garbage_collection(&ule, oldest_referenced_xid, gc_info);
     int rval = le_pack(&ule,                // create packed leafentry
                        new_leafentry_memorysize,
                        new_leafentry_p,

src/indexer-undo-do.cc

@@ -506,7 +506,7 @@ indexer_ft_delete_committed(DB_INDEXER *indexer, DB *hotdb, DBT *hotkey, XIDS xi
     } else {
         result = toku_ydb_check_avail_fs_space(indexer->i->env);
         if (result == 0) {
-            toku_ft_send_delete(db_struct_i(hotdb)->ft_handle, hotkey, xids, TXNID_NONE);
+            toku_ft_send_delete(db_struct_i(hotdb)->ft_handle, hotkey, xids, TXNID_NONE, make_gc_info(true));
         }
     }
     return result;
@@ -542,7 +542,7 @@ indexer_ft_insert_committed(DB_INDEXER *indexer, DB *hotdb, DBT *hotkey, DBT *ho
     } else {
         result = toku_ydb_check_avail_fs_space(indexer->i->env);
         if (result == 0) {
-            toku_ft_send_insert(db_struct_i(hotdb)->ft_handle, hotkey, hotval, xids, FT_INSERT, TXNID_NONE);
+            toku_ft_send_insert(db_struct_i(hotdb)->ft_handle, hotkey, hotval, xids, FT_INSERT, TXNID_NONE, make_gc_info(true));
         }
     }
     return result;
@@ -562,7 +562,7 @@ indexer_ft_commit(DB_INDEXER *indexer, DB *hotdb, DBT *hotkey, XIDS xids) {
         } else {
             result = toku_ydb_check_avail_fs_space(indexer->i->env);
             if (result == 0)
-                toku_ft_send_commit_any(db_struct_i(hotdb)->ft_handle, hotkey, xids, TXNID_NONE);
+                toku_ft_send_commit_any(db_struct_i(hotdb)->ft_handle, hotkey, xids, TXNID_NONE, make_gc_info(true));
         }
     }
     return result;