Fixes #1367.

git-svn-id: file:///svn/toku/tokudb.1032b+1343@8578 c7de825b-a66e-492c-adef-691d508d4ae1

newbrt/brt.c

@@ -2554,8 +2554,13 @@ int toku_brt_insert (BRT brt, DBT *key, DBT *val, TOKUTXN txn)
     if (txn && (brt->txn_that_created != toku_txn_get_txnid(txn))) {
 	toku_cachefile_refup(brt->cf);
 	BYTESTRING keybs  = {key->size, toku_memdup_in_rollback(txn, key->data, key->size)};
-	BYTESTRING databs = {val->size, toku_memdup_in_rollback(txn, val->data, val->size)};
-	r = toku_logger_save_rollback_cmdinsert(txn, toku_txn_get_txnid(txn), toku_cachefile_filenum(brt->cf), keybs, databs);
+	int need_data = (brt->flags&TOKU_DB_DUPSORT)!=0; // dupsorts don't need the data part
+	if (need_data) {
+	    BYTESTRING databs = {val->size, toku_memdup_in_rollback(txn, val->data, val->size)};
+	    r = toku_logger_save_rollback_cmdinsertboth(txn, toku_txn_get_txnid(txn), toku_cachefile_filenum(brt->cf), keybs, databs);
+	} else {
+	    r = toku_logger_save_rollback_cmdinsert    (txn, toku_txn_get_txnid(txn), toku_cachefile_filenum(brt->cf), keybs);
+	}
 	if (r!=0) return r;
 	r = toku_txn_note_brt(txn, brt);
 	if (r!=0) return r;

newbrt/logformat.c

@@ -44,7 +44,12 @@ const struct logtype rollbacks[] = {
 			{"FILENUM", "filenum", 0},
 			{"BYTESTRING", "fname", 0},
 			NULLFIELD}},
+    // cmdinsert is used to insert a key-value pair into a NODUP DB.  For rollback we don't need the data.
     {"cmdinsert", 'i', FA{{"TXNID", "xid", 0},
+			  {"FILENUM", "filenum", 0},
+			  {"BYTESTRING", "key", 0},
+			  NULLFIELD}},
+    {"cmdinsertboth", 'I', FA{{"TXNID", "xid", 0},
 			  {"FILENUM", "filenum", 0},
 			  {"BYTESTRING", "key", 0},
 			  {"BYTESTRING", "data", 0},
@@ -58,7 +63,7 @@ const struct logtype rollbacks[] = {
 			  {"FILENUM", "filenum", 0},
 			  {"BYTESTRING", "key", 0},
 			  NULLFIELD}},
-    {"rollinclude", 'I', FA{{"BYTESTRING", "fname", 0},
+    {"rollinclude", 'r', FA{{"BYTESTRING", "fname", 0},
                             NULLFIELD}},
 //    {"fclose", 'c', FA{{"FILENUM", "filenum", 0},
 //		       {"BYTESTRING", "fname", 0},

newbrt/roll.c

@@ -85,7 +85,16 @@ static int do_nothing_with_filenum(TOKUTXN txn, FILENUM filenum) {
 }
 
 
-int toku_commit_cmdinsert (TXNID xid, FILENUM filenum, BYTESTRING key,BYTESTRING data,TOKUTXN txn) {
+int toku_commit_cmdinsert (TXNID xid, FILENUM filenum, BYTESTRING key, TOKUTXN txn) {
+#if TOKU_DO_COMMIT_CMD_INSERT
+    return do_insertion (BRT_COMMIT_ANY, xid, filenum, key, 0, txn);
+#else
+    xid = xid; key = key;
+    return do_nothing_with_filenum(txn, filenum);
+#endif
+}
+
+int toku_commit_cmdinsertboth (TXNID xid, FILENUM filenum, BYTESTRING key, BYTESTRING data, TOKUTXN txn) {
 #if TOKU_DO_COMMIT_CMD_INSERT
     return do_insertion (BRT_COMMIT_BOTH, xid, filenum, key, &data, txn);
 #else
@@ -94,7 +103,11 @@ int toku_commit_cmdinsert (TXNID xid, FILENUM filenum, BYTESTRING key,BYTESTRING
 #endif
 }
 
-int toku_rollback_cmdinsert (TXNID xid, FILENUM filenum, BYTESTRING key,BYTESTRING data,TOKUTXN txn) {
+int toku_rollback_cmdinsert (TXNID xid, FILENUM filenum, BYTESTRING key, TOKUTXN txn) {
+    return do_insertion (BRT_ABORT_ANY, xid, filenum, key, 0, txn);
+}
+
+int toku_rollback_cmdinsertboth (TXNID xid, FILENUM filenum, BYTESTRING key,BYTESTRING data,TOKUTXN txn) {
     return do_insertion (BRT_ABORT_BOTH, xid, filenum, key, &data, txn);
 }