refs #5700, merge to main

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

ft/ft-internal.h

@@ -913,7 +913,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);
+void toku_ft_root_put_cmd(FT h, FT_MSG_S * cmd, TXNID oldest_referenced_xid);
 
 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
@@ -1073,6 +1073,7 @@ toku_ft_bn_apply_cmd_once (
     const FT_MSG cmd,
     uint32_t idx,
     LEAFENTRY le,
+    TXNID oldest_referenced_xid,
     uint64_t *workdonep,
     STAT64INFO stats_to_update
     );
@@ -1084,6 +1085,7 @@ toku_ft_bn_apply_cmd (
     DESCRIPTOR desc,
     BASEMENTNODE bn,
     FT_MSG cmd,
+    TXNID oldest_referenced_xid,
     uint64_t *workdone,
     STAT64INFO stats_to_update
     );
@@ -1096,6 +1098,7 @@ toku_ft_leaf_apply_cmd (
     FTNODE node,
     int target_childnum,
     FT_MSG cmd,
+    TXNID oldest_referenced_xid,
     uint64_t *workdone,
     STAT64INFO stats_to_update
     );
@@ -1110,6 +1113,7 @@ toku_ft_node_put_cmd (
     FT_MSG cmd,
     bool is_fresh,
     size_t flow_deltas[],
+    TXNID oldest_referenced_xid,
     STAT64INFO stats_to_update
     );
 

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);
-void toku_ft_send_delete(FT_HANDLE brt, DBT *key, XIDS xids);
-void toku_ft_send_commit_any(FT_HANDLE brt, DBT *key, XIDS xids);
+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);
 
 int toku_close_ft_handle_nolsn (FT_HANDLE, char **error_string)  __attribute__ ((warn_unused_result));
 

ft/ft-test-helpers.cc

@@ -146,6 +146,7 @@ int toku_testsetup_insert_to_leaf (FT_HANDLE brt, BLOCKNUM blocknum, const char
         &cmd,
         true,
         zero_flow_deltas,
+        TXNID_NONE,
         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, &workdone, stats_to_update);
+    toku_ft_bn_apply_cmd_once(BLB(leafnode,0), &cmd, idx, NULL, TXNID_NONE, &workdone, stats_to_update);
 }
 
 static int write_literal(struct dbout *out, void*data,  size_t len) {

ft/log-internal.h

@@ -152,6 +152,7 @@ struct tokutxn {
     DB_TXN *container_db_txn; // reference to DB_TXN that contains this tokutxn
     xid_omt_t *live_root_txn_list; // the root txns live when the root ancestor (self if a root) started.
     XIDS xids; // Represents the xid list
+    TXNID oldest_referenced_xid;
 
     bool begin_was_logged;
     // These are not read until a commit, prepare, or abort starts, and

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);
+        toku_ft_root_put_cmd(h, &ftcmd, txn->oldest_referenced_xid);
         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, NULL, NULL);
+    toku_ft_bn_apply_cmd_once(BLB(leafnode,0), &cmd, idx, NULL, TXNID_NONE, 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, TXNID_NONE, 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, TXNID_NONE, 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, TXNID_NONE, 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, TXNID_NONE, nullptr, nullptr);
 
     // message should be rejected, val should still have value in pair2
     {

ft/tests/orthopush-flush.cc

@@ -123,9 +123,9 @@ insert_random_message_to_bn(FT_HANDLE t, BASEMENTNODE blb, LEAFENTRY *save, XIDS
     msg.u.id.val = valdbt;
     size_t memsize;
     int64_t numbytes;
-    int r = apply_msg_to_leafentry(&msg, NULL, &memsize, save, NULL, NULL, NULL, &numbytes);
+    int r = apply_msg_to_leafentry(&msg, NULL, TXNID_NONE, &memsize, save, NULL, NULL, NULL, &numbytes);
     assert_zero(r);
-    toku_ft_bn_apply_cmd(t->ft->compare_fun, t->ft->update_fun, NULL, blb, &msg, NULL, NULL);
+    toku_ft_bn_apply_cmd(t->ft->compare_fun, t->ft->update_fun, NULL, blb, &msg, TXNID_NONE, NULL, NULL);
     if (msn.msn > blb->max_msn_applied.msn) {
         blb->max_msn_applied = msn;
     }
@@ -164,13 +164,13 @@ insert_same_message_to_bns(FT_HANDLE t, BASEMENTNODE blb1, BASEMENTNODE blb2, LE
     msg.u.id.val = valdbt;
     size_t memsize;
     int64_t numbytes;
-    int r = apply_msg_to_leafentry(&msg, NULL, &memsize, save, NULL, NULL, NULL, &numbytes);
+    int r = apply_msg_to_leafentry(&msg, NULL, TXNID_NONE, &memsize, save, NULL, NULL, NULL, &numbytes);
     assert_zero(r);
-    toku_ft_bn_apply_cmd(t->ft->compare_fun, t->ft->update_fun, NULL, blb1, &msg, NULL, NULL);
+    toku_ft_bn_apply_cmd(t->ft->compare_fun, t->ft->update_fun, NULL, blb1, &msg, TXNID_NONE, 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, NULL, NULL);
+    toku_ft_bn_apply_cmd(t->ft->compare_fun, t->ft->update_fun, NULL, blb2, &msg, TXNID_NONE, NULL, NULL);
     if (msn.msn > blb2->max_msn_applied.msn) {
         blb2->max_msn_applied = msn;
     }
@@ -580,7 +580,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], TXNID_NONE, NULL, NULL);
             }
         }
         for (i = 0; i < 8; ++i) {
@@ -803,7 +803,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], TXNID_NONE, NULL, NULL);
             }
         }
         for (i = 0; i < 8; ++i) {
@@ -995,8 +995,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], TXNID_NONE, NULL, NULL);
+                toku_ft_leaf_apply_cmd(t->ft->compare_fun, t->ft->update_fun, &t->ft->descriptor, child2, -1, parent_messages[i], TXNID_NONE, NULL, NULL);
             }
         }
         for (i = 0; i < 8; ++i) {

ft/tests/test-leafentry-nested.cc

@@ -398,6 +398,7 @@ test_le_apply(ULE ule_initial, FT_MSG msg, ULE ule_expected) {
     int64_t ignoreme;
     r = apply_msg_to_leafentry(msg,
                                le_initial,
+                               TXNID_NONE,
                                &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, NULL, NULL);
+    toku_ft_bn_apply_cmd_once(BLB(leafnode, 0), &cmd, idx, NULL, TXNID_NONE, 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, NULL, NULL);
+    toku_ft_bn_apply_cmd_once(BLB(leafnode, 0), &cmd, idx, NULL, TXNID_NONE, 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, NULL, NULL);
+    toku_ft_bn_apply_cmd_once(BLB(leafnode, 0), &cmd, idx, NULL, TXNID_NONE, 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, NULL, NULL);
+    toku_ft_bn_apply_cmd_once(BLB(leafnode, 0), &cmd, idx, NULL, TXNID_NONE, 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, NULL, NULL);
+    toku_ft_bn_apply_cmd_once(BLB(leafnode,0), &cmd, idx, NULL, TXNID_NONE, 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, NULL, NULL);
+    toku_ft_bn_apply_cmd_once(BLB(leafnode, 0), &cmd, idx, NULL, TXNID_NONE, 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, NULL, NULL);
+    toku_ft_bn_apply_cmd_once(BLB(leafnode, 0), &cmd, idx, NULL, TXNID_NONE, NULL, NULL);
 
     // dont forget to dirty the node
     leafnode->dirty = 1;

ft/txn.cc

@@ -151,6 +151,7 @@ void toku_txn_create_txn (
         .container_db_txn = container_db_txn,
         .live_root_txn_list = nullptr,
         .xids = xids,
+        .oldest_referenced_xid = TXNID_NONE,
         .begin_was_logged = false,
         .do_fsync = false,
         .force_fsync_on_commit = false,

ft/txn_manager.cc

@@ -255,6 +255,23 @@ max_xid(TXNID a, TXNID b) {
     return a < b ? b : a;
 }
 
+static TXNID get_oldest_referenced_xid_unlocked(TXN_MANAGER txn_manager) {
+    TXNID oldest_referenced_xid = TXNID_NONE_LIVING;
+    int r = txn_manager->live_root_txns.fetch(0, &oldest_referenced_xid);
+    // this function should only be called when we know there is at least
+    // one live transaction
+    invariant_zero(r);
+
+    struct referenced_xid_tuple* tuple;
+    if (txn_manager->referenced_xids.size() > 0) {
+        r = txn_manager->referenced_xids.fetch(0, &tuple);
+        if (r == 0 && tuple->begin_id < oldest_referenced_xid) {
+            oldest_referenced_xid = tuple->begin_id;
+        }
+    }
+    return oldest_referenced_xid;
+}
+
 int toku_txn_manager_start_txn(
     TOKUTXN *txnp,
     TXN_MANAGER txn_manager,
@@ -318,6 +335,7 @@ int toku_txn_manager_start_txn(
         lazy_assert_zero(r);
     }
 
+
     {
         //
         // maintain the data structures necessary for MVCC:
@@ -342,6 +360,7 @@ int toku_txn_manager_start_txn(
             }
             r = txn_manager->live_root_txns.insert_at(txn->txnid64, idx);
         }
+        txn->oldest_referenced_xid = get_oldest_referenced_xid_unlocked(txn_manager);
 
         // setup information for snapshot reads
         if (txn->snapshot_type != TXN_SNAPSHOT_NONE) {

ft/ule.cc

@@ -207,6 +207,42 @@ xid_reads_committed_xid(TXNID tl1, TXNID xc, const xid_omt_t &snapshot_txnids, c
     return rval;
 }
 
+//
+// This function does some simple garbage collection given a TXNID known
+// to be the oldest referenced xid, that is, the oldest xid in any live list.
+// We find the youngest entry in the stack with an xid less 
+// than oldest_referenced_xid. All elements below this entry are garbage,
+// so we get rid of them.
+//
+static void
+simple_garbage_collection(ULE ule, TXNID oldest_referenced_xid) {
+    uint32_t curr_index = 0;
+    uint32_t num_entries;
+    if (ule->num_cuxrs == 1 || oldest_referenced_xid == TXNID_NONE) {
+        goto done;
+    }
+    // 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++) {
+        curr_index = ule->num_cuxrs - i - 1;
+        if (ule->uxrs[curr_index].xid < oldest_referenced_xid) {
+            break;
+        }
+    }
+    // curr_index is now set to the youngest uxr older than oldest_referenced_xid
+    if (curr_index == 0) {
+        goto done;
+    }
+
+    // now get rid of the entries below curr_index
+    num_entries = ule->num_cuxrs + ule->num_puxrs - curr_index;
+    memmove(&ule->uxrs[0], &ule->uxrs[curr_index], num_entries * sizeof(ule->uxrs[0]));
+    ule->uxrs[0].xid = TXNID_NONE; //New 'bottom of stack' loses its TXNID
+    ule->num_cuxrs -= curr_index;
+    
+done:;
+}
+
 static void
 garbage_collection(ULE ule, const xid_omt_t &snapshot_xids, const rx_omt_t &referenced_xids, const xid_omt_t &live_root_txns) {
     if (ule->num_cuxrs == 1) goto done;
@@ -320,6 +356,7 @@ done:;
 int
 apply_msg_to_leafentry(FT_MSG   msg,		// message to apply to leafentry
                        LEAFENTRY old_leafentry, // NULL if there was no stored data.
+                       TXNID oldest_referenced_xid,
                        size_t *new_leafentry_memorysize,
                        LEAFENTRY *new_leafentry_p,
                        OMT *omtp,
@@ -338,6 +375,7 @@ apply_msg_to_leafentry(FT_MSG   msg,		// message to apply to leafentry
 	oldnumbytes = ule_get_innermost_numbytes(&ule);
     }
     msg_modify_ule(&ule, msg);          // modify unpacked leafentry
+    simple_garbage_collection(&ule, oldest_referenced_xid);
     rval = le_pack(&ule,                // create packed leafentry
                    new_leafentry_memorysize,
                    new_leafentry_p,

ft/ule.h

@@ -54,6 +54,7 @@ void fast_msg_to_leafentry(
 
 int apply_msg_to_leafentry(FT_MSG   msg,
                            LEAFENTRY old_leafentry, // NULL if there was no stored data.
+                           TXNID oldest_referenced_xid,
                            size_t *new_leafentry_memorysize,
                            LEAFENTRY *new_leafentry_p,
                            OMT *omtp,

src/indexer-undo-do.cc

@@ -509,7 +509,7 @@ indexer_ft_delete_committed(DB_INDEXER *indexer, DB *hotdb, DBT *hotkey, XIDS xi
         result = toku_ydb_check_avail_fs_space(indexer->i->env);
         if (result == 0) {
             toku_multi_operation_client_lock();
-            toku_ft_send_delete(db_struct_i(hotdb)->ft_handle, hotkey, xids);
+            toku_ft_send_delete(db_struct_i(hotdb)->ft_handle, hotkey, xids, TXNID_NONE);
             toku_multi_operation_client_unlock();
         }
     }
@@ -549,7 +549,7 @@ indexer_ft_insert_committed(DB_INDEXER *indexer, DB *hotdb, DBT *hotkey, DBT *ho
         result = toku_ydb_check_avail_fs_space(indexer->i->env);
         if (result == 0) {
             toku_multi_operation_client_lock();
-            toku_ft_send_insert(db_struct_i(hotdb)->ft_handle, hotkey, hotval, xids, FT_INSERT);
+            toku_ft_send_insert(db_struct_i(hotdb)->ft_handle, hotkey, hotval, xids, FT_INSERT, TXNID_NONE);
             toku_multi_operation_client_unlock();
         }
     }
@@ -570,7 +570,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);
+                toku_ft_send_commit_any(db_struct_i(hotdb)->ft_handle, hotkey, xids, TXNID_NONE);
         }
     }
     return result;

src/tests/test_updates_single_key.cc

+/* -*- mode: C++; c-basic-offset: 4; indent-tabs-mode: nil -*- */
+// vim: ft=cpp:expandtab:ts=8:sw=4:softtabstop=4:
+#ident "$Id: test_cursor_2.cc 45903 2012-07-19 13:06:39Z leifwalsh $"
+#ident "Copyright (c) 2007 Tokutek Inc.  All rights reserved."
+#include "test.h"
+
+#include <stdio.h>
+#include <stdlib.h>
+
+#include <unistd.h>
+#include <memory.h>
+#include <errno.h>
+#include <sys/stat.h>
+#include <db.h>
+
+
+//
+// This test ensures that we can do many updates to a single key when the dictionary
+// is just that key.
+//
+static void
+run_test (void) {
+
+    DB_ENV * env;
+    DB *db;
+    const char * const fname = "test.updates_single_key.ft_handle";
+    int r;
+
+    r = db_env_create(&env, 0);        assert(r == 0);
+    env->set_errfile(env, stderr);
+    // no need to run with logging, so DB_INIT_LOG not passed in
+    r = env->open(env, ENVDIR, DB_CREATE | DB_PRIVATE | DB_INIT_MPOOL | DB_INIT_TXN | DB_INIT_LOCK | DB_THREAD, S_IRWXU+S_IRWXG+S_IRWXO); CKERR(r);
+    r = db_create(&db, env, 0); assert(r == 0);
+    db->set_errfile(db,stderr); // Turn off those annoying errors
+    r = db->open(db, NULL, fname, "main", DB_BTREE, DB_CREATE, 0666); assert(r == 0);
+
+    int i;
+    for (i=0; i<1000000; i++) {
+        int k = 1;
+        int v = i;
+        DBT key, val;
+        DB_TXN* txn = NULL;
+        r = env->txn_begin(env, NULL, &txn, 0);
+        CKERR(r);
+        // want this test to go as fast as possible, so no need to use the lock tree
+        // we just care to see that #5700 is behaving better, that some garbage collection is happening
+        r = db->put(db, txn, dbt_init(&key, &k, sizeof k), dbt_init(&val, &v, sizeof v), DB_PRELOCKED_WRITE);
+        txn->commit(txn, DB_TXN_NOSYNC);
+        CKERR(r);
+    }
+
+    r = db->close(db, 0); assert(r == 0);
+    r = env->close(env, 0); assert(r == 0);
+}
+
+int
+test_main(int argc, char *const argv[]) {
+
+    parse_args(argc, argv);
+  
+    int r;
+    r = system("rm -rf " ENVDIR);
+    CKERR(r);
+    toku_os_mkdir(ENVDIR, S_IRWXU+S_IRWXG+S_IRWXO);
+    
+    run_test();
+
+    return 0;
+}