[t:4184] #4184 Merge tokudb.4184c to main. Main now supports stat64 and keyrange64.

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

buildheader/db.h_4_1

@@ -206,8 +206,17 @@ typedef struct __toku_engine_status {
   uint64_t         disk_flush_nonleaf;          /* number of nonleaf nodes flushed to disk, not for checkpoint */
   uint64_t         disk_flush_leaf_for_checkpoint; /* number of leaf nodes flushed to disk for checkpoint */
   uint64_t         disk_flush_nonleaf_for_checkpoint; /* number of nonleaf nodes flushed to disk for checkpoint */
+  uint64_t         create_leaf;                 /* number of leaf nodes created */
+  uint64_t         create_nonleaf;              /* number of nonleaf nodes created */
   uint64_t         destroy_leaf;                /* number of leaf nodes destroyed */
   uint64_t         destroy_nonleaf;             /* number of nonleaf nodes destroyed */
+  uint64_t         split_leaf;                  /* number of leaf nodes split */
+  uint64_t         split_nonleaf;               /* number of nonleaf nodes split */
+  uint64_t         merge_leaf;                  /* number of times leaf nodes are merged */
+  uint64_t         merge_nonleaf;               /* number of times nonleaf nodes are merged  */
+  uint64_t         dirty_leaf;                  /* number of times leaf nodes are dirtied when previously clean */
+  uint64_t         dirty_nonleaf;               /* number of times nonleaf nodes are dirtied when previously clean */
+  uint64_t         balance_leaf;                /* number of times a leaf node is balanced inside brt */
   uint64_t         msg_bytes_in;                /* how many bytes of messages injected at root (for all trees)*/
   uint64_t         msg_bytes_out;               /* how many bytes of messages flushed from h1 nodes to leaves*/
   uint64_t         msg_bytes_curr;              /* how many bytes of messages currently in trees (estimate)*/

buildheader/db.h_4_3

@@ -206,8 +206,17 @@ typedef struct __toku_engine_status {
   uint64_t         disk_flush_nonleaf;          /* number of nonleaf nodes flushed to disk, not for checkpoint */
   uint64_t         disk_flush_leaf_for_checkpoint; /* number of leaf nodes flushed to disk for checkpoint */
   uint64_t         disk_flush_nonleaf_for_checkpoint; /* number of nonleaf nodes flushed to disk for checkpoint */
+  uint64_t         create_leaf;                 /* number of leaf nodes created */
+  uint64_t         create_nonleaf;              /* number of nonleaf nodes created */
   uint64_t         destroy_leaf;                /* number of leaf nodes destroyed */
   uint64_t         destroy_nonleaf;             /* number of nonleaf nodes destroyed */
+  uint64_t         split_leaf;                  /* number of leaf nodes split */
+  uint64_t         split_nonleaf;               /* number of nonleaf nodes split */
+  uint64_t         merge_leaf;                  /* number of times leaf nodes are merged */
+  uint64_t         merge_nonleaf;               /* number of times nonleaf nodes are merged  */
+  uint64_t         dirty_leaf;                  /* number of times leaf nodes are dirtied when previously clean */
+  uint64_t         dirty_nonleaf;               /* number of times nonleaf nodes are dirtied when previously clean */
+  uint64_t         balance_leaf;                /* number of times a leaf node is balanced inside brt */
   uint64_t         msg_bytes_in;                /* how many bytes of messages injected at root (for all trees)*/
   uint64_t         msg_bytes_out;               /* how many bytes of messages flushed from h1 nodes to leaves*/
   uint64_t         msg_bytes_curr;              /* how many bytes of messages currently in trees (estimate)*/

buildheader/db.h_4_4

@@ -206,8 +206,17 @@ typedef struct __toku_engine_status {
   uint64_t         disk_flush_nonleaf;          /* number of nonleaf nodes flushed to disk, not for checkpoint */
   uint64_t         disk_flush_leaf_for_checkpoint; /* number of leaf nodes flushed to disk for checkpoint */
   uint64_t         disk_flush_nonleaf_for_checkpoint; /* number of nonleaf nodes flushed to disk for checkpoint */
+  uint64_t         create_leaf;                 /* number of leaf nodes created */
+  uint64_t         create_nonleaf;              /* number of nonleaf nodes created */
   uint64_t         destroy_leaf;                /* number of leaf nodes destroyed */
   uint64_t         destroy_nonleaf;             /* number of nonleaf nodes destroyed */
+  uint64_t         split_leaf;                  /* number of leaf nodes split */
+  uint64_t         split_nonleaf;               /* number of nonleaf nodes split */
+  uint64_t         merge_leaf;                  /* number of times leaf nodes are merged */
+  uint64_t         merge_nonleaf;               /* number of times nonleaf nodes are merged  */
+  uint64_t         dirty_leaf;                  /* number of times leaf nodes are dirtied when previously clean */
+  uint64_t         dirty_nonleaf;               /* number of times nonleaf nodes are dirtied when previously clean */
+  uint64_t         balance_leaf;                /* number of times a leaf node is balanced inside brt */
   uint64_t         msg_bytes_in;                /* how many bytes of messages injected at root (for all trees)*/
   uint64_t         msg_bytes_out;               /* how many bytes of messages flushed from h1 nodes to leaves*/
   uint64_t         msg_bytes_curr;              /* how many bytes of messages currently in trees (estimate)*/

buildheader/db.h_4_5

@@ -206,8 +206,17 @@ typedef struct __toku_engine_status {
   uint64_t         disk_flush_nonleaf;          /* number of nonleaf nodes flushed to disk, not for checkpoint */
   uint64_t         disk_flush_leaf_for_checkpoint; /* number of leaf nodes flushed to disk for checkpoint */
   uint64_t         disk_flush_nonleaf_for_checkpoint; /* number of nonleaf nodes flushed to disk for checkpoint */
+  uint64_t         create_leaf;                 /* number of leaf nodes created */
+  uint64_t         create_nonleaf;              /* number of nonleaf nodes created */
   uint64_t         destroy_leaf;                /* number of leaf nodes destroyed */
   uint64_t         destroy_nonleaf;             /* number of nonleaf nodes destroyed */
+  uint64_t         split_leaf;                  /* number of leaf nodes split */
+  uint64_t         split_nonleaf;               /* number of nonleaf nodes split */
+  uint64_t         merge_leaf;                  /* number of times leaf nodes are merged */
+  uint64_t         merge_nonleaf;               /* number of times nonleaf nodes are merged  */
+  uint64_t         dirty_leaf;                  /* number of times leaf nodes are dirtied when previously clean */
+  uint64_t         dirty_nonleaf;               /* number of times nonleaf nodes are dirtied when previously clean */
+  uint64_t         balance_leaf;                /* number of times a leaf node is balanced inside brt */
   uint64_t         msg_bytes_in;                /* how many bytes of messages injected at root (for all trees)*/
   uint64_t         msg_bytes_out;               /* how many bytes of messages flushed from h1 nodes to leaves*/
   uint64_t         msg_bytes_curr;              /* how many bytes of messages currently in trees (estimate)*/

buildheader/db.h_4_6

@@ -206,8 +206,17 @@ typedef struct __toku_engine_status {
   uint64_t         disk_flush_nonleaf;          /* number of nonleaf nodes flushed to disk, not for checkpoint */
   uint64_t         disk_flush_leaf_for_checkpoint; /* number of leaf nodes flushed to disk for checkpoint */
   uint64_t         disk_flush_nonleaf_for_checkpoint; /* number of nonleaf nodes flushed to disk for checkpoint */
+  uint64_t         create_leaf;                 /* number of leaf nodes created */
+  uint64_t         create_nonleaf;              /* number of nonleaf nodes created */
   uint64_t         destroy_leaf;                /* number of leaf nodes destroyed */
   uint64_t         destroy_nonleaf;             /* number of nonleaf nodes destroyed */
+  uint64_t         split_leaf;                  /* number of leaf nodes split */
+  uint64_t         split_nonleaf;               /* number of nonleaf nodes split */
+  uint64_t         merge_leaf;                  /* number of times leaf nodes are merged */
+  uint64_t         merge_nonleaf;               /* number of times nonleaf nodes are merged  */
+  uint64_t         dirty_leaf;                  /* number of times leaf nodes are dirtied when previously clean */
+  uint64_t         dirty_nonleaf;               /* number of times nonleaf nodes are dirtied when previously clean */
+  uint64_t         balance_leaf;                /* number of times a leaf node is balanced inside brt */
   uint64_t         msg_bytes_in;                /* how many bytes of messages injected at root (for all trees)*/
   uint64_t         msg_bytes_out;               /* how many bytes of messages flushed from h1 nodes to leaves*/
   uint64_t         msg_bytes_curr;              /* how many bytes of messages currently in trees (estimate)*/

buildheader/make_db_h.c

@@ -599,9 +599,17 @@ int main (int argc __attribute__((__unused__)), char *const argv[] __attribute__
     printf("  uint64_t         disk_flush_nonleaf;          /* number of nonleaf nodes flushed to disk, not for checkpoint */\n");
     printf("  uint64_t         disk_flush_leaf_for_checkpoint; /* number of leaf nodes flushed to disk for checkpoint */\n");
     printf("  uint64_t         disk_flush_nonleaf_for_checkpoint; /* number of nonleaf nodes flushed to disk for checkpoint */\n");
+    printf("  uint64_t         create_leaf;                 /* number of leaf nodes created */\n");
+    printf("  uint64_t         create_nonleaf;              /* number of nonleaf nodes created */\n");
     printf("  uint64_t         destroy_leaf;                /* number of leaf nodes destroyed */\n");
     printf("  uint64_t         destroy_nonleaf;             /* number of nonleaf nodes destroyed */\n");
-
+    printf("  uint64_t         split_leaf;                  /* number of leaf nodes split */\n");
+    printf("  uint64_t         split_nonleaf;               /* number of nonleaf nodes split */\n");
+    printf("  uint64_t         merge_leaf;                  /* number of times leaf nodes are merged */\n");
+    printf("  uint64_t         merge_nonleaf;               /* number of times nonleaf nodes are merged  */\n");
+    printf("  uint64_t         dirty_leaf;                  /* number of times leaf nodes are dirtied when previously clean */\n");
+    printf("  uint64_t         dirty_nonleaf;               /* number of times nonleaf nodes are dirtied when previously clean */\n");
+    printf("  uint64_t         balance_leaf;                /* number of times a leaf node is balanced inside brt */\n");
     printf("  uint64_t         msg_bytes_in;                /* how many bytes of messages injected at root (for all trees)*/\n");
     printf("  uint64_t         msg_bytes_out;               /* how many bytes of messages flushed from h1 nodes to leaves*/\n");
     printf("  uint64_t         msg_bytes_curr;              /* how many bytes of messages currently in trees (estimate)*/\n");

buildheader/tdb.h

@@ -206,8 +206,17 @@ typedef struct __toku_engine_status {
   uint64_t         disk_flush_nonleaf;          /* number of nonleaf nodes flushed to disk, not for checkpoint */
   uint64_t         disk_flush_leaf_for_checkpoint; /* number of leaf nodes flushed to disk for checkpoint */
   uint64_t         disk_flush_nonleaf_for_checkpoint; /* number of nonleaf nodes flushed to disk for checkpoint */
+  uint64_t         create_leaf;                 /* number of leaf nodes created */
+  uint64_t         create_nonleaf;              /* number of nonleaf nodes created */
   uint64_t         destroy_leaf;                /* number of leaf nodes destroyed */
   uint64_t         destroy_nonleaf;             /* number of nonleaf nodes destroyed */
+  uint64_t         split_leaf;                  /* number of leaf nodes split */
+  uint64_t         split_nonleaf;               /* number of nonleaf nodes split */
+  uint64_t         merge_leaf;                  /* number of times leaf nodes are merged */
+  uint64_t         merge_nonleaf;               /* number of times nonleaf nodes are merged  */
+  uint64_t         dirty_leaf;                  /* number of times leaf nodes are dirtied when previously clean */
+  uint64_t         dirty_nonleaf;               /* number of times nonleaf nodes are dirtied when previously clean */
+  uint64_t         balance_leaf;                /* number of times a leaf node is balanced inside brt */
   uint64_t         msg_bytes_in;                /* how many bytes of messages injected at root (for all trees)*/
   uint64_t         msg_bytes_out;               /* how many bytes of messages flushed from h1 nodes to leaves*/
   uint64_t         msg_bytes_curr;              /* how many bytes of messages currently in trees (estimate)*/

db-benchmark-test/db-verify.c

@@ -20,12 +20,16 @@ int test_main(int argc, char * const argv[]) {
             verbose++;
             continue;
         }
+        if (strcmp(argv[i], "-q") == 0) {
+            if (verbose > 0)
+                verbose--;
+            continue;
+        }
         if (strcmp(argv[i], "-x") == 0) {
             do_txns = true;
             continue;
         }
-    }
-   
+    }   
 
     DB_ENV *env = NULL;
     r = db_env_create(&env, 0); 
@@ -41,6 +45,12 @@ int test_main(int argc, char * const argv[]) {
     r = db->open(db, NULL, dbfilename, NULL, DB_BTREE, DB_AUTO_COMMIT, S_IRUSR|S_IWUSR|S_IRGRP|S_IROTH);
     assert(r == 0);
 
+    if (verbose) {
+        DB_BTREE_STAT64 s;
+        r = db->stat64(db, NULL, &s); assert(r == 0);
+        printf("nkeys=%" PRIu64" dsize=%" PRIu64 "\n", s.bt_nkeys, s.bt_dsize);
+    }
+
     r = db->verify_with_progress(db, NULL, NULL, verbose > 0, false);
     assert(r == 0);
 

include/db.h

@@ -206,8 +206,17 @@ typedef struct __toku_engine_status {
   uint64_t         disk_flush_nonleaf;          /* number of nonleaf nodes flushed to disk, not for checkpoint */
   uint64_t         disk_flush_leaf_for_checkpoint; /* number of leaf nodes flushed to disk for checkpoint */
   uint64_t         disk_flush_nonleaf_for_checkpoint; /* number of nonleaf nodes flushed to disk for checkpoint */
+  uint64_t         create_leaf;                 /* number of leaf nodes created */
+  uint64_t         create_nonleaf;              /* number of nonleaf nodes created */
   uint64_t         destroy_leaf;                /* number of leaf nodes destroyed */
   uint64_t         destroy_nonleaf;             /* number of nonleaf nodes destroyed */
+  uint64_t         split_leaf;                  /* number of leaf nodes split */
+  uint64_t         split_nonleaf;               /* number of nonleaf nodes split */
+  uint64_t         merge_leaf;                  /* number of times leaf nodes are merged */
+  uint64_t         merge_nonleaf;               /* number of times nonleaf nodes are merged  */
+  uint64_t         dirty_leaf;                  /* number of times leaf nodes are dirtied when previously clean */
+  uint64_t         dirty_nonleaf;               /* number of times nonleaf nodes are dirtied when previously clean */
+  uint64_t         balance_leaf;                /* number of times a leaf node is balanced inside brt */
   uint64_t         msg_bytes_in;                /* how many bytes of messages injected at root (for all trees)*/
   uint64_t         msg_bytes_out;               /* how many bytes of messages flushed from h1 nodes to leaves*/
   uint64_t         msg_bytes_curr;              /* how many bytes of messages currently in trees (estimate)*/

newbrt/brt-cachetable-wrappers.c

@@ -99,7 +99,8 @@ create_new_brtnode_with_dep_nodes(
         n_children,
         h->layout_version,
         h->nodesize,
-        h->flags);
+        h->flags, 
+        h);
 
     assert((*result)->nodesize > 0);
     (*result)->fullhash = fullhash;

newbrt/brt-flusher.h

@@ -68,7 +68,8 @@ brtleaf_split(
     DBT *splitk,
     BOOL create_new_node,
     u_int32_t num_dependent_nodes,
-    BRTNODE* dependent_nodes
+    BRTNODE* dependent_nodes,
+    BRT_STATUS brt_status
     );
 
 /**
@@ -88,7 +89,8 @@ brt_nonleaf_split(
     BRTNODE *nodeb,
     DBT *splitk,
     u_int32_t num_dependent_nodes,
-    BRTNODE* dependent_nodes
+    BRTNODE* dependent_nodes,
+    BRT_STATUS brt_status
     );
 
 C_END

newbrt/brt-internal.h

@@ -51,6 +51,13 @@ enum { BUFFER_HEADER_SIZE = (4 // height//
 			     + TREE_FANOUT * 8 // children
 			     ) };
 
+typedef struct {
+    int64_t numrows;           // delta versions in basements could be negative
+    int64_t numbytes;
+} STAT64INFO_S, *STAT64INFO;
+
+static const STAT64INFO_S ZEROSTATS = {0,0};
+
 //
 // Field in brtnode_fetch_extra that tells the 
 // partial fetch callback what piece of the node
@@ -155,6 +162,7 @@ struct brtnode_leaf_basement_node {
     unsigned int seqinsert;         // number of sequential inserts to this leaf 
     MSN max_msn_applied;            // max message sequence number applied
     bool stale_ancestor_messages_applied;
+    STAT64INFO_S stat64_delta;      // change in stat64 counters since basement was last written to disk
 };
 
 enum  __attribute__((__packed__)) pt_state {  // declare this to be packed so that when used below it will only take 1 byte.
@@ -223,6 +231,7 @@ struct   __attribute__((__packed__)) brtnode_partition {
 
 struct brtnode {
     MSN      max_msn_applied_to_node_on_disk; // max_msn_applied that will be written to disk
+    struct brt_header *h;    // in-memory only
     unsigned int nodesize;
     unsigned int flags;
     BLOCKNUM thisnodename;   // Which block number is this node?
@@ -392,6 +401,9 @@ struct brt_header {
 
     brt_compare_func compare_fun;
     brt_update_func update_fun;
+    STAT64INFO_S in_memory_stats;
+    STAT64INFO_S on_disk_stats;
+    STAT64INFO_S checkpoint_staging_stats;
 };
 
 struct brt {
@@ -447,6 +459,10 @@ int toku_keycompare (bytevec key1, ITEMLEN key1len, bytevec key2, ITEMLEN key2le
 
 void toku_verify_or_set_counts(BRTNODE);
 
+void 
+toku_brt_header_init(struct brt_header *h,
+                     BLOCKNUM root_blocknum_on_disk, LSN checkpoint_lsn, TXNID root_xid_that_created, uint32_t target_nodesize, uint32_t target_basementnodesize);
+
 int toku_serialize_brt_header_size (struct brt_header *h);
 int toku_serialize_brt_header_to (int fd, struct brt_header *h);
 int toku_serialize_brt_header_to_wbuf (struct wbuf *, struct brt_header *h, int64_t address_translation, int64_t size_translation);
@@ -469,6 +485,14 @@ void toku_brt_nonleaf_append_child(BRTNODE node, BRTNODE child, struct kv_pair *
 // append a cmd to a nonleaf node child buffer
 void toku_brt_append_to_child_buffer(brt_compare_func compare_fun, DESCRIPTOR desc, BRTNODE node, int childnum, int type, MSN msn, XIDS xids, bool is_fresh, const DBT *key, const DBT *val);
 
+// Mark a node as dirty and update statistics in header.
+// Other than the node's constructor, this should be the ONLY place
+// a brt node is marked as dirty.
+void toku_mark_node_dirty(BRTNODE node);
+
+STAT64INFO_S toku_get_and_clear_basement_stats(BRTNODE leafnode);
+
+
 #if 1
 #define DEADBEEF ((void*)0xDEADBEEF)
 #else
@@ -545,6 +569,7 @@ struct brt_cursor {
 // is required, such as for flushes.
 //
 static inline void fill_bfe_for_full_read(struct brtnode_fetch_extra *bfe, struct brt_header *h) {
+    invariant(h->type == BRTHEADER_CURRENT);
     bfe->type = brtnode_fetch_all;
     bfe->h = h;
     bfe->search = NULL;
@@ -573,6 +598,7 @@ static inline void fill_bfe_for_subset_read(
     BOOL disable_prefetching
     )
 {
+    invariant(h->type == BRTHEADER_CURRENT);
     bfe->type = brtnode_fetch_subset;
     bfe->h = h;
     bfe->search = search;
@@ -591,6 +617,7 @@ static inline void fill_bfe_for_subset_read(
 // Currently used for stat64.
 //
 static inline void fill_bfe_for_min_read(struct brtnode_fetch_extra *bfe, struct brt_header *h) {
+    invariant(h->type == BRTHEADER_CURRENT);
     bfe->type = brtnode_fetch_none;
     bfe->h = h;
     bfe->search = NULL;
@@ -622,6 +649,7 @@ static inline void destroy_bfe_for_prefetch(struct brtnode_fetch_extra *bfe) {
 static inline void fill_bfe_for_prefetch(struct brtnode_fetch_extra *bfe,
                                          struct brt_header *h,
                                          BRT_CURSOR c) {
+    invariant(h->type == BRTHEADER_CURRENT);
     bfe->type = brtnode_fetch_prefetch;
     bfe->h = h;
     bfe->search = NULL;
@@ -682,7 +710,7 @@ void toku_create_new_brtnode (BRT t, BRTNODE *result, int height, int n_children
 
 // Effect: Fill in N as an empty brtnode.
 void toku_initialize_empty_brtnode (BRTNODE n, BLOCKNUM nodename, int height, int num_children, 
-                                    int layout_version, unsigned int nodesize, unsigned int flags);
+                                    int layout_version, unsigned int nodesize, unsigned int flags, struct brt_header *h);
 
 unsigned int toku_brtnode_which_child(BRTNODE node, const DBT *k,
                                       DESCRIPTOR desc, brt_compare_func cmp)
@@ -796,8 +824,17 @@ struct brt_status {
     uint64_t  disk_flush_nonleaf;          // number of nonleaf nodes flushed to disk, not for checkpoint
     uint64_t  disk_flush_leaf_for_checkpoint; // number of leaf nodes flushed to disk for checkpoint
     uint64_t  disk_flush_nonleaf_for_checkpoint; // number of nonleaf nodes flushed to disk for checkpoint
+    uint64_t  create_leaf;                 // number of leaf nodes created
+    uint64_t  create_nonleaf;              // number of nonleaf nodes created
     uint64_t  destroy_leaf;                // number of leaf nodes destroyed
     uint64_t  destroy_nonleaf;             // number of nonleaf nodes destroyed
+    uint64_t  split_leaf;                  // number of leaf nodes split
+    uint64_t  split_nonleaf;               // number of nonleaf nodes split
+    uint64_t  merge_leaf;                  // number of times leaf nodes are merged
+    uint64_t  merge_nonleaf;               // number of times nonleaf nodes are merged    
+    uint64_t  dirty_leaf;                  // number of times leaf nodes are dirtied when previously clean
+    uint64_t  dirty_nonleaf;               // number of times nonleaf nodes are dirtied when previously clean    
+    uint64_t  balance_leaf;                // number of times a leaf node is balanced inside brt
     uint64_t  msg_bytes_in;                // how many bytes of messages injected at root (for all trees)
     uint64_t  msg_bytes_out;               // how many bytes of messages flushed from h1 nodes to leaves
     uint64_t  msg_bytes_curr;              // how many bytes of messages currently in trees (estimate)
@@ -828,10 +865,11 @@ struct brt_status {
 void toku_brt_get_status(BRT_STATUS);
 
 void
-brtleaf_split (struct brt_header* h, BRTNODE node, BRTNODE *nodea, BRTNODE *nodeb, DBT *splitk, BOOL create_new_node, u_int32_t num_dependent_nodes, BRTNODE* dependent_nodes);
+brtleaf_split (struct brt_header* h, BRTNODE node, BRTNODE *nodea, BRTNODE *nodeb, DBT *splitk, BOOL create_new_node, u_int32_t num_dependent_nodes, BRTNODE* dependent_nodes, BRT_STATUS brt_status);
 
 void
 brt_leaf_apply_cmd_once (
+    BRTNODE leafnode,
     BASEMENTNODE bn,
     const BRT_MSG cmd,
     u_int32_t idx,
@@ -846,6 +884,7 @@ brt_leaf_put_cmd (
     brt_compare_func compare_fun,
     brt_update_func update_fun,
     DESCRIPTOR desc,
+    BRTNODE leafnode,
     BASEMENTNODE bn, 
     BRT_MSG cmd, 
     bool* made_change,

newbrt/brt-serialize.c

@@ -1025,6 +1025,7 @@ BASEMENTNODE toku_create_empty_bn_no_buffer(void) {
     bn->seqinsert = 0;
     bn->stale_ancestor_messages_applied = false;
     toku_mempool_zero(&bn->buffer_mempool);
+    bn->stat64_delta = ZEROSTATS;
     return bn;
 }
 
@@ -1814,6 +1815,9 @@ serialize_brt_header_min_size (u_int32_t version) {
 
 
     switch(version) {
+        case BRT_LAYOUT_VERSION_17:
+	    size += 16;
+	    invariant(sizeof(STAT64INFO_S) == 16);
         case BRT_LAYOUT_VERSION_16:
         case BRT_LAYOUT_VERSION_15:
             size += 4;  // basement node size
@@ -1885,6 +1889,8 @@ int toku_serialize_brt_header_to_wbuf (struct wbuf *wbuf, struct brt_header *h,
     wbuf_TXNID(wbuf, h->root_xid_that_created);
     wbuf_int(wbuf, h->basementnodesize);
     wbuf_ulonglong(wbuf, h->time_of_last_verification);
+    wbuf_ulonglong(wbuf, h->checkpoint_staging_stats.numrows);
+    wbuf_ulonglong(wbuf, h->checkpoint_staging_stats.numbytes);
     u_int32_t checksum = x1764_finish(&wbuf->checksum);
     wbuf_int(wbuf, checksum);
     lazy_assert(wbuf->ndone == wbuf->size);
@@ -2155,6 +2161,12 @@ deserialize_brtheader (int fd, struct rbuf *rb, struct brt_header **brth) {
         h->basementnodesize = rbuf_int(&rc);
         h->time_of_last_verification = rbuf_ulonglong(&rc);
     }
+    if (h->layout_version >= BRT_LAYOUT_VERSION_17) {
+	h->on_disk_stats.numrows = rbuf_ulonglong(&rc);
+	h->on_disk_stats.numbytes = rbuf_ulonglong(&rc);
+	h->in_memory_stats = h->on_disk_stats;
+    }
+
     (void)rbuf_int(&rc); //Read in checksum and ignore (already verified).
     if (rc.ndone!=rc.size) {ret = EINVAL; goto died1;}
     toku_free(rc.buf);
@@ -2207,8 +2219,9 @@ deserialize_brtheader_versioned (int fd, struct rbuf *rb, struct brt_header **br
             case BRT_LAYOUT_VERSION_14:
                 h->basementnodesize = 128*1024;  // basement nodes added in v15
                 //fall through on purpose
-            case BRT_LAYOUT_VERSION_16:
-            case BRT_LAYOUT_VERSION_15: // this may no properly support version 15, we'll fix that on upgrade.
+            case BRT_LAYOUT_VERSION_17:
+            case BRT_LAYOUT_VERSION_16: // version 16 never released to customers
+            case BRT_LAYOUT_VERSION_15: // this will not properly support version 15, we'll fix that on upgrade.
                 invariant(h->layout_version == BRT_LAYOUT_VERSION);
                 h->upgrade_brt_performed = FALSE;
                 if (upgrade) {

newbrt/brt-test-helpers.c

+/* -*- mode: C; c-basic-offset: 4 -*- */
 #ident "$Id$"
 #ident "Copyright (c) 2007-2010 Tokutek Inc.  All rights reserved."
 #ident "The technology is licensed by the Massachusetts Institute of Technology, Rutgers State University of New Jersey, and the Research Foundation of State University of New York at Stony Brook under United States of America Serial No. 11/760379 and to the patents and/or patent applications resulting from it."
@@ -148,11 +149,14 @@ int toku_testsetup_insert_to_leaf (BRT brt, BLOCKNUM blocknum, char *key, int ke
     BASEMENTNODE bn = BLB(node, childnum);
     void * maybe_free = 0;
     
-    r = apply_msg_to_leafentry(&cmd, NULL, //No old leafentry
-                               &newlesize, &leafentry, 
-			       bn->buffer, &bn->buffer_mempool, &maybe_free,
-                               NULL, NULL);
-    assert(r==0);
+    {
+	int64_t ignoreme;
+	r = apply_msg_to_leafentry(&cmd, NULL, //No old leafentry
+				   &newlesize, &leafentry, 
+				   bn->buffer, &bn->buffer_mempool, &maybe_free,
+				   NULL, NULL, &ignoreme);
+	assert(r==0);
+    }
 
 
     struct cmd_leafval_heaviside_extra be = {brt->compare_fun, &brt->h->descriptor, &keydbt};

newbrt/brt.h

@@ -233,7 +233,9 @@ enum brt_header_flags {
     TOKU_DB_VALCMP_BUILTIN_13  = (1<<3),
 };
 
-int toku_brt_keyrange (BRT brt, DBT *key, u_int64_t *less,  u_int64_t *equal,  u_int64_t *greater)  __attribute__ ((warn_unused_result));
+int 
+toku_brt_keyrange (BRT brt, DBT *key, u_int64_t *less,  u_int64_t *equal,  u_int64_t *greater) __attribute__ ((warn_unused_result));
+
 struct brtstat64_s {
     u_int64_t nkeys; /* estimate how many unique keys (even when flattened this may be an estimate)     */
     u_int64_t ndata; /* estimate the number of pairs (exact when flattened and committed)               */
@@ -245,10 +247,9 @@ struct brtstat64_s {
     u_int64_t modify_time_sec; /* time of last serialization, in seconds. */ 
     u_int64_t verify_time_sec; /* time of last verification, in seconds */
 };
-int toku_brt_stat64 (BRT, TOKUTXN,
-		     struct brtstat64_s *stat
-		     )
-    __attribute__ ((warn_unused_result));
+
+int 
+toku_brt_stat64 (BRT, TOKUTXN, struct brtstat64_s *stat) __attribute__ ((warn_unused_result));
 
 int toku_brt_init(void (*ydb_lock_callback)(void),
                   void (*ydb_unlock_callback)(void),

newbrt/brt_layout_version.h

@@ -20,6 +20,8 @@ enum brt_layout_version_e {
     BRT_LAYOUT_VERSION_14 = 14, // Diff from 13 to 14: Added MVCC; deprecated TOKU_DB_VALCMP_BUILTIN(_13); Remove fingerprints; Support QUICKLZ; add end-to-end checksum on uncompressed data.
     BRT_LAYOUT_VERSION_15 = 15, // Diff from 14 to 15: basement nodes, last verification time
     BRT_LAYOUT_VERSION_16 = 16, // Dr. No:  No subtree estimates, partition layout information represented more transparently. 
+                                // ALERT ALERT ALERT: version 16 never released to customers, internal and beta use only
+    BRT_LAYOUT_VERSION_17 = 17, // Dr. No:  Add STAT64INFO_S to brt_header
     BRT_NEXT_VERSION,           // the version after the current version
     BRT_LAYOUT_VERSION   = BRT_NEXT_VERSION-1, // A hack so I don't have to change this line.
     BRT_LAYOUT_MIN_SUPPORTED_VERSION = BRT_LAYOUT_VERSION_13, // Minimum version supported

newbrt/brtdump.c

@@ -105,6 +105,8 @@ dump_header (int f, struct brt_header **header, CACHEFILE cf) {
     printf(" unnamed_root=%" PRId64 "\n", h->root.b);
     printf(" flags=%u\n", h->flags);
     dump_descriptor(&h->descriptor);
+    printf(" estimated numrows=%" PRId64 "\n", h->in_memory_stats.numrows);
+    printf(" estimated numbytes=%" PRId64 "\n", h->in_memory_stats.numbytes);
     *header = h;
 }
 

newbrt/brtloader.c

@@ -2060,15 +2060,17 @@ struct dbout {
     int64_t n_translations_limit;
     struct translation *translation;
     toku_pthread_mutex_t mutex;
+    struct brt_header *h;
 };
 
-static inline void dbout_init(struct dbout *out) {
+static inline void dbout_init(struct dbout *out, struct brt_header *h) {
     out->fd = -1;
     out->current_off = 0;
     out->n_translations = out->n_translations_limit = 0;
     out->translation = NULL;
     int r = toku_pthread_mutex_init(&out->mutex, NULL);
     resource_assert_zero(r);
+    out->h = h;
 }
 
 static inline void dbout_destroy(struct dbout *out) {
@@ -2194,7 +2196,7 @@ static struct leaf_buf *start_leaf (struct dbout *out, const DESCRIPTOR UU(desc)
     }
 
     BRTNODE XMALLOC(node);
-    toku_initialize_empty_brtnode(node, lbuf->blocknum, 0 /*height*/, 1 /*basement nodes*/, BRT_LAYOUT_VERSION, target_nodesize, 0);
+    toku_initialize_empty_brtnode(node, lbuf->blocknum, 0 /*height*/, 1 /*basement nodes*/, BRT_LAYOUT_VERSION, target_nodesize, 0, out->h);
     BP_STATE(node, 0) = PT_AVAIL;
     lbuf->node = node;
 
@@ -2207,7 +2209,7 @@ static int write_nonleaves (BRTLOADER bl, FIDX pivots_fidx, struct dbout *out, s
 CILK_END
 static void add_pair_to_leafnode (struct leaf_buf *lbuf, unsigned char *key, int keylen, unsigned char *val, int vallen, int this_leafentry_size);
 static int write_translation_table (struct dbout *out, long long *off_of_translation_p);
-static int write_header (struct dbout *out, long long translation_location_on_disk, long long translation_size_on_disk, BLOCKNUM root_blocknum_on_disk, LSN load_lsn, TXNID root_xid, uint32_t target_nodesize, uint32_t target_basementnodesize);
+static int write_header (struct dbout *out, long long translation_location_on_disk, long long translation_size_on_disk);
 
 static void drain_writer_q(QUEUE q) {
     void *item;
@@ -2276,8 +2278,15 @@ static int toku_loader_write_brt_from_q (BRTLOADER bl,
     }
     FILE *pivots_stream = toku_bl_fidx2file(bl, pivots_file);
 
+    TXNID root_xid_that_created = TXNID_NONE;
+    if (bl->root_xids_that_created)
+        root_xid_that_created = bl->root_xids_that_created[which_db];
+
+    struct brt_header h; 
+    toku_brt_header_init(&h, (BLOCKNUM){0}, bl->load_lsn, root_xid_that_created, target_nodesize, target_basementnodesize);
+
     struct dbout out;
-    dbout_init(&out);
+    dbout_init(&out, &h);
     out.fd = fd;
     out.current_off = 8192; // leave 8K reserved at beginning
     out.n_translations = 3; // 3 translations reserved at the beginning
@@ -2439,7 +2448,7 @@ static int toku_loader_write_brt_from_q (BRTLOADER bl,
 
     {
 	invariant(sts.n_subtrees==1);
-	BLOCKNUM root_block = make_blocknum(sts.subtrees[0].block);
+	out.h->root = make_blocknum(sts.subtrees[0].block);
 	toku_free(sts.subtrees); sts.subtrees = NULL;
 
 	// write the descriptor
@@ -2472,11 +2481,7 @@ static int toku_loader_write_brt_from_q (BRTLOADER bl,
             result = r; goto error;
         }
 
-        TXNID root_xid_that_created = TXNID_NONE;
-        if (bl->root_xids_that_created) {
-            root_xid_that_created = bl->root_xids_that_created[which_db];
-        }
-        r = write_header(&out, off_of_translation, (out.n_translations+1)*16+4, root_block, bl->load_lsn, root_xid_that_created, target_nodesize, target_basementnodesize);
+        r = write_header(&out, off_of_translation, (out.n_translations+1)*16+4);
         if (r) {
             result = r; goto error;
         }
@@ -2768,7 +2773,7 @@ static void add_pair_to_leafnode (struct leaf_buf *lbuf, unsigned char *key, int
     DBT theval = { .data = val, .size = vallen };
     BRT_MSG_S cmd = { BRT_INSERT, ZERO_MSN, lbuf->xids, .u.id = { &thekey, &theval } };
     uint64_t workdone=0;
-    brt_leaf_apply_cmd_once(BLB(leafnode,0), &cmd, idx, NULL, NULL, NULL, &workdone);
+    brt_leaf_apply_cmd_once(leafnode, BLB(leafnode,0), &cmd, idx, NULL, NULL, NULL, &workdone);
 }
 
 static int write_literal(struct dbout *out, void*data,  size_t len) {
@@ -2844,37 +2849,19 @@ static int write_translation_table (struct dbout *out, long long *off_of_transla
     return result;
 }
 
-
 static int
-write_header (struct dbout *out, long long translation_location_on_disk, long long translation_size_on_disk, BLOCKNUM root_blocknum_on_disk,
-              LSN load_lsn, TXNID root_xid_that_created, uint32_t target_nodesize, uint32_t target_basementnodesize) {
+write_header (struct dbout *out, long long translation_location_on_disk, long long translation_size_on_disk) {
     int result = 0;
 
-    struct brt_header h; memset(&h, 0, sizeof h);
-    h.layout_version   = BRT_LAYOUT_VERSION;
-    h.layout_version_original = BRT_LAYOUT_VERSION;
-    h.build_id         = BUILD_ID;
-    h.build_id_original = BUILD_ID;
-    uint64_t now = (uint64_t) time(NULL);
-    h.time_of_creation = now;
-    h.time_of_last_modification = now;
-    h.time_of_last_verification = 0;
-    h.checkpoint_count = 1;
-    h.checkpoint_lsn   = load_lsn;
-    h.nodesize         = target_nodesize;
-    h.basementnodesize = target_basementnodesize;
-    h.root             = root_blocknum_on_disk;
-    h.flags            = 0;
-    h.root_xid_that_created = root_xid_that_created;
-
-    unsigned int size = toku_serialize_brt_header_size (&h);
+    out->h->checkpoint_staging_stats = out->h->in_memory_stats; // #4184
+    unsigned int size = toku_serialize_brt_header_size (out->h);
     struct wbuf wbuf;
     char *MALLOC_N(size, buf);
     if (buf == NULL) {
         result = errno;
     } else {
         wbuf_init(&wbuf, buf, size);
-        toku_serialize_brt_header_to_wbuf(&wbuf, &h, translation_location_on_disk, translation_size_on_disk);
+        toku_serialize_brt_header_to_wbuf(&wbuf, out->h, translation_location_on_disk, translation_size_on_disk);
         if (wbuf.ndone != size)
             result = EINVAL;
         else
@@ -2993,7 +2980,7 @@ static void write_nonleaf_node (BRTLOADER bl, struct dbout *out, int64_t blocknu
 
     BRTNODE XMALLOC(node);
     toku_initialize_empty_brtnode(node, make_blocknum(blocknum_of_new_node), height, n_children, 
-                                 BRT_LAYOUT_VERSION, target_nodesize, 0);
+				  BRT_LAYOUT_VERSION, target_nodesize, 0, NULL);
     for (int i=0; i<n_children-1; i++)
         node->childkeys[i] = NULL;
     unsigned int totalchildkeylens = 0;

newbrt/cachetable.c

@@ -471,6 +471,13 @@ cachefile_refup (CACHEFILE cf) {
     cf->refcount++;
 }
 
+BOOL 
+toku_cachefile_is_closing (CACHEFILE cf) {
+    BOOL rval = cf->is_closing;
+    return rval;
+}
+
+
 // What cachefile goes with particular iname (iname relative to env)?
 // The transaction that is adding the reference might not have a reference
 // to the brt, therefore the cachefile might be closing.
@@ -1218,7 +1225,9 @@ static void cachetable_remove_pair (CACHETABLE ct, PAIR p) {
 // Maybe remove a pair from the cachetable and free it, depending on whether
 // or not there are any threads interested in the pair.  The flush callback
 // is called with write_me and keep_me both false, and the pair is destroyed.
-
+// The sole purpose of this function is to remove the node, so the write_me 
+// argument to the flush callback is false, and the flush callback won't do
+// anything except destroy the node.
 static void cachetable_maybe_remove_and_free_pair (CACHETABLE ct, PAIR p, BOOL* destroyed) {
     *destroyed = FALSE;
     if (nb_mutex_users(&p->nb_mutex) == 0) {
@@ -1236,6 +1245,8 @@ static void cachetable_maybe_remove_and_free_pair (CACHETABLE ct, PAIR p, BOOL*
         cachetable_evictions++;
         cachetable_unlock(ct);
         PAIR_ATTR new_attr = p->attr;
+	// Note that flush_callback is called with write_me FALSE, so the only purpose of this 
+	// call is to tell the brt layer to evict the node (keep_me is FALSE).
         flush_callback(cachefile, cachefile->fd, key, value, write_extraargs, old_attr, &new_attr, FALSE, FALSE, TRUE);
 
         cachetable_lock(ct);

newbrt/cachetable.h

@@ -60,6 +60,9 @@ int toku_cachefile_of_iname_in_env (CACHETABLE ct, const char *iname_in_env, CAC
 // Return the filename
 char * toku_cachefile_fname_in_cwd (CACHEFILE cf);
 
+// return value of the cachefile is_closing flag
+BOOL toku_cachefile_is_closing(CACHEFILE cf);
+
 // TODO: #1510  Add comments on how these behave
 int toku_cachetable_begin_checkpoint (CACHETABLE ct, TOKULOGGER);
 int toku_cachetable_end_checkpoint(CACHETABLE ct, TOKULOGGER logger, 

newbrt/tests/Makefile

@@ -29,6 +29,7 @@ endif
 # For very verbose output do 
 #   make VERBOSE=2
 
+# Exclude orthopush-flush until it is fixed.
 SRCS=$(sort $(filter-out dir.%.c,$(wildcard *.c)))
 
 REGRESSION_TESTS_RAW = $(patsubst %.c,%,$(SRCS))

newbrt/tests/brt-bfe-query.c

 /* -*- mode: C; c-basic-offset: 4 -*- */
 #ident "$Id$"
-#ident "Copyright (c) 2007, 2008 Tokutek Inc.  All rights reserved."
+#ident "Copyright (c) 2011 Tokutek Inc.  All rights reserved."
 
 #include "test.h"
-
 #include "includes.h"
 
-#define MIN(x, y) (((x) < (y)) ? (x) : (y))
-
-
 static  int
 int64_key_cmp (DB *db UU(), const DBT *a, const DBT *b) {
     int64_t x = *(int64_t *) a->data;

newbrt/tests/brtloader-test-writer.c

@@ -60,6 +60,7 @@ static void verify_dbfile(int n, const char *name) {
     BRT_CURSOR cursor = NULL;
     r = toku_brt_cursor(t, &cursor, NULL, FALSE, FALSE); assert(r == 0);
 
+    size_t userdata = 0;
     int i;
     for (i=0; ; i++) {
 	int kk = i;
@@ -71,11 +72,17 @@ static void verify_dbfile(int n, const char *name) {
 	    break;
 	}
 	assert(pair.call_count==1);
+        userdata += pair.keylen + pair.vallen;
     }
 
     assert(i == n);
 
     r = toku_brt_cursor_close(cursor); assert(r == 0);
+
+    struct brtstat64_s s;
+    r = toku_brt_stat64(t, NULL, &s); assert(r == 0);
+    assert(s.nkeys == (u_int64_t)n && s.ndata == (u_int64_t)n && s.dsize == userdata);
+
     r = toku_close_brt(t, 0); assert(r==0);
     r = toku_cachetable_close(&ct);assert(r==0);
     if (verbose) traceit("verify done");

newbrt/tests/brtloader-test.c

@@ -253,6 +253,7 @@ static void verify_dbfile(int n, int sorted_keys[], const char *sorted_vals[], c
     BRT_CURSOR cursor = NULL;
     r = toku_brt_cursor(t, &cursor, NULL, FALSE, FALSE); assert(r == 0);
 
+    size_t userdata = 0;
     int i;
     for (i=0; i<n; i++) {
 	struct check_pair pair = {sizeof sorted_keys[i], &sorted_keys[i], strlen(sorted_vals[i]), sorted_vals[i], 0};
@@ -262,6 +263,7 @@ static void verify_dbfile(int n, int sorted_keys[], const char *sorted_vals[], c
 	    break;
 	}
 	assert(pair.call_count==1);
+        userdata += pair.keylen + pair.vallen;
     }
     
     struct check_pair pair; memset(&pair, 0, sizeof pair);
@@ -269,11 +271,16 @@ static void verify_dbfile(int n, int sorted_keys[], const char *sorted_vals[], c
     assert(r != 0);
 
     r = toku_brt_cursor_close(cursor); assert(r == 0);
+
+    struct brtstat64_s s;
+    r = toku_brt_stat64(t, NULL, &s); assert(r == 0);
+    assert(s.nkeys == (u_int64_t) n && s.ndata == (u_int64_t) n && s.dsize == userdata);
+    
     r = toku_close_brt(t, 0); assert(r==0);
     r = toku_cachetable_close(&ct);assert(r==0);
 }
 
-    static void test_merge_files (const char *template, const char *output_name) {
+static void test_merge_files (const char *template, const char *output_name) {
     DB *dest_db = NULL;
     struct brtloader_s bl = {
         .temp_file_template = template,

newbrt/tests/cachetable-partial-fetch.c

@@ -68,10 +68,12 @@ static BOOL true_pf_req_callback(void* UU(brtnode_pv), void* UU(read_extraargs))
 
 static int err_pf_callback(void* UU(brtnode_pv), void* UU(read_extraargs), int UU(fd), PAIR_ATTR* UU(sizep)) {
   assert(FALSE);
+  return 0; // gcov
 }
 
 static int pf_callback(void* UU(brtnode_pv), void* UU(read_extraargs), int UU(fd), PAIR_ATTR* UU(sizep)) {
   assert(FALSE);
+  return 0; // gcov
 }
 
 static int true_pf_callback(void* UU(brtnode_pv), void* read_extraargs, int UU(fd), PAIR_ATTR* sizep) {

newbrt/tests/keyrange.c

 /* -*- mode: C; c-basic-offset: 4 -*- */
 #ident "$Id$"
-#ident "Copyright (c) 2008 Tokutek Inc.  All rights reserved."
+#ident "Copyright (c) 2008-2011 Tokutek Inc.  All rights reserved."
 
 // Test keyrange
 
@@ -12,9 +12,9 @@
 static TOKUTXN const null_txn = 0;
 static DB * const null_db = 0;
 
-char fname[]= __FILE__ ".brt";
-CACHETABLE ct;
-BRT t;
+static char fname[]= __FILE__ ".brt";
+static CACHETABLE ct;
+static BRT t;
 
 static void close_brt_and_ct (void) {
     int r;
@@ -35,6 +35,7 @@ static void close_and_reopen (void) {
 }
 
 static void reload (u_int64_t limit) {
+    // insert keys 1, 3, 5, ...
     for (u_int64_t i=0; i<limit; i++) {
 	char key[100],val[100];
 	snprintf(key, 100, "%08llu", (unsigned long long)2*i+1);
@@ -64,9 +65,10 @@ static void maybe_reopen (enum memory_state ms, u_int64_t limit) {
     assert(0);
 }
 
-static void test_keyrange (enum memory_state ms) {
-    u_int64_t limit=30000;
+static void test_keyrange (enum memory_state ms, u_int64_t limit) {
     open_brt_and_ct(true);
+
+    // insert keys 1, 3, 5, ...
     for (u_int64_t i=0; i<limit; i++) {
 	char key[100],val[100];
 	snprintf(key, 100, "%08llu", (unsigned long long)2*i+1);
@@ -75,11 +77,23 @@ static void test_keyrange (enum memory_state ms) {
 	int r = toku_brt_insert(t, toku_fill_dbt(&k, key, 1+strlen(key)), toku_fill_dbt(&v,val, 1+strlen(val)), null_txn);
 	assert(r == 0);
     }
+
+    {
+        struct brtstat64_s s;
+        int r = toku_brt_stat64(t, null_txn, &s); assert(r == 0);
+        
+        assert(0 < s.nkeys && s.nkeys < limit);
+        assert(0 < s.dsize && s.dsize < limit * (9 + 9)); // keylen = 9, vallen = 9
+    }
+    
     maybe_reopen(ms, limit);
+
     {
 	u_int64_t prev_less = 0, prev_greater = 1LL << 60; 
 	u_int64_t count_less_adjacent = 0, count_greater_adjacent = 0; // count the number of times that the next value is 1 more (less) than the previous.
 	u_int64_t equal_count = 0;
+
+        // lookup keys 1, 3, 5, ...
 	for (u_int64_t i=0; i<limit; i++) {
 	    char key[100];
 	    snprintf(key, 100, "%08llu", (unsigned long long)2*i+1);
@@ -87,11 +101,18 @@ static void test_keyrange (enum memory_state ms) {
 	    u_int64_t less,equal,greater;
 	    int r = toku_brt_keyrange(t, toku_fill_dbt(&k, key, 1+strlen(key)), &less, &equal, &greater);
 	    assert(r == 0);
-	    //printf("Pkey %llu/%llu %llu %llu %llu\n", (unsigned long long)2*i+1, (unsigned long long)2*limit, (unsigned long long)less, (unsigned long long)equal, (unsigned long long)greater);
-	    // It's an estimate, and it the values don't even change monotonically.
+	    if (verbose > 1) 
+                printf("Pkey %llu/%llu %llu %llu %llu\n", (unsigned long long)2*i+1, (unsigned long long)2*limit, (unsigned long long)less, (unsigned long long)equal, (unsigned long long)greater);
+
+            assert(0 < less + equal + greater);
+            assert(less + equal + greater <= 2 * limit);
+            assert(equal == 0 || equal == 1);
+
+	    // It's an estimate, and the values don't even change monotonically.
 	    // And all the leaves are in main memory so it's always present.
 	    if (ms!=CLOSE_AND_REOPEN_LEAVE_ON_DISK) {
 		if (equal==1) equal_count++;
+#if 0
 		// The first few items are exact for less.
 		if (i<70) {
 		    assert(less==i);
@@ -100,15 +121,18 @@ static void test_keyrange (enum memory_state ms) {
 		if (limit-i<70) {
 		    assert(greater<=limit-i-1);
 		}
+#endif
 	    } else {
-		// If we closed it, it's not in main memory, and so the less and greater estimates are wrong, and we set equal to 0.
-		assert(equal==0);
+		// after reopen, none of the basements are in memory
+		assert(equal == 0);
+#if 0
 		if (i<10) {
 		    assert(less==0);
 		}
 		if (limit-i<10) {
 		    assert(greater==0);
 		}
+#endif
 	    }
 	    // Count the number of times that prev_less is 1 less than less.
 	    if (prev_less+1 == less) {
@@ -129,7 +153,10 @@ static void test_keyrange (enum memory_state ms) {
 	    assert(equal_count >= 0.9 * limit);
 	}
     }
+
     maybe_reopen(ms, limit);
+
+    // lookup keys 0, 2, 4, ... not in the tree
     for (u_int64_t i=0; i<1+limit; i++) {
 	char key[100];
 	snprintf(key, 100, "%08llu", (unsigned long long)2*i);
@@ -137,8 +164,13 @@ static void test_keyrange (enum memory_state ms) {
 	u_int64_t less,equal,greater;
 	int r = toku_brt_keyrange(t, toku_fill_dbt(&k, key, 1+strlen(key)), &less, &equal, &greater);
 	assert(r == 0);
-	//printf("Akey %llu/%llu %llu %llu %llu\n", (unsigned long long)2*i, (unsigned long long)2*limit, (unsigned long long)less, (unsigned long long)equal, (unsigned long long)greater);
-	assert(equal==0);
+        if (verbose > 1)
+            printf("Akey %llu/%llu %llu %llu %llu\n", (unsigned long long)2*i, (unsigned long long)2*limit, (unsigned long long)less, (unsigned long long)equal, (unsigned long long)greater);
+
+        assert(0 < less + equal + greater);
+        assert(less + equal + greater <= 2 * limit);
+	assert(equal == 0);
+#if 0
 	// The first few items are exact (looking a key that's not there)
 	if (ms!=CLOSE_AND_REOPEN_LEAVE_ON_DISK) {
 	    if (i<70) {
@@ -156,17 +188,34 @@ static void test_keyrange (enum memory_state ms) {
 		assert(greater==0);
 	    }
 	}
+#endif
     }
+
     close_brt_and_ct();
 }
 
 int
 test_main (int argc , const char *argv[]) {
-    default_parse_args(argc, argv);
+    u_int64_t limit = 30000;
+
+    for (int i = 1; i < argc; i++) {
+        if (strcmp(argv[i], "-v") == 0) {
+            verbose++;
+            continue;
+        }
+        if (strcmp(argv[i], "-q") == 0) {
+            if (verbose > 0) verbose--;
+            continue;
+        }
+        if (strcmp(argv[i], "-n") == 0 && i+1 < argc) {
+            limit = atoll(argv[++i]);
+            continue;
+        }
+    }
 
-    test_keyrange(LEAVE_IN_MEMORY);
-    test_keyrange(CLOSE_AND_RELOAD);
-    test_keyrange(CLOSE_AND_REOPEN_LEAVE_ON_DISK);
+    test_keyrange(LEAVE_IN_MEMORY, limit);
+    test_keyrange(CLOSE_AND_REOPEN_LEAVE_ON_DISK, limit);
+    test_keyrange(CLOSE_AND_RELOAD, limit);
 
     if (verbose) printf("test ok\n");
     return 0;

newbrt/tests/make-tree.c

@@ -39,7 +39,7 @@ append_leaf(BRTNODE leafnode, void *key, size_t keylen, void *val, size_t vallen
 
     // apply an insert to the leaf node
     BRT_MSG_S cmd = { BRT_INSERT, msn, xids_get_root_xids(), .u.id = { &thekey, &theval } };
-    brt_leaf_apply_cmd_once(BLB(leafnode,0), &cmd, idx, NULL, NULL, NULL, NULL);
+    brt_leaf_apply_cmd_once(leafnode, BLB(leafnode,0), &cmd, idx, NULL, NULL, NULL, NULL);
 
     leafnode->max_msn_applied_to_node_on_disk = msn;
 

newbrt/tests/orthopush-flush.c

@@ -10,6 +10,8 @@
 static TOKUTXN const null_txn = 0;
 static DB * const null_db = 0;
 static char fname[] = __FILE__ ".brt";
+static struct brt_header my_fake_header;
+static struct brt_header *my_header = &my_fake_header;
 
 static int dummy_cmp(DB *db __attribute__((unused)),
                      const DBT *a, const DBT *b) {
@@ -96,7 +98,7 @@ insert_random_message(NONLEAF_CHILDINFO bnc, BRT_MSG_S **save, bool *is_fresh_ou
 // generate a random message with xids and a key starting with pfx, insert
 // it into blb, and save it in output param save
 static void
-insert_random_message_to_leaf(BRT t, BASEMENTNODE blb, LEAFENTRY *save, XIDS xids, int pfx)
+insert_random_message_to_leaf(BRT t, BRTNODE leafnode, BASEMENTNODE blb, LEAFENTRY *save, XIDS xids, int pfx)
 {
     int keylen = (random() % 16) + 16;
     int vallen = (random() % 1024) + 16;
@@ -122,10 +124,11 @@ insert_random_message_to_leaf(BRT t, BASEMENTNODE blb, LEAFENTRY *save, XIDS xid
     msg.u.id.key = keydbt;
     msg.u.id.val = valdbt;
     size_t memsize;
-    int r = apply_msg_to_leafentry(&msg, NULL, &memsize, save, NULL, NULL, NULL, NULL, NULL);
+    int64_t numbytes;
+    int r = apply_msg_to_leafentry(&msg, NULL, &memsize, save, NULL, NULL, NULL, NULL, NULL, &numbytes);
     assert_zero(r);
     bool made_change;
-    brt_leaf_put_cmd(t->compare_fun, t->update_fun, NULL, blb, &msg, &made_change, NULL, NULL, NULL);
+    brt_leaf_put_cmd(t->compare_fun, t->update_fun, NULL, leafnode, blb, &msg, &made_change, NULL, NULL, NULL);
     if (msn.msn > blb->max_msn_applied.msn) {
         blb->max_msn_applied = msn;
     }
@@ -137,7 +140,7 @@ insert_random_message_to_leaf(BRT t, BASEMENTNODE blb, LEAFENTRY *save, XIDS xid
 // used for making two leaf nodes the same in order to compare the result
 // of 'maybe_apply' and a normal buffer flush
 static void
-insert_same_message_to_leaves(BRT t, BASEMENTNODE blb1, BASEMENTNODE blb2, LEAFENTRY *save, XIDS xids, int pfx)
+insert_same_message_to_leaves(BRT t, BRTNODE child1, BASEMENTNODE blb1, BRTNODE child2, BASEMENTNODE blb2, LEAFENTRY *save, XIDS xids, int pfx)
 {
     int keylen = (random() % 16) + 16;
     int vallen = (random() % 1024) + 16;
@@ -163,14 +166,15 @@ insert_same_message_to_leaves(BRT t, BASEMENTNODE blb1, BASEMENTNODE blb2, LEAFE
     msg.u.id.key = keydbt;
     msg.u.id.val = valdbt;
     size_t memsize;
-    int r = apply_msg_to_leafentry(&msg, NULL, &memsize, save, NULL, NULL, NULL, NULL, NULL);
+    int64_t numbytes;
+    int r = apply_msg_to_leafentry(&msg, NULL, &memsize, save, NULL, NULL, NULL, NULL, NULL, &numbytes);
     assert_zero(r);
     bool made_change;
-    brt_leaf_put_cmd(t->compare_fun, t->update_fun, NULL, blb1, &msg, &made_change, NULL, NULL, NULL);
+    brt_leaf_put_cmd(t->compare_fun, t->update_fun, NULL, child1, blb1, &msg, &made_change, NULL, NULL, NULL);
     if (msn.msn > blb1->max_msn_applied.msn) {
         blb1->max_msn_applied = msn;
     }
-    brt_leaf_put_cmd(t->compare_fun, t->update_fun, NULL, blb2, &msg, &made_change, NULL, NULL, NULL);
+    brt_leaf_put_cmd(t->compare_fun, t->update_fun, NULL, child2, blb2, &msg, &made_change, NULL, NULL, NULL);
     if (msn.msn > blb2->max_msn_applied.msn) {
         blb2->max_msn_applied = msn;
     }
@@ -271,7 +275,7 @@ flush_to_internal(BRT t) {
 
     BRTNODE XMALLOC(child);
     BLOCKNUM blocknum = { 42 };
-    toku_initialize_empty_brtnode(child, blocknum, 1, 1, BRT_LAYOUT_VERSION, 4*M, 0);
+    toku_initialize_empty_brtnode(child, blocknum, 1, 1, BRT_LAYOUT_VERSION, 4*M, 0, my_header);
     destroy_nonleaf_childinfo(BNC(child, 0));
     set_BNC(child, 0, child_bnc);
     BP_STATE(child, 0) = PT_AVAIL;
@@ -401,7 +405,7 @@ flush_to_internal_multiple(BRT t) {
 
     BRTNODE XMALLOC(child);
     BLOCKNUM blocknum = { 42 };
-    toku_initialize_empty_brtnode(child, blocknum, 1, 8, BRT_LAYOUT_VERSION, 4*M, 0);
+    toku_initialize_empty_brtnode(child, blocknum, 1, 8, BRT_LAYOUT_VERSION, 4*M, 0, my_header);
     for (i = 0; i < 8; ++i) {
         destroy_nonleaf_childinfo(BNC(child, i));
         set_BNC(child, i, child_bncs[i]);
@@ -519,7 +523,7 @@ flush_to_leaf(BRT t, bool make_leaf_up_to_date, bool use_flush) {
     CKERR(r);
     r = xids_create_child(xids_0, &xids_234, (TXNID)234);
     CKERR(r);
-
+    
     BASEMENTNODE child_blbs[8];
     DBT childkeys[7];
     int i;
@@ -532,7 +536,7 @@ flush_to_leaf(BRT t, bool make_leaf_up_to_date, bool use_flush) {
 
     BRTNODE XMALLOC(child);
     BLOCKNUM blocknum = { 42 };
-    toku_initialize_empty_brtnode(child, blocknum, 0, 8, BRT_LAYOUT_VERSION, 4*M, 0);
+    toku_initialize_empty_brtnode(child, blocknum, 0, 8, BRT_LAYOUT_VERSION, 4*M, 0, my_header);
     for (i = 0; i < 8; ++i) {
         destroy_basement_node(BLB(child, i));
         set_BLB(child, i, child_blbs[i]);
@@ -542,7 +546,7 @@ flush_to_leaf(BRT t, bool make_leaf_up_to_date, bool use_flush) {
     int total_size = 0;
     for (i = 0; total_size < 4*M; ++i) {
         total_size -= child_blbs[i%8]->n_bytes_in_buffer;
-        insert_random_message_to_leaf(t, child_blbs[i%8], &child_messages[i], xids_123, i%8);
+        insert_random_message_to_leaf(t, child, child_blbs[i%8], &child_messages[i], xids_123, i%8);
         total_size += child_blbs[i%8]->n_bytes_in_buffer;
         if (i % 8 < 7) {
             u_int32_t keylen;
@@ -609,7 +613,7 @@ flush_to_leaf(BRT t, bool make_leaf_up_to_date, bool use_flush) {
     } else {
         BRTNODE XMALLOC(parentnode);
         BLOCKNUM parentblocknum = { 17 };
-        toku_initialize_empty_brtnode(parentnode, parentblocknum, 1, 1, BRT_LAYOUT_VERSION, 4*M, 0);
+        toku_initialize_empty_brtnode(parentnode, parentblocknum, 1, 1, BRT_LAYOUT_VERSION, 4*M, 0, my_header);
         destroy_nonleaf_childinfo(BNC(parentnode, 0));
         set_BNC(parentnode, 0, parent_bnc);
         BP_STATE(parentnode, 0) = PT_AVAIL;
@@ -759,7 +763,7 @@ flush_to_leaf_with_keyrange(BRT t, bool make_leaf_up_to_date) {
 
     BRTNODE XMALLOC(child);
     BLOCKNUM blocknum = { 42 };
-    toku_initialize_empty_brtnode(child, blocknum, 0, 8, BRT_LAYOUT_VERSION, 4*M, 0);
+    toku_initialize_empty_brtnode(child, blocknum, 0, 8, BRT_LAYOUT_VERSION, 4*M, 0, my_header);
     for (i = 0; i < 8; ++i) {
         destroy_basement_node(BLB(child, i));
         set_BLB(child, i, child_blbs[i]);
@@ -769,7 +773,7 @@ flush_to_leaf_with_keyrange(BRT t, bool make_leaf_up_to_date) {
     int total_size = 0;
     for (i = 0; total_size < 4*M; ++i) {
         total_size -= child_blbs[i%8]->n_bytes_in_buffer;
-        insert_random_message_to_leaf(t, child_blbs[i%8], &child_messages[i], xids_123, i%8);
+        insert_random_message_to_leaf(t, child, child_blbs[i%8], &child_messages[i], xids_123, i%8);
         total_size += child_blbs[i%8]->n_bytes_in_buffer;
         u_int32_t keylen;
         char *key = le_key_and_len(child_messages[i], &keylen);
@@ -829,7 +833,7 @@ flush_to_leaf_with_keyrange(BRT t, bool make_leaf_up_to_date) {
 
     BRTNODE XMALLOC(parentnode);
     BLOCKNUM parentblocknum = { 17 };
-    toku_initialize_empty_brtnode(parentnode, parentblocknum, 1, 1, BRT_LAYOUT_VERSION, 4*M, 0);
+    toku_initialize_empty_brtnode(parentnode, parentblocknum, 1, 1, BRT_LAYOUT_VERSION, 4*M, 0, my_header);
     destroy_nonleaf_childinfo(BNC(parentnode, 0));
     set_BNC(parentnode, 0, parent_bnc);
     BP_STATE(parentnode, 0) = PT_AVAIL;
@@ -936,8 +940,8 @@ compare_apply_and_flush(BRT t, bool make_leaf_up_to_date) {
 
     BRTNODE XMALLOC(child1), XMALLOC(child2);
     BLOCKNUM blocknum = { 42 };
-    toku_initialize_empty_brtnode(child1, blocknum, 0, 8, BRT_LAYOUT_VERSION, 4*M, 0);
-    toku_initialize_empty_brtnode(child2, blocknum, 0, 8, BRT_LAYOUT_VERSION, 4*M, 0);
+    toku_initialize_empty_brtnode(child1, blocknum, 0, 8, BRT_LAYOUT_VERSION, 4*M, 0, my_header);
+    toku_initialize_empty_brtnode(child2, blocknum, 0, 8, BRT_LAYOUT_VERSION, 4*M, 0, my_header);
     for (i = 0; i < 8; ++i) {
         destroy_basement_node(BLB(child1, i));
         set_BLB(child1, i, child1_blbs[i]);
@@ -950,7 +954,7 @@ compare_apply_and_flush(BRT t, bool make_leaf_up_to_date) {
     int total_size = 0;
     for (i = 0; total_size < 4*M; ++i) {
         total_size -= child1_blbs[i%8]->n_bytes_in_buffer;
-        insert_same_message_to_leaves(t, child1_blbs[i%8], child2_blbs[i%8], &child_messages[i], xids_123, i%8);
+        insert_same_message_to_leaves(t, child1, child1_blbs[i%8], child2, child2_blbs[i%8], &child_messages[i], xids_123, i%8);
         total_size += child1_blbs[i%8]->n_bytes_in_buffer;
         if (i % 8 < 7) {
             u_int32_t keylen;
@@ -1013,7 +1017,7 @@ compare_apply_and_flush(BRT t, bool make_leaf_up_to_date) {
 
     BRTNODE XMALLOC(parentnode);
     BLOCKNUM parentblocknum = { 17 };
-    toku_initialize_empty_brtnode(parentnode, parentblocknum, 1, 1, BRT_LAYOUT_VERSION, 4*M, 0);
+    toku_initialize_empty_brtnode(parentnode, parentblocknum, 1, 1, BRT_LAYOUT_VERSION, 4*M, 0, my_header);
     destroy_nonleaf_childinfo(BNC(parentnode, 0));
     set_BNC(parentnode, 0, parent_bnc);
     BP_STATE(parentnode, 0) = PT_AVAIL;
@@ -1128,7 +1132,7 @@ test_main (int argc, const char *argv[]) {
 
     // normally, just check a few things, but if --slow is provided, then
     // be thorough about it and repeat tests (since they're randomized)
-    if (!slow) {
+    if (!slow) {;
         flush_to_internal(t);
         flush_to_internal_multiple(t);
         flush_to_leaf(t, true, false);

newbrt/tests/test-leafentry-nested.c

@@ -393,12 +393,13 @@ test_le_apply(ULE ule_initial, BRT_MSG msg, ULE ule_expected) {
     CKERR(r);
 
     size_t result_memsize;
+    int64_t ignoreme;
     r = apply_msg_to_leafentry(msg,
                                le_initial,
                                &result_memsize,
                                &le_result,
                                NULL, NULL, NULL, 
-                               NULL, NULL);
+                               NULL, NULL, &ignoreme);
     CKERR(r);
 
     if (le_result)

newbrt/tests/test3884.c

@@ -27,6 +27,8 @@ static TOKUTXN const null_txn = 0;
 static DB * const null_db = 0;
 static const char fname[]= __FILE__ ".brt";
 
+static BRT_STATUS_S my_brt_status;
+
 static int omt_long_cmp(OMTVALUE p, void *q)
 {
     LEAFENTRY a = p, b = q;
@@ -170,7 +172,7 @@ test_split_on_boundary(void)
     BRTNODE nodea, nodeb;
     DBT splitk;
     // if we haven't done it right, we should hit the assert in the top of move_leafentries
-    brtleaf_split(brt->h, &sn, &nodea, &nodeb, &splitk, TRUE, 0, NULL);
+    brtleaf_split(brt->h, &sn, &nodea, &nodeb, &splitk, TRUE, 0, NULL, &my_brt_status);
 
     verify_basement_node_msns(nodea, dummy_msn_3884);
     verify_basement_node_msns(nodeb, dummy_msn_3884);
@@ -243,7 +245,7 @@ test_split_with_everything_on_the_left(void)
     BRTNODE nodea, nodeb;
     DBT splitk;
     // if we haven't done it right, we should hit the assert in the top of move_leafentries
-    brtleaf_split(brt->h, &sn, &nodea, &nodeb, &splitk, TRUE, 0, NULL);
+    brtleaf_split(brt->h, &sn, &nodea, &nodeb, &splitk, TRUE, 0, NULL, &my_brt_status);
 
     toku_unpin_brtnode(brt, nodeb);
     r = toku_close_brt(brt, NULL); assert(r == 0);
@@ -318,7 +320,7 @@ test_split_on_boundary_of_last_node(void)
     BRTNODE nodea, nodeb;
     DBT splitk;
     // if we haven't done it right, we should hit the assert in the top of move_leafentries
-    brtleaf_split(brt->h, &sn, &nodea, &nodeb, &splitk, TRUE, 0, NULL);
+    brtleaf_split(brt->h, &sn, &nodea, &nodeb, &splitk, TRUE, 0, NULL, &my_brt_status);
 
     toku_unpin_brtnode(brt, nodeb);
     r = toku_close_brt(brt, NULL); assert(r == 0);
@@ -386,7 +388,7 @@ test_split_at_begin(void)
     BRTNODE nodea, nodeb;
     DBT splitk;
     // if we haven't done it right, we should hit the assert in the top of move_leafentries
-    brtleaf_split(brt->h, &sn, &nodea, &nodeb, &splitk, TRUE, 0, NULL);
+    brtleaf_split(brt->h, &sn, &nodea, &nodeb, &splitk, TRUE, 0, NULL, &my_brt_status);
 
     toku_unpin_brtnode(brt, nodeb);
     r = toku_close_brt(brt, NULL); assert(r == 0);
@@ -450,7 +452,7 @@ test_split_at_end(void)
     BRTNODE nodea, nodeb;
     DBT splitk;
     // if we haven't done it right, we should hit the assert in the top of move_leafentries
-    brtleaf_split(brt->h, &sn, &nodea, &nodeb, &splitk, TRUE, 0, NULL);
+    brtleaf_split(brt->h, &sn, &nodea, &nodeb, &splitk, TRUE, 0, NULL, &my_brt_status);
 
     toku_unpin_brtnode(brt, nodeb);
     r = toku_close_brt(brt, NULL); assert(r == 0);
@@ -504,7 +506,7 @@ test_split_odd_nodes(void)
     BRTNODE nodea, nodeb;
     DBT splitk;
     // if we haven't done it right, we should hit the assert in the top of move_leafentries
-    brtleaf_split(brt->h, &sn, &nodea, &nodeb, &splitk, TRUE, 0, NULL);
+    brtleaf_split(brt->h, &sn, &nodea, &nodeb, &splitk, TRUE, 0, NULL, &my_brt_status);
 
     verify_basement_node_msns(nodea, dummy_msn_3884);
     verify_basement_node_msns(nodeb, dummy_msn_3884);

newbrt/tests/verify-bad-msn.c

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

newbrt/tests/verify-bad-pivots.c

@@ -30,7 +30,7 @@ append_leaf(BRTNODE leafnode, void *key, size_t keylen, void *val, size_t vallen
     // apply an insert to the leaf node
     MSN msn = next_dummymsn();
     BRT_MSG_S cmd = { BRT_INSERT, msn, xids_get_root_xids(), .u.id = { &thekey, &theval } };
-    brt_leaf_apply_cmd_once(BLB(leafnode, 0), &cmd, idx, NULL, NULL, NULL, NULL);
+    brt_leaf_apply_cmd_once(leafnode, BLB(leafnode, 0), &cmd, idx, NULL, NULL, NULL, NULL);
 
     // dont forget to dirty the node
     leafnode->dirty = 1;

newbrt/tests/verify-dup-in-leaf.c

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

newbrt/tests/verify-dup-pivots.c

@@ -30,7 +30,7 @@ append_leaf(BRTNODE leafnode, void *key, size_t keylen, void *val, size_t vallen
     // apply an insert to the leaf node
     MSN msn = next_dummymsn();
     BRT_MSG_S cmd = { BRT_INSERT, msn, xids_get_root_xids(), .u.id = { &thekey, &theval } };
-    brt_leaf_apply_cmd_once(BLB(leafnode, 0), &cmd, idx, NULL, NULL, NULL, NULL);
+    brt_leaf_apply_cmd_once(leafnode, BLB(leafnode, 0), &cmd, idx, NULL, NULL, NULL, NULL);
 
     // dont forget to dirty the node
     leafnode->dirty = 1;

newbrt/tests/verify-misrouted-msgs.c

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

newbrt/tests/verify-unsorted-leaf.c

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

newbrt/tests/verify-unsorted-pivots.c

@@ -30,7 +30,7 @@ append_leaf(BRTNODE leafnode, void *key, size_t keylen, void *val, size_t vallen
     // apply an insert to the leaf node
     MSN msn = next_dummymsn();
     BRT_MSG_S cmd = { BRT_INSERT, msn, xids_get_root_xids(), .u.id = { &thekey, &theval } };
-    brt_leaf_apply_cmd_once(BLB(leafnode, 0), &cmd, idx, NULL, NULL, NULL, NULL);
+    brt_leaf_apply_cmd_once(leafnode, BLB(leafnode, 0), &cmd, idx, NULL, NULL, NULL, NULL);
 
     // dont forget to dirty the node
     leafnode->dirty = 1;

newbrt/ule.c

@@ -33,6 +33,8 @@
 
 static LE_STATUS_S status;
 
+static uint32_t ule_get_innermost_numbytes(ULE ule);
+
 
 ///////////////////////////////////////////////////////////////////////////////////
 // Accessor functions used by outside world (e.g. indexer)
@@ -305,14 +307,19 @@ apply_msg_to_leafentry(BRT_MSG   msg,		// message to apply to leafentry
 		       struct mempool *mp, 
 		       void **maybe_free,
                        OMT snapshot_xids,
-                       OMT live_list_reverse) {
+                       OMT live_list_reverse,
+                       int64_t * numbytes_delta_p) {  // change in total size of key and val, not including any overhead
     ULE_S ule;
     int rval;
+    int64_t oldnumbytes = 0;
+    int64_t newnumbytes = 0;
 
-    if (old_leafentry == NULL)           // if leafentry does not exist ...
-        msg_init_empty_ule(&ule, msg);   // ... create empty unpacked leaf entry
-    else 
+    if (old_leafentry == NULL)            // if leafentry does not exist ...
+        msg_init_empty_ule(&ule, msg);    // ... create empty unpacked leaf entry
+    else {
         le_unpack(&ule, old_leafentry); // otherwise unpack leafentry 
+	oldnumbytes = ule_get_innermost_numbytes(&ule);
+    }
     msg_modify_ule(&ule, msg);          // modify unpacked leafentry
     if (snapshot_xids && live_list_reverse) {
         garbage_collection(&ule, snapshot_xids, live_list_reverse);
@@ -322,8 +329,10 @@ apply_msg_to_leafentry(BRT_MSG   msg,		// message to apply to leafentry
 		   new_leafentry_p,
 		   omt,
 		   mp,
-		   maybe_free
-        );                       
+		   maybe_free);                       
+    if (new_leafentry_p) 
+	newnumbytes = ule_get_innermost_numbytes(&ule);
+    *numbytes_delta_p = newnumbytes - oldnumbytes;
     ule_cleanup(&ule);
     return rval;
 }
@@ -1710,6 +1719,22 @@ ule_get_keylen(ULE ule) {
     return ule->keylen;
 }
 
+
+// return size of data for innermost uxr, size of key plus size of val
+uint32_t
+ule_get_innermost_numbytes(ULE ule) {
+    uint32_t rval;
+    UXR uxr = ule_get_innermost_uxr(ule);
+    if (uxr_is_delete(uxr))
+	rval = 0;
+    else {
+	rval = uxr_get_vallen(uxr);
+	rval += ule_get_keylen(ule);
+    }
+    return rval;
+}
+
+
 /////////////////////////////////////////////////////////////////////////////////
 //  This layer of abstraction (uxr_xxx) understands uxr and nothing else.
 //

newbrt/ule.h

@@ -60,7 +60,8 @@ int apply_msg_to_leafentry(BRT_MSG   msg,
 			   struct mempool *mp, 
 			   void **maybe_free,
                            OMT snapshot_xids,
-                           OMT live_list_reverse);
+                           OMT live_list_reverse,
+                           int64_t * numbytes_delta_p);
 
 
 TXNID toku_get_youngest_live_list_txnid_for(TXNID xc, OMT live_list_reverse);

src/tests/Makefile

@@ -139,9 +139,6 @@ BDB_DONTRUN_TESTS = \
         isolation \
         isolation-read-committed \
         keyrange \
-        keyrange-dupsort \
-        keyrange-dupsort-unflat \
-        keyrange-unflat \
 	last-verify-time \
         loader-cleanup-test \
 	loader-create-abort \
@@ -242,7 +239,7 @@ BDB_DONTRUN_TESTS = \
         root_fifo_32 \
         root_fifo_41 \
 	shutdown-3344 \
-        stat64 stat64-create-modify-times stat64_flatten stat64-null-txn \
+        stat64 stat64-create-modify-times stat64_flatten stat64-null-txn stat64-root-changes \
 	stress-gc \
         test1324 \
         test1426 \
@@ -1012,6 +1009,30 @@ upgrade-tests.tdbrun: $(UPGRADE_TESTS_VALGRIND)
 upgrade-test-%.valgrind: upgrade-test-%.tdb
 	$(VGRIND) ./$< $(VERBVERBOSE) $(SUMMARIZE_CMD)
 
+# keyrange tests
+keyrange.tdbrun: keyrange-get0.tdbrun keyrange-get1.tdbrun \
+	keyrange-random-get0.tdbrun keyrange-random-get1.tdbrun \
+	keyrange-loader-get0.tdbrun keyrange-loader-get1.tdbrun
+	true
+
+keyrange-get0.tdbrun: keyrange.tdb
+	$(TDBVGRIND) ./$< $(VERBVERBOSE) --get 0 --envdir dir.$@ $(MAYBEINVERTER) $(SUMMARIZE_CMD)
+
+keyrange-get1.tdbrun: keyrange.tdb
+	$(TDBVGRIND) ./$< $(VERBVERBOSE) --get 1 --envdir dir.$@ $(MAYBEINVERTER) $(SUMMARIZE_CMD)
+
+keyrange-random-get0.tdbrun: keyrange.tdb
+	$(TDBVGRIND) ./$< $(VERBVERBOSE) --get 0 --random_keys 1 --envdir dir.$@ $(MAYBEINVERTER) $(SUMMARIZE_CMD)
+
+keyrange-random-get1.tdbrun: keyrange.tdb
+	$(TDBVGRIND) ./$< $(VERBVERBOSE) --get 1 --random_keys 1 --envdir dir.$@ $(MAYBEINVERTER) $(SUMMARIZE_CMD)
+
+keyrange-loader-get0.tdbrun: keyrange.tdb
+	$(TDBVGRIND) ./$< $(VERBVERBOSE) --get 0 --loader 1 --envdir dir.$@ $(MAYBEINVERTER) $(SUMMARIZE_CMD)
+
+keyrange-loader-get1.tdbrun: keyrange.tdb
+	$(TDBVGRIND) ./$< $(VERBVERBOSE) --get 1 --loader 1 --envdir dir.$@ $(MAYBEINVERTER) $(SUMMARIZE_CMD)
+
 clean:
 	rm -f $(ALL_BINS)
 	rm -rf dir.* *.check.output *.check.valgrind

src/tests/loader-stress-test.c

@@ -432,10 +432,7 @@ static void test_loader(DB **dbs)
                            NUM_ROWS, stats.bt_nkeys, stats.bt_ndata, stats.bt_dsize, stats.bt_fsize);
 		assert(stats.bt_nkeys <= (u_int64_t)NUM_ROWS);  // Fix as part of #4129.  Was ==
 		assert(stats.bt_ndata <= (u_int64_t)NUM_ROWS);
-		// Fix as part of #4129.
-		// Was: 	//assert(stats.bt_dsize == ((u_int64_t)NUM_ROWS) * 2 * sizeof(unsigned int));
-		// Now is
-		assert(stats.bt_dsize == stats.bt_nkeys * 100);
+		assert(stats.bt_dsize == ((u_int64_t)NUM_ROWS) * 2 * sizeof(unsigned int));
 		r = txn->commit(txn, 0);
 		CKERR(r);
 	    }

src/tests/stat64-null-txn.c

@@ -76,9 +76,9 @@ test_stat64 (unsigned int N) {
             printf("fsize=%" PRIu64 "\n", s.bt_fsize);
             printf("expected dsize=%" PRIu64 "\n", dsize); 
         }
-        assert(s.bt_nkeys <= 4*N);      // This can probably be tightened up when we fix #3995.
-        assert(s.bt_ndata <= 4*N);      // This can probably be tightened up when we fix #3995.
-        assert(s.bt_dsize <= 15*dsize); // This can probably be tightened up when we fix #3995.
+        assert(0 < s.bt_nkeys && s.bt_nkeys <= N);
+        assert(s.bt_ndata == s.bt_nkeys);
+        assert(0 < s.bt_dsize && s.bt_dsize <= dsize);
         assert(s.bt_fsize > N);
     }
     /* r=txn->commit(txn, 0); CKERR(r); */
@@ -109,9 +109,9 @@ test_stat64 (unsigned int N) {
             printf("fsize=%" PRIu64 "\n", s.bt_fsize);
             printf("expected dsize=%" PRIu64 "\n", dsize); 
         }
-        assert(s.bt_nkeys <= 4*N);      // This can probably be tightened up when we fix #3995.
-        assert(s.bt_ndata <= 4*N);      // This can probably be tightened up when we fix #3995.
-        assert(s.bt_dsize <= 15*dsize); // This can probably be tightened up when we fix #3995.
+        assert(0 < s.bt_nkeys && s.bt_nkeys <= N);
+        assert(s.bt_ndata == s.bt_nkeys);
+        assert(0 < s.bt_dsize && s.bt_dsize <= dsize);
         assert(s.bt_fsize > N);
     }
     /* r=txn->commit(txn, 0); CKERR(r); */

src/tests/stat64-root-changes.c

+/* -*- mode: C; c-basic-offset: 4 -*- */
+#ident "Copyright (c) 2007-2011 Tokutek Inc.  All rights reserved."
+#include "test.h"
+
+// verify stats after a new row inserted into the root
+// verify stats after a row overwrite in the root
+// verify stats after a row deletion in the root
+// verify stats after an update callback inserts row
+// verify stats after an update callback overwrites row
+// verify ststs after an update callback deletes row
+
+#include <db.h>
+#include <unistd.h>
+#include <sys/stat.h>
+
+static int 
+my_update_callback(DB *db UU(), const DBT *key UU(), const DBT *old_val, const DBT *extra, void (*set_val)(const DBT *new_val, void *set_extra), void *set_extra) {
+    if (old_val != NULL && old_val->size == 42) // special code for delete
+        set_val(NULL, set_extra);
+    else
+        set_val(extra, set_extra);
+    return 0;
+}
+
+static void 
+run_test (void) {
+    int r;
+
+    r = system("rm -rf " ENVDIR); CKERR(r);
+    r = toku_os_mkdir(ENVDIR, S_IRWXU+S_IRWXG+S_IRWXO);       CKERR(r);
+
+    DB_ENV *env = NULL;
+    r = db_env_create(&env, 0); CKERR(r);
+    env->set_errfile(env, stderr);
+    r = env->set_redzone(env, 0); CKERR(r);
+    env->set_update(env, my_update_callback);
+    r = env->open(env, ENVDIR, DB_INIT_LOCK|DB_INIT_LOG|DB_INIT_MPOOL|DB_INIT_TXN|DB_CREATE|DB_PRIVATE, S_IRWXU+S_IRWXG+S_IRWXO); CKERR(r);
+
+    DB *db = NULL;
+    r = db_create(&db, env, 0); CKERR(r);
+    DB_TXN *txn = NULL;
+    r = env->txn_begin(env, 0, &txn, 0); CKERR(r);
+    r = db->open(db, txn, "foo.db", 0, DB_BTREE, DB_CREATE, S_IRWXU+S_IRWXG+S_IRWXO); CKERR(r);
+    r = txn->commit(txn, 0);    CKERR(r);
+
+    // verify that stats include a new row inserted into the root
+    {
+        r = env->txn_begin(env, 0, &txn, 0); CKERR(r);
+        int key = 1; char val = 1;
+        DBT k,v;
+        r = db->put(db, txn, dbt_init(&k, &key, sizeof key), dbt_init(&v, &val, sizeof val), 0); CKERR(r);
+        r = txn->commit(txn, 0);    CKERR(r);
+
+        DB_BTREE_STAT64 s;
+        r = db->stat64(db, NULL, &s); CKERR(r);
+        assert(s.bt_nkeys == 1 && s.bt_dsize == sizeof key + sizeof val);
+
+        r = db->close(db, 0);     CKERR(r);
+
+        r = db_create(&db, env, 0); CKERR(r);
+        r = env->txn_begin(env, 0, &txn, 0); CKERR(r);
+        r = db->open(db, txn, "foo.db", 0, DB_BTREE, DB_CREATE, S_IRWXU+S_IRWXG+S_IRWXO); CKERR(r);
+        r = txn->commit(txn, 0);    CKERR(r);
+
+        r = db->stat64(db, NULL, &s); CKERR(r);
+        assert(s.bt_nkeys == 1 && s.bt_dsize == sizeof key + sizeof val);
+    }
+
+    // verify that stats are updated by row overwrite in the root
+    {
+        r = env->txn_begin(env, 0, &txn, 0); CKERR(r);
+        int key = 1; int val = 2;
+        DBT k,v;
+        r = db->put(db, txn, dbt_init(&k, &key, sizeof key), dbt_init(&v, &val, sizeof val), 0); CKERR(r);
+        r = txn->commit(txn, 0);    CKERR(r);
+
+        DB_BTREE_STAT64 s;
+        r = db->stat64(db, NULL, &s); CKERR(r);
+        assert(s.bt_nkeys == 1 && s.bt_dsize == sizeof key + sizeof val);
+
+        r = db->close(db, 0);     CKERR(r);
+
+        r = db_create(&db, env, 0); CKERR(r);
+        r = env->txn_begin(env, 0, &txn, 0); CKERR(r);
+        r = db->open(db, txn, "foo.db", 0, DB_BTREE, DB_CREATE, S_IRWXU+S_IRWXG+S_IRWXO); CKERR(r);
+        r = txn->commit(txn, 0);    CKERR(r);
+
+        r = db->stat64(db, NULL, &s); CKERR(r);
+        assert(s.bt_nkeys == 1 && s.bt_dsize == sizeof key + sizeof val);
+    }
+
+    // verify that stats are updated by row deletion in the root
+    {
+        r = env->txn_begin(env, 0, &txn, 0); CKERR(r);
+        int key = 1;
+        DBT k;
+        r = db->del(db, txn, dbt_init(&k, &key, sizeof key), 0); CKERR(r);
+        r = txn->commit(txn, 0);    CKERR(r);
+
+        DB_BTREE_STAT64 s;
+        r = db->stat64(db, NULL, &s); CKERR(r);
+        assert(s.bt_nkeys == 0 && s.bt_dsize == 0);
+
+        r = db->close(db, 0);     CKERR(r);
+
+        r = db_create(&db, env, 0); CKERR(r);
+        r = env->txn_begin(env, 0, &txn, 0); CKERR(r);
+        r = db->open(db, txn, "foo.db", 0, DB_BTREE, DB_CREATE, S_IRWXU+S_IRWXG+S_IRWXO); CKERR(r);
+        r = txn->commit(txn, 0);    CKERR(r);
+
+        r = db->stat64(db, NULL, &s); CKERR(r);
+        assert(s.bt_nkeys == 0 && s.bt_dsize == 0);
+    }
+
+    // verify update of non-existing key inserts a row
+    {
+        r = env->txn_begin(env, 0, &txn, 0); CKERR(r);
+        int key = 1; char val = 1;
+        DBT k = { .data = &key, .size = sizeof key };
+        DBT e = { .data = &val, .size = sizeof val };
+        r = db->update(db, txn, &k, &e, 0); CKERR(r);
+        r = txn->commit(txn, 0);    CKERR(r);
+
+        DB_BTREE_STAT64 s;
+        r = db->stat64(db, NULL, &s); CKERR(r);
+        assert(s.bt_nkeys == 1 && s.bt_dsize == sizeof key + sizeof val);
+
+        r = db->close(db, 0);     CKERR(r);
+
+        r = db_create(&db, env, 0); CKERR(r);
+        r = env->txn_begin(env, 0, &txn, 0); CKERR(r);
+        r = db->open(db, txn, "foo.db", 0, DB_BTREE, DB_CREATE, S_IRWXU+S_IRWXG+S_IRWXO); CKERR(r);
+        r = txn->commit(txn, 0);    CKERR(r);
+
+        r = db->stat64(db, NULL, &s); CKERR(r);
+        assert(s.bt_nkeys == 1 && s.bt_dsize == sizeof key + sizeof val);
+    }
+
+    // verify update callback overwrites the row
+    {
+        r = env->txn_begin(env, 0, &txn, 0); CKERR(r);
+        int key = 1; int val = 2;
+        DBT k = { .data = &key, .size = sizeof key };
+        DBT e = { .data = &val, .size = sizeof val };
+        r = db->update(db, txn, &k, &e, 0); CKERR(r);
+        r = txn->commit(txn, 0);    CKERR(r);
+
+        DB_BTREE_STAT64 s;
+        r = db->stat64(db, NULL, &s); CKERR(r);
+        assert(s.bt_nkeys == 1 && s.bt_dsize == sizeof key + sizeof val);
+
+        r = db->close(db, 0);     CKERR(r);
+
+        r = db_create(&db, env, 0); CKERR(r);
+        r = env->txn_begin(env, 0, &txn, 0); CKERR(r);
+        r = db->open(db, txn, "foo.db", 0, DB_BTREE, DB_CREATE, S_IRWXU+S_IRWXG+S_IRWXO); CKERR(r);
+        r = txn->commit(txn, 0);    CKERR(r);
+
+        r = db->stat64(db, NULL, &s); CKERR(r);
+        assert(s.bt_nkeys == 1 && s.bt_dsize == sizeof key + sizeof val);
+    }
+
+    // verify update callback deletes the row
+    {
+        // insert a new row
+        r = env->txn_begin(env, 0, &txn, 0); CKERR(r);
+        int key = 1; char val[42]; memset(val, 0, sizeof val);
+        DBT k = { .data = &key, .size = sizeof key };
+        DBT e = { .data = &val, .size = sizeof val };
+        r = db->update(db, txn, &k, &e, 0); CKERR(r);
+        r = txn->commit(txn, 0);    CKERR(r);
+
+        DB_BTREE_STAT64 s;
+        r = db->stat64(db, NULL, &s); CKERR(r);
+        assert(s.bt_nkeys == 1 && s.bt_dsize == sizeof key + sizeof val);
+
+        // update it again, this should delete the row
+        r = env->txn_begin(env, 0, &txn, 0); CKERR(r);
+        r = db->update(db, txn, &k, &e, 0); CKERR(r);
+        r = txn->commit(txn, 0);    CKERR(r);
+
+        r = db->stat64(db, NULL, &s); CKERR(r);
+        assert(s.bt_nkeys == 0 && s.bt_dsize == 0);
+
+        r = db->close(db, 0);     CKERR(r);
+
+        r = db_create(&db, env, 0); CKERR(r);
+        r = env->txn_begin(env, 0, &txn, 0); CKERR(r);
+        r = db->open(db, txn, "foo.db", 0, DB_BTREE, DB_CREATE, S_IRWXU+S_IRWXG+S_IRWXO); CKERR(r);
+        r = txn->commit(txn, 0);    CKERR(r);
+
+        r = db->stat64(db, NULL, &s); CKERR(r);
+        assert(s.bt_nkeys == 0 && s.bt_dsize == 0);
+    }
+
+    r = db->close(db, 0);     CKERR(r);
+
+    r = env->close(env, 0);   CKERR(r);
+}
+
+int
+test_main (int argc , char * const argv[]) {
+    parse_args(argc, argv);
+    run_test();
+    return 0;
+}

src/tests/stat64.c

@@ -36,7 +36,6 @@ test_stat64 (unsigned int N) {
     r=env->txn_begin(env, 0, &txn, 0);  CKERR(r);
 
     // insert sequential keys into the databases
-
     unsigned int i;
     u_int64_t dsize=0;
     for (i=0; i<N; i++) {
@@ -71,14 +70,14 @@ test_stat64 (unsigned int N) {
             printf("fsize=%" PRIu64 "\n", s.bt_fsize);
             printf("expected dsize=%" PRIu64 "\n", dsize); 
         }
-        assert(s.bt_nkeys <= 4*N);      // This can probably be tightened up when we fix #3995.
-        assert(s.bt_ndata <= 4*N);      // This can probably be tightened up when we fix #3995.
-        assert(s.bt_dsize <= 16*dsize); // This can probably be tightened up when we fix #3995.
+        assert(0 < s.bt_nkeys && s.bt_nkeys <= N);
+        assert(s.bt_ndata == s.bt_nkeys);
+        assert(0 < s.bt_dsize && s.bt_dsize <= dsize);
         assert(s.bt_fsize > N);
     }
     r=txn->commit(txn, 0); CKERR(r);
 
-    // get the last row, this forces the root estimates to be updated
+    // get the last row, this forces the root estimates to be updated.
     {
         r = env->txn_begin(env, NULL, &txn, 0); CKERR(r);
         DBC *c = NULL;
@@ -105,9 +104,9 @@ test_stat64 (unsigned int N) {
             printf("fsize=%" PRIu64 "\n", s.bt_fsize);
             printf("expected dsize=%" PRIu64 "\n", dsize); 
         }
-        assert(s.bt_nkeys <= 4*N);
-        assert(s.bt_ndata <= 4*N);
-        assert(s.bt_dsize <= 16*dsize); // This can probably be tightened up when we fix #3995.
+        assert(0 < s.bt_nkeys && s.bt_nkeys <= N);
+        assert(s.bt_ndata == s.bt_nkeys);
+        assert(0 < s.bt_dsize && s.bt_dsize <= dsize);
         assert(s.bt_fsize > N);
     }
     r=txn->commit(txn, 0); CKERR(r);

src/ydb.c

@@ -2083,8 +2083,17 @@ env_get_engine_status(DB_ENV * env, ENGINE_STATUS * engstat, char * env_panic_st
             engstat->disk_flush_nonleaf = brt_stat.disk_flush_nonleaf; 
             engstat->disk_flush_leaf_for_checkpoint = brt_stat.disk_flush_leaf_for_checkpoint; 
             engstat->disk_flush_nonleaf_for_checkpoint = brt_stat.disk_flush_nonleaf_for_checkpoint; 
+            engstat->create_leaf = brt_stat.create_leaf;
+            engstat->create_nonleaf = brt_stat.create_nonleaf;
             engstat->destroy_leaf = brt_stat.destroy_leaf;
             engstat->destroy_nonleaf = brt_stat.destroy_nonleaf;
+            engstat->split_leaf = brt_stat.split_leaf;
+            engstat->split_nonleaf = brt_stat.split_nonleaf;
+            engstat->merge_leaf = brt_stat.merge_leaf;
+            engstat->merge_nonleaf = brt_stat.merge_nonleaf;
+            engstat->dirty_leaf = brt_stat.dirty_leaf;
+            engstat->dirty_nonleaf = brt_stat.dirty_nonleaf;
+            engstat->balance_leaf = brt_stat.balance_leaf;
             engstat->msg_bytes_in = brt_stat.msg_bytes_in;
             engstat->msg_bytes_out = brt_stat.msg_bytes_out;
             engstat->msg_bytes_curr = brt_stat.msg_bytes_curr;
@@ -2364,8 +2373,17 @@ env_get_engine_status_text(DB_ENV * env, char * buff, int bufsiz) {
         n += snprintf(buff + n, bufsiz - n, "disk_flush_nonleaf               %"PRIu64"\n", engstat.disk_flush_nonleaf); 
         n += snprintf(buff + n, bufsiz - n, "disk_flush_leaf_for_checkpoint   %"PRIu64"\n", engstat.disk_flush_leaf_for_checkpoint); 
         n += snprintf(buff + n, bufsiz - n, "disk_flush_nonleaf_for_checkpoint %"PRIu64"\n", engstat.disk_flush_nonleaf_for_checkpoint); 
+        n += snprintf(buff + n, bufsiz - n, "create_leaf                      %"PRIu64"\n", engstat.create_leaf); 
+        n += snprintf(buff + n, bufsiz - n, "create_nonleaf                   %"PRIu64"\n", engstat.create_nonleaf); 
         n += snprintf(buff + n, bufsiz - n, "destroy_leaf                     %"PRIu64"\n", engstat.destroy_leaf); 
         n += snprintf(buff + n, bufsiz - n, "destroy_nonleaf                  %"PRIu64"\n", engstat.destroy_nonleaf); 
+        n += snprintf(buff + n, bufsiz - n, "split_leaf                       %"PRIu64"\n", engstat.split_leaf); 
+        n += snprintf(buff + n, bufsiz - n, "split_nonleaf                    %"PRIu64"\n", engstat.split_nonleaf); 
+        n += snprintf(buff + n, bufsiz - n, "merge_leaf                       %"PRIu64"\n", engstat.merge_leaf); 
+        n += snprintf(buff + n, bufsiz - n, "merge_nonleaf                    %"PRIu64"\n", engstat.merge_nonleaf); 
+        n += snprintf(buff + n, bufsiz - n, "dirty_leaf                       %"PRIu64"\n", engstat.dirty_leaf); 
+        n += snprintf(buff + n, bufsiz - n, "dirty_nonleaf                    %"PRIu64"\n", engstat.dirty_nonleaf); 
+        n += snprintf(buff + n, bufsiz - n, "balance_leaf                     %"PRIu64"\n", engstat.balance_leaf); 
         n += snprintf(buff + n, bufsiz - n, "msg_bytes_in                     %"PRIu64"\n", engstat.msg_bytes_in); 
         n += snprintf(buff + n, bufsiz - n, "msg_bytes_out                    %"PRIu64"\n", engstat.msg_bytes_out); 
         n += snprintf(buff + n, bufsiz - n, "msg_bytes_curr                   %"PRIu64"\n", engstat.msg_bytes_curr);