Log brtnode (but dumper cannot understand it yet.) Addresses #11

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

newbrt/brt.c

@@ -1427,7 +1427,7 @@ int brt_create_cachetable(CACHETABLE *ct, long cachesize, LSN initial_lsn, TOKUL
     return toku_create_cachetable(ct, cachesize, initial_lsn, logger);
 }
 
-static int setup_brt_root_node (BRT t, DISKOFF offset) {
+static int setup_brt_root_node (BRT t, DISKOFF offset, TOKUTXN txn) {
     int r;
     TAGMALLOC(BRTNODE, node);
     assert(node);
@@ -1451,6 +1451,7 @@ static int setup_brt_root_node (BRT t, DISKOFF offset) {
     //printf("%s:%d created %lld\n", __FILE__, __LINE__, node->thisnodename);
     toku_verify_counts(node);
 //    verify_local_fingerprint_nonleaf(node);
+    tokulogger_log_newbrtnode(txn, toku_cachefile_filenum(t->cf), offset, 0, t->h->nodesize, (t->flags&TOKU_DB_DUPSORT)!=0, node->rand4fingerprint);
     r=toku_cachetable_unpin(t->cf, node->thisnodename, node->dirty, brtnode_size(node));
     if (r!=0) {
 	toku_free(node);
@@ -1576,9 +1577,9 @@ int brt_open(BRT t, const char *fname, const char *fname_in_env, const char *dbn
 		t->h->names=0;
 		t->h->roots=0;
 	    }
-	    if ((r=setup_brt_root_node(t, t->nodesize))!=0) { died6: if (dbname) goto died5; else goto died2;	}
-	    if ((r=toku_cachetable_put(t->cf, 0, t->h, 0, brtheader_flush_callback, brtheader_fetch_callback, 0))) { goto died6; }
 	    if ((r=tokulogger_log_header(txn, toku_cachefile_filenum(t->cf), t->h)))                               { goto died6; }
+	    if ((r=setup_brt_root_node(t, t->nodesize, txn))!=0) { died6: if (dbname) goto died5; else goto died2;	}
+	    if ((r=toku_cachetable_put(t->cf, 0, t->h, 0, brtheader_flush_callback, brtheader_fetch_callback, 0))) { goto died6; }
 	} else {
 	    int i;
 	    assert(r==0);
@@ -1601,7 +1602,7 @@ int brt_open(BRT t, const char *fname, const char *fname_in_env, const char *dbn
 	    //printf("%s:%d t=%p\n", __FILE__, __LINE__, t);
 	    t->h->roots[t->h->n_named_roots-1] = malloc_diskblock_header_is_in_memory(t, t->h->nodesize);
 	    t->h->dirty = 1;
-	    if ((r=setup_brt_root_node(t, t->h->roots[t->h->n_named_roots-1]))!=0) goto died1;
+	    if ((r=setup_brt_root_node(t, t->h->roots[t->h->n_named_roots-1], txn))!=0) goto died1;
 	}
     } else {
 	if ((r = toku_read_and_pin_brt_header(t->cf, &t->h))!=0) goto died1;

newbrt/log-internal.h

@@ -24,6 +24,7 @@ enum lt_command {
     LT_FCREATE                  = 'F',
     LT_FHEADER                  = 'H',
     LT_INSERT_WITH_NO_OVERWRITE = 'I',
+    LT_NEWBRTNODE               = 'N',
     LT_FOPEN                    = 'O',
     LT_CHECKPOINT               = 'P',
     LT_BLOCK_RENAME             = 'R',

newbrt/log.c

@@ -357,6 +357,35 @@ int tokulogger_log_header (TOKUTXN txn, FILENUM filenum, struct brt_header *h) {
     return r;
 }
 
+int tokulogger_log_newbrtnode (TOKUTXN txn, FILENUM filenum, DISKOFF offset, u_int32_t height, u_int32_t nodesize, char is_dup_sort_mode, u_int32_t rand4fingerprint) {
+    if (txn==0) return 0;
+    int buflen=(1+
+		+ 8 // lsn
+		+ 8 // txnid
+		+ 4 // filenum
+		+ 8 // diskoff
+		+ 4 // height
+		+ 4 // nodesize
+		+ 1 // is_dup_sort_mode
+		+ 4 // rand4fingerprint
+		+ 8 // crc & len
+		);
+    unsigned char buf[buflen];
+    struct wbuf wbuf;
+    wbuf_init (&wbuf, buf, buflen);
+    wbuf_char(&wbuf, LT_NEWBRTNODE);
+    wbuf_lsn    (&wbuf, txn->logger->lsn);
+    txn->logger->lsn.lsn++;
+    wbuf_txnid(&wbuf, txn->txnid64);
+    wbuf_filenum(&wbuf, filenum);
+    wbuf_diskoff(&wbuf, offset);
+    wbuf_int(&wbuf, height);
+    wbuf_int(&wbuf, nodesize);
+    wbuf_char(&wbuf, is_dup_sort_mode);
+    wbuf_int(&wbuf, rand4fingerprint);
+    return tokulogger_finish(txn->logger, &wbuf);
+}
+
 /*
 int brtenv_checkpoint (BRTENV env) {
     init the checkpointing lock

newbrt/log.h

@@ -24,4 +24,6 @@ int tokulogger_log_unlink (TOKUTXN, const char */*fname*/);
 
 int tokulogger_log_header (TOKUTXN, FILENUM, struct brt_header *);
 
+int tokulogger_log_newbrtnode (TOKUTXN txn, FILENUM filenum, DISKOFF offset, u_int32_t height, u_int32_t nodesize, char is_dup_sort_mode, u_int32_t rand4fingerprint);
+
 #endif