Log the fingerprints correctly. Addresses #27.

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

newbrt/brt-internal.h

@@ -49,8 +49,15 @@ struct brtnode {
     unsigned int nodesize;
     unsigned int flags;
     DISKOFF thisnodename;   // The size of the node allocated on disk.  Not all is necessarily in use.
-    LSN     disk_lsn;       // The LSN as of the most recent version on disk.
-    LSN     log_lsn;        // The LSN as of the most recent log write. 
+    //  These two LSNs are used to decide when to make a copy of a node instead of overwriting it.
+    //  In the TOKULOGGER is a field called checkpoint_lsn which is the lsn of the most recent checkpoint
+    LSN     disk_lsn;       // The LSN as of the most recent version on disk.  (Updated by brt-serialize)  This lsn is saved in the node.
+    LSN     log_lsn;        // The LSN of the youngest log entry that affects the current in-memory state.   The log write may not have actually made it to disk.  This lsn is not saved in disk (since the two lsns are the same for any node not in main memory.)
+    //  The checkpointing works as follows:
+    //      When we unpin a node: if it is dirty and disk_lsn<checkpoint_lsn then we need to make a new copy.
+    //      When we checkpoint:  Create a checkpoint record, and cause every dirty node to be written to disk.  The new checkpoint record is *not* incorporated into the disk_lsn of the written nodes.
+    //      While we are checkpointing, someone may modify a dirty node that has not yet been written.   In that case, when we unpin the node, we make the new copy (because the disk_lsn<checkpoint_lsn), just as we would usually.
+    //
     int     layout_version; // What version of the data structure?
     int    height; /* height is always >= 0.  0 for leaf, >0 for nonleaf. */
     u_int32_t rand4fingerprint;

newbrt/brtdump.c

@@ -52,7 +52,7 @@ void dump_node (int f, DISKOFF off, struct brt_header *h) {
     printf(" flags       =%u\n", n->flags);
     printf(" thisnodename=%lld\n", n->thisnodename);
     printf(" disk_lsn    =%lld\n", n->disk_lsn.lsn);
-    printf(" log_lsn     =%lld\n", n->log_lsn.lsn);
+    //printf(" log_lsn     =%lld\n", n->log_lsn.lsn); // The log_lsn is a memory-only value.
     printf(" height      =%d\n",   n->height);
     printf(" rand4fp     =%08x\n", n->rand4fingerprint);
     printf(" localfp     =%08x\n", n->local_fingerprint);

newbrt/cachetable.c

@@ -415,8 +415,9 @@ int toku_cachetable_get_and_pin(CACHEFILE cachefile, CACHEKEY key, void**value,
 	int r;
 	LSN written_lsn;
 	WHEN_TRACE_CT(printf("%s:%d CT: fetch_callback(%lld...)\n", __FILE__, __LINE__, key));
-	if ((r=fetch_callback(cachefile, key, &toku_value, &size, extraargs, &written_lsn))) 
+	if ((r=fetch_callback(cachefile, key, &toku_value, &size, extraargs, &written_lsn))) {
             return r;
+	}
 	cachetable_insert_at(cachefile, hashit(t,key), key, toku_value, size, flush_callback, fetch_callback, extraargs, 0, written_lsn);
 	*value = toku_value;
         if (sizep)

newbrt/logformat.c

@@ -59,8 +59,15 @@ const struct logtype logtypes[] = {
 			   {"u_int32_t", "height", 0},
 			   {"u_int32_t", "nodesize", 0},
 			   {"u_int8_t", "is_dup_sort", 0},
-			   {"u_int32_t", "rand4fingerprint", 0},
+			   {"u_int32_t", "rand4fingerprint", "%08x"},
 			   NULLFIELD}},
+    {"changechildfingerprint", 'f', FA{{"TXNID",   "txnid", 0},
+				       {"FILENUM", "filenum", 0},
+				       {"DISKOFF", "diskoff", 0},
+				       {"u_int32_t", "childnum", 0},
+				       {"u_int32_t", "oldfingerprint", "%08x"},
+				       {"u_int32_t", "newfingerprint", "%08x"},
+				       NULLFIELD}},
     {"changeunnamedroot", 'u', FA{{"TXNID",   "txnid", 0},
 				 {"FILENUM", "filenum", 0},
 				 {"DISKOFF", "oldroot", 0},

newbrt/roll.c

@@ -169,7 +169,8 @@ void toku_recover_newbrtnode (struct logtype_newbrtnode *c) {
     TAGMALLOC(BRTNODE, n);
     n->nodesize     = c->nodesize;
     n->thisnodename = c->diskoff;
-    n->log_lsn = n->disk_lsn  = c->lsn; printf("%s:%d %p->disk_lsn=%"PRId64"\n", __FILE__, __LINE__, n, n->disk_lsn.lsn);
+    n->log_lsn = n->disk_lsn  = c->lsn;
+    //printf("%s:%d %p->disk_lsn=%"PRId64"\n", __FILE__, __LINE__, n, n->disk_lsn.lsn);
     n->layout_version = 1;
     n->height         = c->height;
     n->rand4fingerprint = c->rand4fingerprint;
@@ -285,6 +286,23 @@ void toku_recover_setpivot (struct logtype_setpivot *le) {
     toku_free(le->pivotkey.data);
 }
 
+void toku_recover_changechildfingerprint (struct logtype_changechildfingerprint *le) {
+    struct cf_pair *pair;
+    int r = find_cachefile(le->filenum, &pair);
+    assert(r==0);
+    void *node_v;
+    assert(pair->brt);
+    r = toku_cachetable_get_and_pin(pair->cf, le->diskoff, &node_v, NULL, toku_brtnode_flush_callback, toku_brtnode_fetch_callback, pair->brt);
+    assert(r==0);
+    BRTNODE node = node_v;
+    assert(node->height>0);
+    BRTNODE_CHILD_SUBTREE_FINGERPRINTS(node, le->childnum) = le->newfingerprint;
+    r = toku_cachetable_unpin(pair->cf, le->diskoff, 1, toku_serialize_brtnode_size(node));
+    assert(r==0);
+    
+}
+int toku_rollback_changechildfingerprint (struct logtype_changechildfingerprint *le, TOKUTXN txn) ABORTIT
+
 void toku_recover_fopen (struct logtype_fopen *c) {
     char *fname = fixup_fname(&c->fname);
     CACHEFILE cf;
@@ -465,3 +483,4 @@ void toku_recover_changeunusedmemory (struct logtype_changeunusedmemory *le)  {
     r = toku_unpin_brt_header(pair->brt);
 }
 int toku_rollback_changeunusedmemory (struct logtype_changeunusedmemory *le, TOKUTXN txn) ABORTIT
+