Log node creation more accurately. Addresses #27.

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

newbrt/brt.c

@@ -1613,16 +1613,17 @@ static int brt_init_new_root(BRT brt, BRTNODE nodea, BRTNODE nodeb, DBT splitk,
     //verify_local_fingerprint_nonleaf(nodea);
     //verify_local_fingerprint_nonleaf(nodeb);
     r=toku_log_newbrtnode(txn, toku_txn_get_txnid(txn), toku_cachefile_filenum(brt->cf), newroot_diskoff, new_height, new_nodesize, (brt->flags&TOKU_DB_DUPSORT)!=0, newroot->rand4fingerprint);
+    r=toku_log_addchild(txn, toku_txn_get_txnid(txn), toku_cachefile_filenum(brt->cf), newroot_diskoff, 0);
     if (r!=0) return r;
-    r=toku_log_insertchild(txn, toku_txn_get_txnid(txn), toku_cachefile_filenum(brt->cf), newroot_diskoff, 0, nodea->thisnodename);
+    r=toku_log_setchild(txn, toku_txn_get_txnid(txn), toku_cachefile_filenum(brt->cf), newroot_diskoff, 0, nodea->thisnodename);
     if (r!=0) return r;
-    r=toku_log_insertchild(txn, toku_txn_get_txnid(txn), toku_cachefile_filenum(brt->cf), newroot_diskoff, 1, nodeb->thisnodename);
+    r=toku_log_setchild(txn, toku_txn_get_txnid(txn), toku_cachefile_filenum(brt->cf), newroot_diskoff, 1, nodeb->thisnodename);
     if (r!=0) return r;
     {
 	BYTESTRING bs;
 	bs.len = splitk.size;
 	bs.data = splitk.data;
-	r=toku_log_insertpivot(txn, toku_txn_get_txnid(txn), toku_cachefile_filenum(brt->cf), newroot_diskoff, 0, bs);
+	r=toku_log_setpivot(txn, toku_txn_get_txnid(txn), toku_cachefile_filenum(brt->cf), newroot_diskoff, 0, bs);
 	if (r!=0) return r;
     }
     r=toku_cachetable_unpin(brt->cf, nodea->thisnodename, nodea->dirty, brtnode_size(nodea)); 

newbrt/logformat.c

@@ -60,18 +60,23 @@ const struct logtype logtypes[] = {
 			   {"u_int8_t", "is_dup_sort", 0},
 			   {"u_int32_t", "rand4fingerprint", 0},
 			   NULLFIELD}},
-    {"insertchild", 'i', FA{{"TXNID",   "txnid", 0},
-			    {"FILENUM", "filenum", 0},
-			    {"DISKOFF", "diskoff", 0},
-			    {"u_int32_t", "childnum", 0},
-			    {"DISKOFF",  "child", 0},
-			    NULLFIELD}},
-    {"insertpivot", 'k', FA{{"TXNID",   "txnid", 0},
-			    {"FILENUM", "filenum", 0},
-			    {"DISKOFF", "diskoff", 0},
-			    {"u_int32_t", "childnum", 0},
-			    {"BYTESTRING", "pivotkey", 0},
-			    NULLFIELD}},
+    {"addchild", 'c', FA{{"TXNID",   "txnid", 0},
+			 {"FILENUM", "filenum", 0},
+			 {"DISKOFF", "diskoff", 0},
+			 {"u_int32_t", "childnum", 0}, // children scoot over
+			 NULLFIELD}},
+    {"setchild", 'i', FA{{"TXNID",     "txnid", 0},
+			 {"FILENUM",   "filenum", 0},
+			 {"DISKOFF",   "diskoff", 0},
+			 {"u_int32_t", "childnum", 0},
+			 {"DISKOFF",   "child", 0},
+			 NULLFIELD}},
+    {"setpivot", 'k', FA{{"TXNID",   "txnid", 0},
+			 {"FILENUM", "filenum", 0},
+			 {"DISKOFF", "diskoff", 0},
+			 {"u_int32_t", "childnum", 0},
+			 {"BYTESTRING", "pivotkey", 0},
+			 NULLFIELD}},
     {"fopen",   'O', FA{{"TXNID",      "txnid", 0},
 			{"BYTESTRING", "fname", 0},
 			{"FILENUM",    "filenum", 0},

newbrt/roll.c

@@ -203,10 +203,27 @@ int toku_rollback_newbrtnode (struct logtype_newbrtnode *le, TOKUTXN txn) {
 }
 
 
-int toku_rollback_insertchild (struct logtype_insertchild *le, TOKUTXN txn) ABORTIT
-void toku_recover_insertchild (struct logtype_insertchild *le) { le=le; abort(); }
-int toku_rollback_insertpivot (struct logtype_insertpivot *le, TOKUTXN txn) ABORTIT
-void toku_recover_insertpivot (struct logtype_insertpivot *le) { le=le; abort(); }
+int toku_rollback_addchild (struct logtype_addchild *le, TOKUTXN txn) ABORTIT
+void toku_recover_addchild (struct logtype_addchild *le) { le=le; abort(); }
+
+int toku_rollback_setchild (struct logtype_setchild *le, TOKUTXN txn) ABORTIT
+void toku_recover_setchild (struct logtype_setchild *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);
+    assert(le->childnum < (unsigned)node->u.n.n_children);
+    node->u.n.children[le->childnum] = le->child;
+    r = toku_cachetable_unpin(pair->cf, le->diskoff, 1, toku_serialize_brtnode_size(node));
+    assert(r==0);
+}
+int toku_rollback_setpivot (struct logtype_setpivot *le, TOKUTXN txn) ABORTIT
+void toku_recover_setpivot (struct logtype_setpivot *le) { le=le; abort(); }
 
 void toku_recover_fopen (struct logtype_fopen *c) {
     char *fname = fixup_fname(&c->fname);