refs #5469, make fix

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

ft/roll.cc

@@ -408,7 +408,9 @@ toku_commit_load (FILENUM    old_filenum,
         lazy_assert(r == 0);
     }
 
-    toku_cachefile_unlink_on_close(old_cf);
+    if (!toku_cachefile_is_unlink_on_close(old_cf)) {
+        toku_cachefile_unlink_on_close(old_cf);
+    }
 done:
     return r;
 }

src/loader.cc

@@ -172,13 +172,8 @@ static int ft_loader_close_and_redirect(DB_LOADER *loader) {
     return r;
 }
 
-// loader_flags currently has three possible values:
-//   0                   use brt loader
-//   USE_PUTS            do not use brt loader, use log suppression mechanism (2440)
-//                       which results in recursive call here via toku_db_pre_acquire_table_lock()
-//   DB_PRELOCKED_WRITE  do not use brt loader, this is the recursive (inner) call via 
-//                       toku_db_pre_acquire_table_lock()
-int toku_loader_create_loader(DB_ENV *env, 
+
+static int create_loader(DB_ENV *env, 
                               DB_TXN *txn, 
                               DB_LOADER **blp, 
                               DB *src_db, 
@@ -186,7 +181,8 @@ int toku_loader_create_loader(DB_ENV *env,
                               DB *dbs[], 
                               uint32_t db_flags[/*N*/], 
                               uint32_t dbt_flags[/*N*/], 
-                              uint32_t loader_flags)
+                              uint32_t loader_flags,
+                              bool check_empty)
 {
     int rval;
     bool use_ft_loader = (loader_flags == 0); 
@@ -233,10 +229,12 @@ int toku_loader_create_loader(DB_ENV *env,
                 goto create_exit;
             }
         }
-        bool empty = toku_ft_is_empty_fast(dbs[i]->i->ft_handle);
-        if (!empty) {
-            rval = ENOTEMPTY;
-            goto create_exit;
+        if (check_empty) {
+            bool empty = toku_ft_is_empty_fast(dbs[i]->i->ft_handle);
+            if (!empty) {
+                rval = ENOTEMPTY;
+                goto create_exit;
+            }
         }
     }
 
@@ -316,6 +314,37 @@ int toku_loader_create_loader(DB_ENV *env,
     return rval;
 }
 
+// loader_flags currently has three possible values:
+//   0                   use brt loader
+//   USE_PUTS            do not use brt loader, use log suppression mechanism (2440)
+//                       which results in recursive call here via toku_db_pre_acquire_table_lock()
+//   DB_PRELOCKED_WRITE  do not use brt loader, this is the recursive (inner) call via 
+//                       toku_db_pre_acquire_table_lock()
+int toku_loader_create_loader(DB_ENV *env, 
+                              DB_TXN *txn, 
+                              DB_LOADER **blp, 
+                              DB *src_db, 
+                              int N, 
+                              DB *dbs[], 
+                              uint32_t db_flags[/*N*/], 
+                              uint32_t dbt_flags[/*N*/], 
+                              uint32_t loader_flags)
+{
+    return create_loader(
+        env, 
+        txn, 
+        blp, 
+        src_db, 
+        N, 
+        dbs, 
+        db_flags, 
+        dbt_flags, 
+        loader_flags, 
+        true
+        );
+}
+
+
 int toku_loader_set_poll_function(DB_LOADER *loader,
                                   int (*poll_func)(void *extra, float progress),
                                   void *poll_extra) 
@@ -434,7 +463,25 @@ int toku_loader_abort(DB_LOADER *loader)
 
     if (!(loader->i->loader_flags & LOADER_USE_PUTS) ) {
         r = toku_ft_loader_abort(loader->i->ft_loader, true);
+        lazy_assert_zero(r);
     }
+
+    DB_LOADER* tmp_loader = NULL;
+    r = create_loader(
+        loader->i->env,
+        loader->i->txn,
+        &tmp_loader,
+        loader->i->src_db,
+        loader->i->N,
+        loader->i->dbs,
+        loader->i->db_flags,
+        loader->i->dbt_flags,
+        0,
+        false
+        );
+    lazy_assert_zero(r);
+    r = toku_loader_close(tmp_loader);
+    
     free_loader(loader);
     return r;
 }

src/tests/CMakeLists.txt

@@ -214,6 +214,7 @@ if(BUILD_TESTING)
       test_4368
       test_4657
       test_5015
+      test_5469
       test-5138
       test938c
       test_abort1

src/tests/test_5469.cc

+/* -*- mode: C++; c-basic-offset: 4; indent-tabs-mode: nil -*- */
+// vim: ft=cpp:expandtab:ts=8:sw=4:softtabstop=4:
+#ident "$Id: test_cursor_2.cc 45903 2012-07-19 13:06:39Z leifwalsh $"
+#ident "Copyright (c) 2007 Tokutek Inc.  All rights reserved."
+#include "test.h"
+
+#include <stdio.h>
+#include <stdlib.h>
+
+#include <unistd.h>
+#include <memory.h>
+#include <errno.h>
+#include <sys/stat.h>
+#include <db.h>
+
+
+static int put_multiple_generate(DB *UU(dest_db), DB *UU(src_db), DBT *dest_key, DBT *dest_val, const DBT *src_key, const DBT *src_val) {
+    dbt_init(dest_key, src_key->data, src_key->size);
+    dbt_init(dest_val, src_val->data, src_val->size);
+    return 0;
+}
+
+static void
+test_loader_abort (bool use_puts, bool abort_loader) {
+    DB_ENV * env;
+    DB *db;
+    DB_TXN *txn;
+    DB_TXN* const null_txn = 0;
+    const char * const fname = "test.loader_abort.ft_handle";
+    int r;
+    r = system("rm -rf " ENVDIR);
+    CKERR(r);
+    toku_os_mkdir(ENVDIR, S_IRWXU+S_IRWXG+S_IRWXO);
+
+    DB_LOADER *loader;
+    uint32_t db_flags = 0;
+    uint32_t dbt_flags = 0;    
+    uint32_t loader_flags = use_puts ? LOADER_USE_PUTS : 0;
+    DBC* cursor = NULL;
+
+    /* create the dup database file */
+    r = db_env_create(&env, 0);        assert(r == 0);
+    env->set_errfile(env, stderr);
+    r = env->set_generate_row_callback_for_put(env, put_multiple_generate);
+    CKERR(r);
+    r = env->open(env, ENVDIR, DB_INIT_MPOOL|DB_CREATE|DB_THREAD |DB_INIT_LOCK|DB_INIT_LOG|DB_INIT_TXN|DB_PRIVATE, S_IRWXU+S_IRWXG+S_IRWXO); CKERR(r);
+    r = db_create(&db, env, 0); assert(r == 0);
+    db->set_errfile(db,stderr); // Turn off those annoying errors
+    r = db->open(db, null_txn, fname, "main", DB_BTREE, DB_CREATE, 0666); assert(r == 0);
+
+    r = env->txn_begin(env, NULL, &txn, 0);                                                               
+    CKERR(r);
+
+    r = env->create_loader(env, txn, &loader, db, 1, &db, &db_flags, &dbt_flags, loader_flags);
+    CKERR(r);
+
+    DBT key, val;
+    uint32_t k;
+    uint32_t v;
+    uint32_t num_elements = 2;
+    for (uint32_t i = 0; i < num_elements; i++) {
+        k = i;
+        v = i;
+        r = loader->put(
+            loader, 
+            dbt_init(&key, &k, sizeof k), 
+            dbt_init(&val, &v, sizeof v)
+            );
+        assert(r == 0); 
+    }
+    if (abort_loader) {
+        loader->abort(loader);
+    }
+    else {
+        loader->close(loader);
+    }
+    k = num_elements;
+    v = num_elements;
+    r = db->put(db, txn, dbt_init(&key, &k, sizeof k), dbt_init(&val, &v, sizeof v), 0);
+    r = txn->commit(txn, 0);
+    CKERR(r);
+
+    
+    r = env->txn_begin(env, NULL, &txn, 0);                                                               
+    CKERR(r);
+    r = db->cursor(db, txn, &cursor, 0); assert(r == 0);
+    DBT k1; memset(&k1, 0, sizeof k1);
+    DBT v1; memset(&v1, 0, sizeof v1);
+    if (!abort_loader) {
+        for (uint32_t i = 0; i < num_elements; i++) {
+            r = cursor->c_get(cursor, &k1, &v1, DB_NEXT); assert(r == 0);
+            assert(k1.size == sizeof(uint32_t));
+            assert(v1.size == sizeof(uint32_t));
+            assert(*(uint32_t *)k1.data == i);
+            assert(*(uint32_t *)v1.data == i);
+        }
+    }
+    r = cursor->c_get(cursor, &k1, &v1, DB_NEXT); assert(r == 0);
+    assert(k1.size == sizeof(uint32_t));
+    assert(v1.size == sizeof(uint32_t));
+    assert(*(uint32_t *)k1.data == num_elements);
+    assert(*(uint32_t *)v1.data == num_elements);
+    r = cursor->c_get(cursor, &k1, &v1, DB_NEXT); assert(r == DB_NOTFOUND);
+
+    r = cursor->c_close(cursor); assert(r == 0);
+    r = txn->commit(txn, 0);
+    CKERR(r);
+
+    r = db->close(db, 0); assert(r == 0);
+    r = env->close(env, 0); assert(r == 0);
+}
+
+int
+test_main(int argc, char *const argv[]) {
+    parse_args(argc, argv);
+    test_loader_abort(false, false);
+    test_loader_abort(false, true);
+    test_loader_abort(true, false);
+    test_loader_abort(true, true);
+    return 0;
+}