Merge commit 'tokudb-ft-index/tokudb-7.5.6' into 5.5

storage/tokudb/ft-index/buildheader/make_tdb.cc

@@ -572,7 +572,7 @@ static void print_db_txn_struct (void) {
     STRUCT_SETUP(DB_TXN, abort,       "int (*%s) (DB_TXN *)");
     STRUCT_SETUP(DB_TXN, api_internal,"void *%s");
     STRUCT_SETUP(DB_TXN, commit,      "int (*%s) (DB_TXN*, uint32_t)");
-    STRUCT_SETUP(DB_TXN, prepare,     "int (*%s) (DB_TXN*, uint8_t gid[DB_GID_SIZE])");
+    STRUCT_SETUP(DB_TXN, prepare,     "int (*%s) (DB_TXN*, uint8_t gid[DB_GID_SIZE], uint32_t flags)");
     STRUCT_SETUP(DB_TXN, discard,     "int (*%s) (DB_TXN*, uint32_t)");
     STRUCT_SETUP(DB_TXN, id,          "uint32_t (*%s) (DB_TXN *)");
     STRUCT_SETUP(DB_TXN, mgrp,        "DB_ENV *%s /* In TokuFT, mgrp is a DB_ENV, not a DB_TXNMGR */");
@@ -581,11 +581,12 @@ static void print_db_txn_struct (void) {
 	"int (*txn_stat)(DB_TXN *, struct txn_stat **)", 
 	"int (*commit_with_progress)(DB_TXN*, uint32_t, TXN_PROGRESS_POLL_FUNCTION, void*)",
 	"int (*abort_with_progress)(DB_TXN*, TXN_PROGRESS_POLL_FUNCTION, void*)",
-	"int (*xa_prepare) (DB_TXN*, TOKU_XA_XID *)",
+	"int (*xa_prepare) (DB_TXN*, TOKU_XA_XID *, uint32_t flags)",
         "uint64_t (*id64) (DB_TXN*)",
         "void (*set_client_id)(DB_TXN *, uint64_t client_id)",
         "uint64_t (*get_client_id)(DB_TXN *)",
         "bool (*is_prepared)(DB_TXN *)",
+        "DB_TXN *(*get_child)(DB_TXN *)",
 	NULL};
     sort_and_dump_fields("db_txn", false, extra);
 }
@@ -614,7 +615,7 @@ static void print_dbc_struct (void) {
 	"int (*c_getf_set_range_reverse)(DBC *, uint32_t, DBT *, YDB_CALLBACK_FUNCTION, void *)",
 	"int (*c_getf_set_range_with_bound)(DBC *, uint32_t, DBT *k, DBT *k_bound, YDB_CALLBACK_FUNCTION, void *)",
 	"int (*c_set_bounds)(DBC*, const DBT*, const DBT*, bool pre_acquire, int out_of_range_error)",
-        "void (*c_set_check_interrupt_callback)(DBC*, bool (*)(void*), void *)",
+        "void (*c_set_check_interrupt_callback)(DBC*, bool (*)(void*, uint64_t deleted_rows), void *)",
 	"void (*c_remove_restriction)(DBC*)",
         "char _internal[512]",
 	NULL};

storage/tokudb/ft-index/ft/ft-internal.h

@@ -655,7 +655,7 @@ int toku_upgrade_msn_from_root_to_header(int fd, FT ft) __attribute__((nonnull))
 // When lock_only is true, the callback only does optional lock tree locking.
 typedef int (*FT_GET_CALLBACK_FUNCTION)(uint32_t keylen, const void *key, uint32_t vallen, const void *val, void *extra, bool lock_only);
 
-typedef bool (*FT_CHECK_INTERRUPT_CALLBACK)(void *extra);
+typedef bool (*FT_CHECK_INTERRUPT_CALLBACK)(void *extra, uint64_t deleted_rows);
 
 struct ft_cursor;
 int toku_ft_search(FT_HANDLE ft_handle, ft_search *search, FT_GET_CALLBACK_FUNCTION getf, void *getf_v, struct ft_cursor *ftcursor, bool can_bulk_fetch);

storage/tokudb/ft-index/ft/ft-ops.cc

@@ -3387,7 +3387,7 @@ ok: ;
                 idx++;
                 if (idx >= bn->data_buffer.num_klpairs() || ((n_deleted % 64) == 0 && !search_continue(search, key, keylen))) {
                     STATUS_INC(FT_CURSOR_SKIP_DELETED_LEAF_ENTRY, n_deleted);
-                    if (ftcursor->interrupt_cb && ftcursor->interrupt_cb(ftcursor->interrupt_cb_extra)) {
+                    if (ftcursor->interrupt_cb && ftcursor->interrupt_cb(ftcursor->interrupt_cb_extra, n_deleted)) {
                         return TOKUDB_INTERRUPTED;
                     }
                     return DB_NOTFOUND;
@@ -3396,7 +3396,7 @@ ok: ;
             case FT_SEARCH_RIGHT:
                 if (idx == 0) {
                     STATUS_INC(FT_CURSOR_SKIP_DELETED_LEAF_ENTRY, n_deleted);
-                    if (ftcursor->interrupt_cb && ftcursor->interrupt_cb(ftcursor->interrupt_cb_extra)) {
+                    if (ftcursor->interrupt_cb && ftcursor->interrupt_cb(ftcursor->interrupt_cb_extra, n_deleted)) {
                         return TOKUDB_INTERRUPTED;
                     }
                     return DB_NOTFOUND;
@@ -3410,6 +3410,8 @@ ok: ;
             assert_zero(r); // we just validated the index
             if (!le_val_is_del(le, ftcursor->is_snapshot_read, ftcursor->ttxn)) {
                 STATUS_INC(FT_CURSOR_SKIP_DELETED_LEAF_ENTRY, n_deleted);
+                if (ftcursor->interrupt_cb)
+                    ftcursor->interrupt_cb(ftcursor->interrupt_cb_extra, n_deleted);
                 goto got_a_good_value;
             }
         }

storage/tokudb/ft-index/ft/ft.cc

@@ -1064,6 +1064,11 @@ garbage_helper(BLOCKNUM blocknum, int64_t UU(size), int64_t UU(address), void *e
             goto exit;
         }
     }
+    {
+        float a = info->used_space, b=info->total_space;
+        float percentage = (1 - (a / b)) * 100;
+        printf("LeafNode# %d has %d BasementNodes and %2.1f%% of the allocated space is garbage\n", (int)blocknum.b, node->n_children, percentage);
+    }
 exit:
     toku_ftnode_free(&node);
     toku_free(ndd);

storage/tokudb/ft-index/ft/logger/recover.cc

@@ -714,7 +714,7 @@ static int toku_recover_xstillopenprepared (struct logtype_xstillopenprepared *l
     }
     switch (renv->ss.ss) {
         case FORWARD_BETWEEN_CHECKPOINT_BEGIN_END: {
-            toku_txn_prepare_txn(txn, l->xa_xid);
+            toku_txn_prepare_txn(txn, l->xa_xid, 0);
             break;
         }
         case FORWARD_NEWER_CHECKPOINT_END: {
@@ -778,7 +778,7 @@ static int toku_recover_xprepare (struct logtype_xprepare *l, RECOVER_ENV renv)
     assert(txn!=NULL);
 
     // Save the transaction
-    toku_txn_prepare_txn(txn, l->xa_xid);
+    toku_txn_prepare_txn(txn, l->xa_xid, 0);
 
     return 0;
 }

storage/tokudb/ft-index/ft/txn/txn.cc

@@ -558,7 +558,7 @@ static void copy_xid (TOKU_XA_XID *dest, TOKU_XA_XID *source) {
     memcpy(dest->data, source->data, source->gtrid_length+source->bqual_length);
 }
 
-void toku_txn_prepare_txn (TOKUTXN txn, TOKU_XA_XID *xa_xid) {
+void toku_txn_prepare_txn (TOKUTXN txn, TOKU_XA_XID *xa_xid, int nosync) {
     if (txn->parent || toku_txn_is_read_only(txn)) {
         // We do not prepare children.
         //
@@ -573,7 +573,7 @@ void toku_txn_prepare_txn (TOKUTXN txn, TOKU_XA_XID *xa_xid) {
     txn->state = TOKUTXN_PREPARING; 
     toku_txn_unlock_state(txn);
     // Do we need to do an fsync?
-    txn->do_fsync = (txn->force_fsync_on_commit || txn->roll_info.num_rollentries>0);
+    txn->do_fsync = txn->force_fsync_on_commit || (!nosync && txn->roll_info.num_rollentries>0);
     copy_xid(&txn->xa_xid, xa_xid);
     // This list will go away with #4683, so we wn't need the ydb lock for this anymore.
     toku_log_xprepare(txn->logger, &txn->do_fsync_lsn, 0, txn, txn->txnid, xa_xid);

storage/tokudb/ft-index/ft/txn/txn.h

@@ -302,7 +302,7 @@ int toku_txn_abort_with_lsn(struct tokutxn *txn, LSN oplsn,
 
 int toku_txn_discard_txn(struct tokutxn *txn);
 
-void toku_txn_prepare_txn (struct tokutxn *txn, TOKU_XA_XID *xid);
+void toku_txn_prepare_txn (struct tokutxn *txn, TOKU_XA_XID *xid, int nosync);
 // Effect: Do the internal work of preparing a transaction (does not log the prepare record).
 
 void toku_txn_get_prepared_xa_xid(struct tokutxn *txn, TOKU_XA_XID *xa_xid);

storage/tokudb/ft-index/src/tests/recover-5146.cc

@@ -147,7 +147,7 @@ run_test(void) {
         uint8_t gid[DB_GID_SIZE];
         memset(gid, 0, DB_GID_SIZE);
         gid[0]=42;
-        r = txn->prepare(txn, gid);                                                            CKERR(r);
+        r = txn->prepare(txn, gid, 0);                                                            CKERR(r);
     }
 
     r = env->txn_checkpoint(env, 0, 0, 0);                                              CKERR(r);

storage/tokudb/ft-index/src/tests/test-prepare.cc

@@ -122,7 +122,7 @@ static void setup_env_and_prepare (DB_ENV **envp, const char *envdir, bool commi
     uint8_t gid[DB_GID_SIZE];
     memset(gid, 0, DB_GID_SIZE);
     gid[0]=42;
-    CKERR(txn->prepare(txn, gid));
+    CKERR(txn->prepare(txn, gid, 0));
     if (commit)
 	CKERR(txn->commit(txn, 0));
 }

storage/tokudb/ft-index/src/tests/test-prepare2.cc

@@ -124,7 +124,7 @@ static void setup_env_and_prepare (DB_ENV **envp, const char *envdir, bool commi
     uint8_t gid[DB_GID_SIZE];
     memset(gid, 0, DB_GID_SIZE);
     gid[0]=42;
-    CKERR(txn->prepare(txn, gid));
+    CKERR(txn->prepare(txn, gid, 0));
     if (commit)
 	CKERR(txn->commit(txn, 0));
 }

storage/tokudb/ft-index/src/tests/test-prepare3.cc

@@ -147,7 +147,7 @@ static void setup_env_and_prepare (DB_ENV **envp, const char *envdir) {
 	uint8_t gid[DB_GID_SIZE];
 	memset(gid, 0, DB_GID_SIZE);
 	gid[0]='a'+tnum;
-	CKERR(txn->prepare(txn, gid));
+	CKERR(txn->prepare(txn, gid, 0));
 	// Drop txn on the ground, since we will commit or abort it after recovery
 	if (tnum==0) {
 	    //printf("commit %d\n", tnum);

storage/tokudb/ft-index/src/tests/test-xa-prepare.cc

@@ -125,7 +125,7 @@ static void setup_env_and_prepare (DB_ENV **envp, const char *envdir, bool commi
 		     .gtrid_length = 8,
 		     .bqual_length = 9};
     for (int i=0; i<8+9; i++) x.data[i] = 42+i;
-    CKERR(txn->xa_prepare(txn, &x));
+    CKERR(txn->xa_prepare(txn, &x, 0));
     if (commit)
 	CKERR(txn->commit(txn, 0));
 }

storage/tokudb/ft-index/src/tests/test5092.cc

@@ -118,7 +118,7 @@ static void setup_env_and_prepare (DB_ENV **envp, const char *envdir, bool commi
     uint8_t gid[DB_GID_SIZE];
     memset(gid, 0, DB_GID_SIZE);
     gid[0]=42;
-    CKERR(txn->prepare(txn, gid));
+    CKERR(txn->prepare(txn, gid, 0));
     { int chk_r = db->close(db, 0); CKERR(chk_r); }
     if (commit)
         CKERR(txn->commit(txn, 0));

storage/tokudb/ft-index/src/tests/test_5015.cc

@@ -133,7 +133,7 @@ test_main(int argc, char *const argv[]) {
     uint8_t gid[DB_GID_SIZE];
     memset(gid, 0, DB_GID_SIZE);
     gid[0]='a';
-    r = child_txn->prepare(child_txn, gid);
+    r = child_txn->prepare(child_txn, gid, 0);
     CKERR(r);
 
     r = env->txn_checkpoint(env, 0, 0, 0);

storage/tokudb/ft-index/src/tests/test_cursor_interrupt.cc

@@ -95,12 +95,12 @@ PATENT RIGHTS GRANT:
 
 
 int num_interrupts_called;
-static bool interrupt(void* extra UU()) {
+static bool interrupt(void* extra UU(), uint64_t rows UU()) {
     num_interrupts_called++;
     return false;
 }
 
-static bool interrupt_true(void* extra UU()) {
+static bool interrupt_true(void* extra UU(), uint64_t rows UU()) {
     num_interrupts_called++;
     return true;
 }

storage/tokudb/ft-index/src/tests/test_stress_hot_indexing.cc

@@ -211,7 +211,7 @@ cleanup:
     gid_count++;
     uint32_t *hi_gid_count_p = cast_to_typeof(hi_gid_count_p) hi_gid;  // make gcc --happy about -Wstrict-aliasing
     *hi_gid_count_p = gid_count;
-    int rr = hi_txn->prepare(hi_txn, hi_gid);
+    int rr = hi_txn->prepare(hi_txn, hi_gid, 0);
     CKERR(rr);
     if (r || (random() % 2)) {
         rr = hi_txn->abort(hi_txn);

storage/tokudb/ft-index/src/tests/test_txn_close_before_prepare_commit.cc

@@ -130,7 +130,7 @@ test_txn_close_before_prepare_commit (void) {
 
     uint8_t gid[DB_GID_SIZE];
     memset(gid, 1, DB_GID_SIZE);
-    r = txn->prepare(txn, gid);   assert(r == 0);
+    r = txn->prepare(txn, gid, 0);   assert(r == 0);
     r = txn->commit(txn, 0); assert(r == 0);
 
     r = env->close(env, 0); assert(r == 0);

storage/tokudb/ft-index/src/tests/threaded_stress_test_helpers.h

@@ -593,7 +593,7 @@ static void *worker(void *arg_v) {
             uint64_t gid_val = txn->id64(txn);
             uint64_t *gid_count_p = cast_to_typeof(gid_count_p) gid;  // make gcc --happy about -Wstrict-aliasing
             *gid_count_p = gid_val;
-            int rr = txn->prepare(txn, gid);
+            int rr = txn->prepare(txn, gid, 0);
             assert_zero(rr);
         }
         if (r == 0) {

storage/tokudb/ft-index/src/tests/xa-dirty-commit.cc

@@ -136,7 +136,7 @@ static void create_prepared_txn(void) {
     for (int i = 0; i < 8+9; i++) {
         xid.data[i] = i;
     }
-    r = txn->xa_prepare(txn, &xid);
+    r = txn->xa_prepare(txn, &xid, 0);
     CKERR(r);
 
     // discard the txn so that we can close the env and run xa recovery later

storage/tokudb/ft-index/src/tests/xa-dirty-rollback.cc

@@ -136,7 +136,7 @@ static void create_prepared_txn(void) {
     for (int i = 0; i < 8+9; i++) {
         xid.data[i] = i;
     }
-    r = txn->xa_prepare(txn, &xid);
+    r = txn->xa_prepare(txn, &xid, 0);
     CKERR(r);
 
     // discard the txn so that we can close the env and run xa recovery later

storage/tokudb/ft-index/src/tests/xa-txn-discard-abort.cc

@@ -138,7 +138,7 @@ static void create_prepared_txn(void) {
     for (int i = 0; i < 8+9; i++) {
         xid.data[i] = i;
     }
-    r = txn->xa_prepare(txn, &xid);
+    r = txn->xa_prepare(txn, &xid, 0);
     CKERR(r);
 
     // discard the txn so that we can close the env and run xa recovery later

storage/tokudb/ft-index/src/tests/xa-txn-discard-commit.cc

@@ -139,7 +139,7 @@ static void create_prepared_txn(void) {
     for (int i = 0; i < 8+9; i++) {
         xid.data[i] = i;
     }
-    r = txn->xa_prepare(txn, &xid);
+    r = txn->xa_prepare(txn, &xid, 0);
     CKERR(r);
 
     // discard the txn so that we can close the env and run xa recovery later

storage/tokudb/ft-index/src/ydb_cursor.cc

@@ -769,7 +769,7 @@ c_remove_restriction(DBC *dbc) {
 }
 
 static void
-c_set_check_interrupt_callback(DBC* dbc, bool (*interrupt_callback)(void*), void *extra) {
+c_set_check_interrupt_callback(DBC* dbc, bool (*interrupt_callback)(void*, uint64_t), void *extra) {
     toku_ft_cursor_set_check_interrupt_cb(dbc_ftcursor(dbc), interrupt_callback, extra);
 }
 

storage/tokudb/ft-index/src/ydb_txn.cc

@@ -240,7 +240,7 @@ static int toku_txn_abort(DB_TXN * txn,
     return r;
 }
 
-static int toku_txn_xa_prepare (DB_TXN *txn, TOKU_XA_XID *xid) {
+static int toku_txn_xa_prepare (DB_TXN *txn, TOKU_XA_XID *xid, uint32_t flags) {
     int r = 0;
     if (!txn) {
         r = EINVAL;
@@ -273,9 +273,11 @@ static int toku_txn_xa_prepare (DB_TXN *txn, TOKU_XA_XID *xid) {
         HANDLE_PANICKED_ENV(txn->mgrp);
     }
     assert(!db_txn_struct_i(txn)->child);
+    int nosync;
+    nosync = (flags & DB_TXN_NOSYNC)!=0 || (db_txn_struct_i(txn)->flags&DB_TXN_NOSYNC);
     TOKUTXN ttxn;
     ttxn = db_txn_struct_i(txn)->tokutxn;
-    toku_txn_prepare_txn(ttxn, xid);
+    toku_txn_prepare_txn(ttxn, xid, nosync);
     TOKULOGGER logger;
     logger = txn->mgrp->i->logger;
     LSN do_fsync_lsn;
@@ -292,14 +294,14 @@ exit:
 
 // requires: must hold the multi operation lock. it is
 //           released in toku_txn_xa_prepare before the fsync.
-static int toku_txn_prepare (DB_TXN *txn, uint8_t gid[DB_GID_SIZE]) {
+static int toku_txn_prepare (DB_TXN *txn, uint8_t gid[DB_GID_SIZE], uint32_t flags) {
     TOKU_XA_XID xid;
     TOKU_ANNOTATE_NEW_MEMORY(&xid, sizeof(xid));
     xid.formatID=0x756b6f54; // "Toku"
     xid.gtrid_length=DB_GID_SIZE/2;  // The maximum allowed gtrid length is 64.  See the XA spec in source:/import/opengroup.org/C193.pdf page 20.
     xid.bqual_length=DB_GID_SIZE/2; // The maximum allowed bqual length is 64.
     memcpy(xid.data, gid, DB_GID_SIZE);
-    return toku_txn_xa_prepare(txn, &xid);
+    return toku_txn_xa_prepare(txn, &xid, flags);
 }
 
 static int toku_txn_txn_stat (DB_TXN *txn, struct txn_stat **txn_stat) {
@@ -427,6 +429,10 @@ static bool toku_txn_is_prepared(DB_TXN *txn) {
     return toku_txn_get_state(ttxn) == TOKUTXN_PREPARING;
 }
 
+static DB_TXN *toku_txn_get_child(DB_TXN *txn) {
+    return db_txn_struct_i(txn)->child;
+}
+
 static inline void txn_func_init(DB_TXN *txn) {
 #define STXN(name) txn->name = locked_txn_ ## name
     STXN(abort);
@@ -444,6 +450,7 @@ static inline void txn_func_init(DB_TXN *txn) {
 #undef SUTXN
     txn->id64 = toku_txn_id64;
     txn->is_prepared = toku_txn_is_prepared;
+    txn->get_child = toku_txn_get_child;
 }
 
 //

storage/tokudb/ft-index/tools/tokuftdump.cc

@@ -99,7 +99,10 @@ PATENT RIGHTS GRANT:
 #include <stdlib.h>
 #include <inttypes.h>
 #include <limits.h>
-
+#include <string>
+#include <iostream>
+#include <fstream>
+#include <string.h>
 #include "ft/serialize/block_table.h"
 #include "ft/cachetable/cachetable.h"
 #include "ft/ft.h"
@@ -108,8 +111,11 @@ PATENT RIGHTS GRANT:
 #include "ft/serialize/ft_node-serialize.h"
 #include "ft/node.h"
 
+using namespace std;
+
 static int do_dump_data = 1;
 static int do_interactive = 0;
+static int do_json = 0;
 static int do_header = 0;
 static int do_fragmentation = 0;
 static int do_garbage = 0;
@@ -118,10 +124,24 @@ static int do_rootnode = 0;
 static int do_node = 0;
 static BLOCKNUM do_node_num;
 static int do_tsv = 0;
-
 static const char *arg0;
 static const char *fname;
 
+//it holdes the messges count for each FT's node
+typedef struct nodeMessage{
+    int id;
+    int clean;//0=clean >=1 dirty
+    int *count;//holds the messages
+    nodeMessage *nextNode;
+}NMC;
+enum { maxline = 128};
+
+static int printNodeMessagesToSTDout(NMC* ptr);
+
+static int printLevelSTDout(int  *);
+
+static void treeToSTDout(NMC *msgs[], int height);
+
 static void format_time(const uint64_t time_int, char *buf) {
     time_t timer = (time_t) time_int;
     ctime_r(&timer, buf);
@@ -225,12 +245,202 @@ static void dump_header(FT ft) {
     printf(" estimated numbytes=%" PRId64 "\n", ft->in_memory_stats.numbytes);
 }
 
+static int64_t getRootNode(FT ft) {
+    return ft->h->root_blocknum.b;
+}
+
 static int print_le(const void* key, const uint32_t keylen, const LEAFENTRY &le, const uint32_t idx UU(), void *const ai UU()) {
     print_klpair(stdout, key, keylen, le);
     printf("\n");
     return 0;
 }
 
+static int getHeight(int fd, BLOCKNUM blocknum, FT ft){
+    FTNODE n;
+    FTNODE_DISK_DATA ndd = nullptr;
+    ftnode_fetch_extra bfe;
+    bfe.create_for_full_read(ft);
+    int r = toku_deserialize_ftnode_from (fd, blocknum, 0 /*pass zero for hash, it doesn't matter*/, &n, &ndd, &bfe);
+    assert_zero(r);
+    assert(n!=0);
+    return n->height;
+}
+
+static FTNODE  getNode(int fd, BLOCKNUM blocknum, FT ft) {
+    FTNODE n;
+    FTNODE_DISK_DATA ndd = nullptr;
+    ftnode_fetch_extra bfe;
+    bfe.create_for_full_read(ft);
+    int r = toku_deserialize_ftnode_from (fd, blocknum, 0 /*pass zero for hash, it doesn't matter*/, &n, &ndd, &bfe);
+    assert_zero(r);;
+    return n;
+}
+
+static int countNodes(NMC *level){
+    int count=0;
+    NMC *ptr=level;
+    while(ptr!=NULL){
+        count++;
+        ptr=ptr->nextNode;
+    }
+    return count;
+}
+
+static int * countMessages(NMC *level){
+    int *counts=new int[16];
+    for(int i=0;i<16;i++){
+        counts[i]=0;
+    }
+    NMC *ptr=level;
+    while(ptr!=NULL){
+        for(int i=0;i<16;i++){
+            counts[i]+=ptr->count[i];
+        }
+        ptr=ptr->nextNode;
+    }
+    return counts;
+}
+
+static NMC * getLast(NMC *level){
+    if (level==NULL) return NULL;
+    NMC *ptr=level;
+    while(ptr->nextNode!=NULL){
+        ptr=ptr->nextNode;
+    }
+    return ptr;
+}
+
+/*
+ * Prints the total messages at each to STDout
+ */
+static int  printLevelSTDout(int *count){
+    int isEmpty=0;
+    for(int j=0;j<16;j++){
+        if(count[j]>0){
+            cout <<count[j]<<" ";
+            isEmpty++;
+            switch (j)   {
+                case FT_INSERT: printf("INSERT(s) "); break;
+                case FT_INSERT_NO_OVERWRITE: printf("INSERT_NO_OVERWRITE(s) "); break;
+                case FT_DELETE_ANY: printf("DELETE_ANY(s) "); break;
+                case FT_ABORT_ANY: printf("ABORT_ANY(s) "); break;
+                case FT_COMMIT_ANY: printf("COMMIT_ANY(s) "); break;
+                case FT_COMMIT_BROADCAST_ALL: printf("COMMIT_BROADCAST_ALL(s) "); break;
+                case FT_COMMIT_BROADCAST_TXN: printf("COMMIT_BROADCAST_TXN(s) "); break;
+                case FT_ABORT_BROADCAST_TXN: printf("ABORT_BROADCAST_TXN(s) "); break;
+                case FT_OPTIMIZE: printf("OPTIMIZE(s) "); break;
+                case FT_OPTIMIZE_FOR_UPGRADE: printf("OPTIMIZE_FOR_UPGRADE(s) "); break;
+                case FT_UPDATE:   printf("UPDATE(s) "); break;
+                case FT_UPDATE_BROADCAST_ALL: printf("UPDATE_BROADCAST_ALL(s) "); break;
+            }
+
+        }
+    }
+    return isEmpty;
+}
+
+/*
+ * Prints the total # of messages in a node  to STD output
+ */
+static int  printNodeMessagesToSTDout(NMC *ptr){
+    cout <<"\nNode :"<<ptr->id<<" has :";
+        for(int j=0;j<16;j++){
+        if(ptr->count[j]>0){
+            cout <<ptr->count[j]<<" ";
+            switch (j)   {
+                case FT_INSERT: printf("INSERT(s) "); break;
+                case FT_INSERT_NO_OVERWRITE: printf("INSERT_NO_OVERWRITE(s) "); break;
+                case FT_DELETE_ANY: printf("DELETE_ANY(s) "); break;
+                case FT_ABORT_ANY: printf("ABORT_ANY(s) "); break;
+                case FT_COMMIT_ANY: printf("COMMIT_ANY(s) "); break;
+                case FT_COMMIT_BROADCAST_ALL: printf("COMMIT_BROADCAST_ALL(s) "); break;
+                case FT_COMMIT_BROADCAST_TXN: printf("COMMIT_BROADCAST_TXN(s) "); break;
+                case FT_ABORT_BROADCAST_TXN: printf("ABORT_BROADCAST_TXN(s) "); break;
+                case FT_OPTIMIZE: printf("OPTIMIZE(s) "); break;
+                case FT_OPTIMIZE_FOR_UPGRADE: printf("OPTIMIZE_FOR_UPGRADE(s) "); break;
+                case FT_UPDATE:   printf("UPDATE(s) "); break;
+                case FT_UPDATE_BROADCAST_ALL: printf("UPDATE_BROADCAST_ALL(s) "); break;
+            }
+        }
+    }
+    return 1;
+}
+
+static void levelToSTDout(NMC *list, int level){
+    NMC *ptr=list;
+    cout <<endl<<"Height : "<<level<<endl;
+    while(ptr!=NULL){
+        if(ptr->clean!=0){
+            printNodeMessagesToSTDout(ptr);
+        }
+        else{
+            cout << "\nNode : "<<ptr->id<<" has no messages";
+        }
+        ptr=ptr->nextNode;
+    }
+    cout <<endl;
+}
+
+/*
+ * prints the tree total # of nodes and total # of  messages at each height in :
+ * STDout in human readable format
+ */
+static void treeToSTDout(NMC *msgs[], int height){
+    for(int i=height; i>=0 ; i--){
+        cout <<"At height "<<i;
+        int *counts=countMessages(msgs[i]);
+        cout <<"\n     Node Count: "<< countNodes(msgs[i])<<endl;
+        cout <<"           Messages: ";
+        if(printLevelSTDout(counts)==0) cout <<"0\n";
+        else cout <<endl;
+    }
+}
+
+//traverse through the FT and report back the count of messages in every node
+static void  countMessagesInFT(int fd, BLOCKNUM blocknum, FT ft,NMC *msgs[]){
+    FTNODE n=getNode(fd,blocknum,ft);
+
+    NMC *last=NULL;
+    if(msgs[n->height]==NULL){
+        last = msgs[n->height]=new NMC;
+    }else {
+        last=getLast(msgs[n->height]);
+        last->nextNode=new NMC;
+        last=last->nextNode;
+    }
+    last->id=blocknum.b;
+    last->count=new int[16];
+    for(int i=0;i<16;i++){
+        last->count[i]=0;
+    }
+    last->clean=0;
+    last->nextNode=NULL;
+
+    if (n->height==0){
+        toku_ftnode_free(&n);
+        return;
+    }
+    for(int i=0;i<n->n_children;i++){
+        NONLEAF_CHILDINFO bnc = BNC(n, i);
+        if (n->height==1 && n->bp[i].ptr.tag==BCT_NULL){
+            cout <<n->bp[i].ptr.tag;
+        }
+        auto dump_fn=[&](const ft_msg &msg, bool UU(is_fresh)) {
+            enum ft_msg_type type = (enum ft_msg_type) msg.type();
+            last->count[type]++;
+            last->clean=1;
+            return 0;
+        };
+
+        bnc->msg_buffer.iterate(dump_fn);
+
+        blocknum=make_blocknum(BP_BLOCKNUM(n, i).b);
+        countMessagesInFT(fd,blocknum,ft, msgs);
+    }
+
+    toku_ftnode_free(&n);
+}
+
 static void dump_node(int fd, BLOCKNUM blocknum, FT ft) {
     FTNODE n;
     FTNODE_DISK_DATA ndd = nullptr;
@@ -254,9 +464,9 @@ static void dump_node(int fd, BLOCKNUM blocknum, FT ft) {
     printf(" layout_version_read_from_disk=%d\n", n->layout_version_read_from_disk);
     printf(" build_id=%d\n", n->build_id);
     printf(" max_msn_applied_to_node_on_disk=%" PRId64 " (0x%" PRIx64 ")\n", n->max_msn_applied_to_node_on_disk.msn, n->max_msn_applied_to_node_on_disk.msn);
-    printf(" io time %lf decompress time %lf deserialize time %lf\n", 
-           tokutime_to_seconds(bfe.io_time), 
-           tokutime_to_seconds(bfe.decompress_time), 
+    printf(" io time %lf decompress time %lf deserialize time %lf\n",
+           tokutime_to_seconds(bfe.io_time),
+           tokutime_to_seconds(bfe.decompress_time),
            tokutime_to_seconds(bfe.deserialize_time));
 
     printf(" n_children=%d\n", n->n_children);
@@ -277,7 +487,7 @@ static void dump_node(int fd, BLOCKNUM blocknum, FT ft) {
         if (n->height > 0) {
             printf("%" PRId64 "\n", BP_BLOCKNUM(n, i).b);
             NONLEAF_CHILDINFO bnc = BNC(n, i);
-            unsigned int n_bytes = toku_bnc_nbytesinbuf(bnc); 
+            unsigned int n_bytes = toku_bnc_nbytesinbuf(bnc);
             int n_entries = toku_bnc_n_entries(bnc);
             if (n_bytes > 0 || n_entries > 0) {
                 printf("   buffer contains %u bytes (%d items)\n", n_bytes, n_entries);
@@ -402,8 +612,12 @@ static void dump_garbage_stats(int fd, FT ft) {
     uint64_t total_space = 0;
     uint64_t used_space = 0;
     toku_ft_get_garbage(ft, &total_space, &used_space);
-    printf("garbage total size\t%" PRIu64 "\n", total_space);
-    printf("garbage used size\t%" PRIu64 "\n", used_space);
+    printf("garbage total size :%20" PRIu64 "\n", total_space);
+    printf("garbage used size  :%20" PRIu64 "\n", used_space);
+    float a=used_space,b=total_space;
+
+    float percentage=((1-a/b)*100);
+    printf("Total garbage : %2.3f%%\n", percentage);
 }
 
 typedef struct __dump_node_extra {
@@ -438,7 +652,7 @@ static void sub_block_deserialize(struct dump_sub_block *sb, unsigned char *sub_
 static void verify_block(unsigned char *cp, uint64_t file_offset, uint64_t size) {
     // verify the header checksum
     const size_t node_header = 8 + sizeof (uint32_t) + sizeof (uint32_t) + sizeof (uint32_t);
-    
+
     printf("%.8s layout_version=%u %u build=%d\n", cp, get_unaligned_uint32(cp+8), get_unaligned_uint32(cp+12), get_unaligned_uint32(cp+16));
 
     unsigned char *sub_block_header = &cp[node_header];
@@ -544,7 +758,9 @@ static uint64_t getuint64(const char *f) {
 static void interactive_help(void) {
     fprintf(stderr, "help\n");
     fprintf(stderr, "header\n");
-    fprintf(stderr, "node NUMBER\n");
+    cout <<"mr/MessagesReport [NUMBER] \n      Reports messages for the level of the tree you want get more details about\n";
+    cout <<"rf/readFile ft-file-name \n      Switch to a different FT\n";
+    fprintf(stderr, "node NUMBER \n");
     fprintf(stderr, "bx OFFSET | block_translation OFFSET\n");
     fprintf(stderr, "dumpdata 0|1\n");
     fprintf(stderr, "fragmentation\n");
@@ -554,10 +770,160 @@ static void interactive_help(void) {
     fprintf(stderr, "quit\n");
 }
 
+static void freeNMC(NMC *msgs[], int height){
+    for(int i=0;i<height;i++){
+        if(msgs[i]!=NULL){
+            delete(msgs[i]->count);
+
+            while(msgs[i]->nextNode!=NULL){
+                NMC* ptr=msgs[i]->nextNode;
+                msgs[i]=msgs[i]->nextNode;
+                delete ptr;
+
+            }
+            msgs[i]=NULL;
+        }
+    }
+}
+
+static void writeTree(NMC *msgs[],int height,char *name UU()){
+    ofstream mytree ("/tmp/tree.txt",fstream::out);
+    if (mytree.is_open()){
+        for(int i=height;i>=0;i--){
+            NMC * ptr=msgs[i];
+            mytree <<i<<endl;
+            while(ptr!=NULL){
+                mytree << ptr->id<<"\t";
+                if(ptr->clean!=0)mytree << "1"<<"\t";
+                else mytree << "0"<<"\t";
+                for(int j=0;j<15;j++)mytree << ptr->count[j]<<" ";
+                mytree << ptr->count[i]<<endl;
+                ptr=ptr->nextNode;
+            }
+             mytree <<endl;
+        }
+    }
+    else cout << "Unable to open file";
+    mytree.close();
+}
+
+static void writeJson(NMC *msgs[],int height,const char *name){
+    ofstream mytree (name,fstream::out);
+    if (mytree.is_open()){
+        mytree <<"{\n \"FT\":[";
+        for(int i=height;i>=0;i--){
+            NMC * ptr=msgs[i];
+            mytree <<"{\n\"Level\": {\"Height\":\""<<i<<"\",\n \"Nodes\":[";
+            while(ptr!=NULL){
+                mytree <<"{\"ID\":\""<< ptr->id<<"\",";
+                if(ptr->clean!=0){
+                    mytree <<"\"Messages\":[";
+                    for(int j=0;j<16;j++)
+                        {
+                        mytree <<"{";
+                        switch (j)   {
+                            case FT_INSERT: mytree <<"\"INSERT\":\""<<ptr->count[j]<<"\""; break;
+                            case FT_INSERT_NO_OVERWRITE: mytree <<"\"INSERT_NOVERWTE\":\""<<ptr->count[j]<<"\""; break;
+                            case FT_DELETE_ANY: mytree <<"\"DELETE\":\""<<ptr->count[j]<<"\""; break;
+                            case FT_ABORT_ANY: mytree <<"\"ABORT\":\""<<ptr->count[j]<<"\""; break;
+                            case FT_COMMIT_ANY: mytree <<"\"COMMITY\":\""<<ptr->count[j]<<"\""; break;
+                            case FT_COMMIT_BROADCAST_ALL: mytree <<"\"COMMIT_BROADCAST_ALL\":\""<<ptr->count[j]<<"\"" ;    break;
+                            case FT_COMMIT_BROADCAST_TXN: mytree <<"\"COMMIT_BROADCAST_TXN\":\""<<ptr->count[j]<<"\""; break;
+                            case FT_ABORT_BROADCAST_TXN: mytree <<"\"ABORT_BROADCAST_TXN\":\""<<ptr->count[j]<<"\"";break;
+                            case FT_OPTIMIZE: mytree <<"\"OPTIMIZE\":\""<<ptr->count[j]<<"\""; break;
+                            case FT_OPTIMIZE_FOR_UPGRADE: mytree <<"\"OPTIMIZE_FOR_UPGRADE\":\""<<ptr->count[j]<<"\"";break;
+                            case FT_UPDATE:   mytree <<"\"UPDATE\":\""<<ptr->count[j]<<"\""; break;
+                            case FT_UPDATE_BROADCAST_ALL: mytree <<"\"UPDATE_BROADCAST_ALL\":\""<<ptr->count[j]<<"\""; break;
+                        }
+                        mytree <<"}";
+                        if(j<15)mytree<<",";
+                    }
+
+                    mytree <<"]}";
+
+                }
+                else {
+                    mytree <<"\"Messages\":\""<< "0"<<"\"}";
+                }
+                if(ptr->nextNode!=NULL)mytree <<",\n";
+                else mytree <<"]}\n";
+                ptr=ptr->nextNode;
+            }
+            mytree <<"\n}\n";
+            if(i!=0)mytree <<",\n";
+        }
+        mytree <<"\n]}\n";
+
+    }
+    else cout << "Unable to open file";
+    mytree.close();
+}
+
+static void writeTree(NMC *msgs[],int height){
+    ofstream mytree ("/tmp/tree1.txt",fstream::out);
+    if (mytree.is_open()){
+        for(int i=height;i>=0;i--){
+            NMC * ptr=msgs[i];
+            mytree <<i<<endl;
+            while(ptr!=NULL){
+                mytree << ptr->id<<",";
+                if(ptr->clean!=0)mytree << "1"<<",";
+                else mytree << "0"<<",";
+                for(int j=0;j<15;j++)mytree << ptr->count[j]<<",";
+                mytree << ptr->count[i]<<endl;
+                ptr=ptr->nextNode;
+            }
+             mytree <<".\"";
+        }
+    }
+    else cout << "Unable to open file";
+    mytree.close();
+}
+
+static void FT_to_JSON(int fd, FT ft, CACHEFILE cf, const char * JsonFile){
+    toku_ft_free(ft);
+    open_header(fd, &ft, cf);
+    int root=getRootNode(ft);
+    BLOCKNUM off = make_blocknum(root);
+    int height=getHeight(fd,off, ft);
+    NMC *msgs[height];
+    for(int i=0;i<=height;i++){
+        msgs[i]=NULL;
+    }
+    open_header(fd, &ft, cf);
+    root=getRootNode(ft);
+    off = make_blocknum(root);
+    countMessagesInFT(fd,off, ft,msgs);
+    cout <<"to STD output: \n";
+    treeToSTDout(msgs,height);
+    writeTree(msgs,height);
+    cout<<"FT's json file was generated here:";
+    if(JsonFile!=NULL)  {
+        cout <<JsonFile;
+        writeJson(msgs,height,JsonFile);
+    }
+    else {
+        cout <<"./FT.json";
+        writeJson(msgs,height,"./FT.json");
+    }
+    cout<<endl;
+    freeNMC(msgs,height);
+    exit(0);
+}
+
 static void run_iteractive_loop(int fd, FT ft, CACHEFILE cf) {
+    toku_ft_free(ft);
+    open_header(fd, &ft, cf);
+    int root=getRootNode(ft);
+    BLOCKNUM off = make_blocknum(root);
+    int height=getHeight(fd,off, ft);
+    NMC *msgs[height];
+    for(int i=0;i<=height;i++){
+        msgs[i]=NULL;
+    }
     while (1) {
-        printf("ftdump>"); fflush(stdout);
-        enum { maxline = 64};
+        printf("ftdump>");
+        fflush(stdout);
         char line[maxline+1];
         int r = readline(line, maxline);
         if (r == EOF)
@@ -565,23 +931,57 @@ static void run_iteractive_loop(int fd, FT ft, CACHEFILE cf) {
         const int maxfields = 4;
         char *fields[maxfields];
         int nfields = split_fields(line, fields, maxfields);
-        if (nfields == 0) 
+        if (nfields == 0)
             continue;
         if (strcmp(fields[0], "help") == 0) {
             interactive_help();
         } else if (strcmp(fields[0], "header") == 0) {
             toku_ft_free(ft);
             open_header(fd, &ft, cf);
-            dump_header(ft);
+        } else if (strcmp(fields[0], "rn") == 0||strcmp(fields[0], "rootNode")==0||strcmp(fields[0], "rootnode") == 0) {
+            printf("Root node :%d\n",root);
         } else if (strcmp(fields[0], "block") == 0 && nfields == 2) {
             BLOCKNUM blocknum = make_blocknum(getuint64(fields[1]));
             dump_block(fd, blocknum, ft);
+        }else if ((strcmp(fields[0], "readFile") == 0 ||strcmp(fields[0], "readfile") == 0 ||strcmp(fields[0], "rf") == 0 )&& nfields == 2) {
+            fname=fields[1];
+            fd = open(fname, O_RDWR + O_BINARY);
+            toku_ft_free(ft);
+            open_header(fd, &ft, cf);
+            root=getRootNode(ft);
+            off = make_blocknum(root);
+            height=getHeight(fd,off, ft);
+            if (fd < 0) {
+                fprintf(stderr, "%s: can not open the FT dump %s errno %d\n", arg0, fname, errno);
+                continue;
+            }
         } else if (strcmp(fields[0], "node") == 0 && nfields == 2) {
-            BLOCKNUM off = make_blocknum(getuint64(fields[1]));
+            off = make_blocknum(getuint64(fields[1]));
             dump_node(fd, off, ft);
-        } else if (strcmp(fields[0], "dumpdata") == 0 && nfields == 2) {
+        }else if ((strcmp(fields[0], "mr") == 0||(strcmp(fields[0], "nc")) == 0 ||strcmp(fields[0], "messagesReport") == 0 )) {
+            freeNMC(msgs,height);
+            toku_ft_free(ft);
+            open_header(fd, &ft, cf);
+            root=getRootNode(ft);
+            off = make_blocknum(root);
+            countMessagesInFT(fd,off, ft,msgs);
+            int level=-1;
+            if(nfields == 2)level=getuint64(fields[1]);
+            if(level>=0){
+                levelToSTDout(msgs[level], level);
+            }
+            else{
+                cout <<"to STD output: \n";
+                treeToSTDout(msgs,height);
+            }
+            writeTree(msgs,height);
+            writeTree(msgs,height, NULL);
+
+        }else if (strcmp(fields[0], "dumpdata") == 0 && nfields == 2) {
+
             do_dump_data = strtol(fields[1], NULL, 10);
-        } else if (strcmp(fields[0], "block_translation") == 0 || strcmp(fields[0], "bx") == 0) {
+        }
+        else if (strcmp(fields[0], "block_translation") == 0 || strcmp(fields[0], "bx") == 0) {
             uint64_t offset = 0;
             if (nfields == 2)
                 offset = getuint64(fields[1]);
@@ -590,7 +990,7 @@ static void run_iteractive_loop(int fd, FT ft, CACHEFILE cf) {
             dump_fragmentation(fd, ft, do_tsv);
         } else if (strcmp(fields[0], "nodesizes") == 0) {
             dump_nodesizes(fd, ft);
-        } else if (strcmp(fields[0], "garbage") == 0) {
+        } else if (strcmp(fields[0], "garbage") == 0||strcmp(fields[0], "g") == 0) {
             dump_garbage_stats(fd, ft);
         } else if (strcmp(fields[0], "file") == 0 && nfields >= 3) {
             uint64_t offset = getuint64(fields[1]);
@@ -604,14 +1004,18 @@ static void run_iteractive_loop(int fd, FT ft, CACHEFILE cf) {
             unsigned char newc = getuint64(fields[2]);
             set_file(fd, offset, newc);
         } else if (strcmp(fields[0], "quit") == 0 || strcmp(fields[0], "q") == 0) {
-            break;
+            toku_ft_free(ft);
+            exit(0);
         }
     }
+    freeNMC(msgs,height);
 }
 
 static int usage(void) {
     fprintf(stderr, "Usage: %s ", arg0);
     fprintf(stderr, "--interactive ");
+    fprintf(stderr, "--support /path/to/fractal-tree/file \n\t an interactive way to see what messages and/or switch between FTs");
+    fprintf(stderr, "--json /path/to/fractal-tree/file [output json file]\n\t if left empty an FT.json will be created automatically");
     fprintf(stderr, "--nodata ");
     fprintf(stderr, "--dumpdata 0|1 ");
     fprintf(stderr, "--header ");
@@ -632,10 +1036,15 @@ int main (int argc, const char *const argv[]) {
     while (argc>0) {
         if (strcmp(argv[0], "--interactive") == 0 || strcmp(argv[0], "--i") == 0) {
             do_interactive = 1;
+        }
+        else if ((strcmp(argv[0], "--json") == 0 || strcmp(argv[0], "--s")== 0)&& argc >= 2) {
+            do_json = 1;
+            fname=argv[1];
+            argc--; argv++;
+            break;
         } else if (strcmp(argv[0], "--nodata") == 0) {
             do_dump_data = 0;
         } else if (strcmp(argv[0], "--dumpdata") == 0 && argc > 1) {
-            argc--; argv++;
             do_dump_data = atoi(argv[0]);
         } else if (strcmp(argv[0], "--header") == 0) {
             do_header = 1;
@@ -660,39 +1069,39 @@ int main (int argc, const char *const argv[]) {
         }
         argc--; argv++;
     }
-    if (argc != 1) 
-        return usage();
+    if (argc != 1 && do_json==0)
+    return usage();
 
     int r = toku_ft_layer_init();
     assert_zero(r);
-
-    fname = argv[0];
+    if(fname==NULL)fname = argv[0];
     int fd = open(fname, O_RDWR + O_BINARY);
     if (fd < 0) {
         fprintf(stderr, "%s: can not open %s errno %d\n", arg0, fname, errno);
         return 1;
     }
-
     // create a cachefile for the header
     CACHETABLE ct = NULL;
     toku_cachetable_create(&ct, 1<<25, (LSN){0}, 0);
-
     CACHEFILE cf = NULL;
     r = toku_cachetable_openfd (&cf, ct, fd, fname);
     assert_zero(r);
-
     FT ft = NULL;
     open_header(fd, &ft, cf);
-
+    if (do_json ) {
+        const char *arg=argv[1];
+        FT_to_JSON(fd, ft, cf,arg);
+    }
     if (do_interactive) {
         run_iteractive_loop(fd, ft, cf);
-    } else {
+    }
+    else {
         if (do_header) {
             dump_header(ft);
         }
         if (do_rootnode) {
             dump_node(fd, ft->h->root_blocknum, ft);
-        } 
+        }
         if (do_node) {
             dump_node(fd, do_node_num, ft);
         }
@@ -708,14 +1117,12 @@ int main (int argc, const char *const argv[]) {
         if (!do_header && !do_rootnode && !do_fragmentation && !do_translation_table && !do_garbage) {
             printf("Block translation:");
             ft->blocktable.dump_translation_table(stdout);
-
             dump_header(ft);
-            
             struct __dump_node_extra info;
             info.fd = fd;
             info.ft = ft;
             ft->blocktable.iterate(block_table::TRANSLATION_CHECKPOINTED,
-                                   dump_node_wrapper, &info, true, true);
+            dump_node_wrapper, &info, true, true);
         }
     }
     toku_cachefile_close(&cf, false, ZERO_LSN);