delete aborts work right now. Addresses #27.

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

newbrt/Makefile

@@ -8,7 +8,7 @@
 
 # GCOV_FLAGS = -fprofile-arcs -ftest-coverage
 # PROF_FLAGS = -pg
-OPTFLAGS = -O2
+# OPTFLAGS = -O2
 
 ifeq ($(VERBOSE),2)
 VERBVERBOSE=-v

newbrt/pma.c

@@ -921,23 +921,26 @@ int toku_pma_insert_or_replace (PMA pma, DBT *k, DBT *v,
     *replaced_v_size = -1;
     //printf("%s:%d txn=%p\n", __FILE__, __LINE__, txn);
  logit_and_update_fingerprint:
-    {
-	TOKUTXN txn;
-	if ((r=toku_txnid2txn(logger, xid, &txn))) return r;
-	const BYTESTRING key  = { k->size, toku_memdup(k->data, k->size) };
-	const BYTESTRING data = { v->size, toku_memdup(v->data, v->size) };
-	if ((r = toku_logger_save_rollback_insertatleaf(txn, pma->filenum, key, data))) {
-	     toku_free(key.data); toku_free(data.data);
-	     return r;
+    r=0;
+    if (logger) {
+	{
+	    TOKUTXN txn;
+	    if ((r=toku_txnid2txn(logger, xid, &txn))) return r;
+	    const BYTESTRING key  = { k->size, toku_memdup(k->data, k->size) };
+	    const BYTESTRING data = { v->size, toku_memdup(v->data, v->size) };
+	    if ((r = toku_logger_save_rollback_insertatleaf(txn, pma->filenum, key, data))) {
+		toku_free(key.data); toku_free(data.data);
+		return r;
+	    }
+	}
+	{
+	    const BYTESTRING key  = { k->size, k->data };
+	    const BYTESTRING data = { v->size, k->data };
+	    r = toku_log_insertinleaf (logger, xid, pma->filenum, diskoff, idx, key, data);
+	    if (logger && node_lsn) *node_lsn = toku_logger_last_lsn(logger);
+	    if (r!=0) return r;
+	    /* We don't record the insert here for rollback.  The insert should have been logged at the top-level. */
 	}
-    }
-    {
-	const BYTESTRING key  = { k->size, k->data };
-	const BYTESTRING data = { v->size, k->data };
-	r = toku_log_insertinleaf (logger, xid, pma->filenum, diskoff, idx, key, data);
-	if (logger && node_lsn) *node_lsn = toku_logger_last_lsn(logger);
-	if (r!=0) return r;
-	/* We don't record the insert here for rollback.  The insert should have been logged at the top-level. */
     }
     *fingerprint += rand4fingerprint*toku_calccrc32_kvpair(k->data, k->size, v->data, v->size);
     return r;

newbrt/roll.c

@@ -420,7 +420,7 @@ int toku_rollback_deleteatleaf (FILENUM filenum, BYTESTRING key, BYTESTRING data
     r = toku_brt_insert(brt,
 			toku_fill_dbt(&key_dbt, key.data, key.len),
 			toku_fill_dbt(&data_dbt, data.data, data.len),
-			txn);
+			0); // Do the insertion unconditionally
     return r;
 }
 

src/tests/test_txn_abort5a.c

@@ -10,7 +10,7 @@
 
 
 void test_txn_abort(int n) {
-    if (verbose) printf("test_txn_abort:%d\n", n);
+    if (verbose>1) printf("%s %s:%d\n", __FILE__, __FUNCTION__, n);
 
     system("rm -rf " DIR);
     mkdir(DIR, 0777);
@@ -37,6 +37,7 @@ void test_txn_abort(int n) {
     for (i=0; i<n; i++) {
         DBT key, val;
 	int i2=htonl(i*2);
+	if (verbose>2) printf("put %d\n", i*2);
         r = db->put(db, txn, dbt_init(&key, &i2, sizeof i2), dbt_init(&val, &i, sizeof i), 0); 
         if (r != 0) printf("%s:%d:%d:%s\n", __FILE__, __LINE__, r, db_strerror(r));
         assert(r == 0);
@@ -47,6 +48,7 @@ void test_txn_abort(int n) {
     for (i=0; i<n; i++) {
         DBT key;
 	int i2=htonl(i*2);
+	if (verbose>2) printf("del %d\n", i*2);
         r = db->del(db, txn, dbt_init(&key, &i2, sizeof i2), 0);
         if (r != 0) printf("%s:%d:%d:%s\n", __FILE__, __LINE__, r, db_strerror(r));
         assert(r == 0);
@@ -63,7 +65,7 @@ void test_txn_abort(int n) {
     for (i=0; 1; i++) {
 	r = cursor->c_get(cursor, &key, &val, DB_NEXT);
 	if (r!=0) break;
-	printf("%d present\n", ntohl(*(int*)key.data));
+	if (verbose>2) printf("%d present\n", ntohl(*(int*)key.data));
 	assert(key.size==4);
 	assert(ntohl(*(int*)key.data)==2*i);
     }
@@ -84,7 +86,10 @@ int main(int argc, char *argv[]) {
             continue;
         }
     }
+    if (verbose>0) printf("%s", __FILE__); if (verbose>1) printf("\n");
     for (i=1; i<100; i++) 
         test_txn_abort(i);
+    if (verbose>1) printf("%s OK\n", __FILE__);
+    if (verbose>0) printf(" OK\n");
     return 0;
 }