require at least 1 log entry during recovery [t:1959]

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

newbrt/recover.c

@@ -543,7 +543,7 @@ static int toku_delete_rolltmp_files (const char *log_dir) {
 // Effects: If there are no log files, or if there is a "clean" checkpoint at the end of the log,
 // then we don't need recovery to run.  Skip the shutdown log entry if there is one.
 // Returns: TRUE if we need recovery, otherwise FALSE.
-int tokudb_needs_recovery(const char *log_dir) {
+int tokudb_needs_recovery(const char *log_dir, BOOL ignore_log_empty) {
     int needs_recovery;
     int r;
     TOKULOGCURSOR logcursor = NULL;
@@ -555,7 +555,7 @@ int tokudb_needs_recovery(const char *log_dir) {
     
     struct log_entry *le = NULL;
     r = toku_logcursor_last(logcursor, &le);
-    if (r == DB_NOTFOUND) {
+    if (r == DB_NOTFOUND && ignore_log_empty) {
         needs_recovery = FALSE; goto exit;
     }
     if (r != 0) {
@@ -626,6 +626,7 @@ static int do_recovery(RECOVER_ENV env, const char *data_dir, const char *log_di
     int r;
     int rr = 0;
     TOKULOGCURSOR logcursor = NULL;
+    struct log_entry *le = NULL;
 
     char org_wd[1000];
     {
@@ -639,6 +640,19 @@ static int do_recovery(RECOVER_ENV env, const char *data_dir, const char *log_di
 
     LSN lastlsn = toku_logger_last_lsn(env->logger);
 
+    // there must be at least one log entry
+    r = toku_logcursor_create(&logcursor, log_dir);
+    assert(r == 0);
+    
+    r = toku_logcursor_last(logcursor, &le);
+    if (r != 0) {
+        rr = DB_RUNRECOVERY; goto errorexit;
+    }
+
+    // TODO use logcursor->invalidate()
+    r = toku_logcursor_destroy(&logcursor);
+    assert(r == 0);
+
     r = toku_logcursor_create(&logcursor, log_dir);
     assert(r == 0);
 
@@ -648,8 +662,6 @@ static int do_recovery(RECOVER_ENV env, const char *data_dir, const char *log_di
         rr = errno; goto errorexit;
     }
 
-    struct log_entry *le;
-
     // scan backwards
     backward_scan_state_init(&env->bs);
     while (1) {
@@ -759,7 +771,7 @@ int tokudb_recover(const char *data_dir, const char *log_dir, brt_compare_func b
     }
 
     int rr = 0;
-    if (tokudb_needs_recovery(log_dir)) {
+    if (tokudb_needs_recovery(log_dir, FALSE)) {
         struct recover_env renv;
         r = recover_env_init(&renv, bt_compare, dup_compare);
         assert(r == 0);

newbrt/recover.h

@@ -22,6 +22,6 @@ int tokudb_recover(const char *datadir, const char *logdir, brt_compare_func bt_
 // If the log is empty or if there is a clean shutdown at the end of the log, then we
 // dont need to run recovery.
 // Returns: TRUE if we need recovery, otherwise FALSE.
-int tokudb_needs_recovery(const char *logdir);
+int tokudb_needs_recovery(const char *logdir, BOOL ignore_empty_log);
 
 #endif // TOKURECOVER_H

src/tests/recover-checkpoint-fopen-1.c

@@ -3,7 +3,7 @@
 #include <sys/stat.h>
 #include "test.h"
 
-const int envflags = DB_INIT_MPOOL|DB_CREATE|DB_THREAD |DB_INIT_LOCK|DB_INIT_LOG|DB_INIT_TXN |DB_RECOVER;
+const int envflags = DB_INIT_MPOOL|DB_CREATE|DB_THREAD |DB_INIT_LOCK|DB_INIT_LOG|DB_INIT_TXN;
 char *namea="a.db";
 char *nameb="b.db";
 
@@ -51,7 +51,7 @@ static void run_recover (BOOL did_commit) {
     DB *dba, *dbb;
     int r;
     r = db_env_create(&env, 0);                                                             CKERR(r);
-    r = env->open(env, ENVDIR, envflags, S_IRWXU+S_IRWXG+S_IRWXO);                          CKERR(r);
+    r = env->open(env, ENVDIR, envflags|DB_RECOVER, S_IRWXU+S_IRWXG+S_IRWXO);                          CKERR(r);
     r = db_create(&dba, env, 0);                                                            CKERR(r);
     r = dba->open(dba, NULL, namea, NULL, DB_BTREE, DB_AUTO_COMMIT|DB_CREATE, 0666);        CKERR(r);
 
@@ -113,7 +113,7 @@ static void run_recover_only (void) {
     int r;
 
     r = db_env_create(&env, 0);                                                             CKERR(r);
-    r = env->open(env, ENVDIR, envflags, S_IRWXU+S_IRWXG+S_IRWXO);                          CKERR(r);
+    r = env->open(env, ENVDIR, envflags|DB_RECOVER, S_IRWXU+S_IRWXG+S_IRWXO);                          CKERR(r);
     r = env->close(env, 0);                                                                 CKERR(r);
 }
 
@@ -122,7 +122,8 @@ static void run_no_recover (void) {
     int r;
 
     r = db_env_create(&env, 0);                                                             CKERR(r);
-    r = env->open(env, ENVDIR, envflags & ~DB_RECOVER, S_IRWXU+S_IRWXG+S_IRWXO);            CKERR(r);
+    r = env->open(env, ENVDIR, envflags & ~DB_RECOVER, S_IRWXU+S_IRWXG+S_IRWXO);
+    assert(r == DB_RUNRECOVERY);
     r = env->close(env, 0);                                                                 CKERR(r);
 }
 

src/tests/recover-checkpoint-fopen-dupsort-1.c

@@ -3,7 +3,7 @@
 #include <sys/stat.h>
 #include "test.h"
 
-const int envflags = DB_INIT_MPOOL|DB_CREATE|DB_THREAD |DB_INIT_LOCK|DB_INIT_LOG|DB_INIT_TXN |DB_RECOVER;
+const int envflags = DB_INIT_MPOOL|DB_CREATE|DB_THREAD |DB_INIT_LOCK|DB_INIT_LOG|DB_INIT_TXN;
 char *namea="a.db";
 char *nameb="b.db";
 
@@ -53,7 +53,7 @@ static void run_recover (BOOL did_commit) {
     DB *dba, *dbb;
     int r;
     r = db_env_create(&env, 0);                                                             CKERR(r);
-    r = env->open(env, ENVDIR, envflags, S_IRWXU+S_IRWXG+S_IRWXO);                          CKERR(r);
+    r = env->open(env, ENVDIR, envflags|DB_RECOVER, S_IRWXU+S_IRWXG+S_IRWXO);                          CKERR(r);
     r = db_create(&dba, env, 0);                                                            CKERR(r);
     r = dba->open(dba, NULL, namea, NULL, DB_BTREE, DB_AUTO_COMMIT|DB_CREATE, 0666);        CKERR(r);
 
@@ -115,7 +115,7 @@ static void run_recover_only (void) {
     int r;
 
     r = db_env_create(&env, 0);                                                             CKERR(r);
-    r = env->open(env, ENVDIR, envflags, S_IRWXU+S_IRWXG+S_IRWXO);                          CKERR(r);
+    r = env->open(env, ENVDIR, envflags|DB_RECOVER, S_IRWXU+S_IRWXG+S_IRWXO);                          CKERR(r);
     r = env->close(env, 0);                                                                 CKERR(r);
 }
 
@@ -124,7 +124,8 @@ static void run_no_recover (void) {
     int r;
 
     r = db_env_create(&env, 0);                                                             CKERR(r);
-    r = env->open(env, ENVDIR, envflags & ~DB_RECOVER, S_IRWXU+S_IRWXG+S_IRWXO);            CKERR(r);
+    r = env->open(env, ENVDIR, envflags & ~DB_RECOVER, S_IRWXU+S_IRWXG+S_IRWXO);
+    assert(r == DB_RUNRECOVERY);
     r = env->close(env, 0);                                                                 CKERR(r);
 }
 

src/tests/recover-x1-nested2.c

@@ -15,7 +15,7 @@
 #include <sys/wait.h>
 #include "test.h"
 
-const int envflags = DB_INIT_MPOOL|DB_CREATE|DB_THREAD |DB_INIT_LOCK|DB_INIT_LOG|DB_INIT_TXN |DB_RECOVER;
+const int envflags = DB_INIT_MPOOL|DB_CREATE|DB_THREAD |DB_INIT_LOCK|DB_INIT_LOG|DB_INIT_TXN;
 char *namea="a.db";
 char *nameb="b.db";
 
@@ -65,7 +65,7 @@ do_x1_recover (BOOL did_commit) {
     DB *dba, *dbb;
     int r;
     r = db_env_create(&env, 0);                                                             CKERR(r);
-    r = env->open(env, ENVDIR, envflags, S_IRWXU+S_IRWXG+S_IRWXO);                          CKERR(r);
+    r = env->open(env, ENVDIR, envflags|DB_RECOVER, S_IRWXU+S_IRWXG+S_IRWXO);                          CKERR(r);
     r = db_create(&dba, env, 0);                                                            CKERR(r);
     r = dba->open(dba, NULL, namea, NULL, DB_BTREE, DB_AUTO_COMMIT|DB_CREATE, 0666);        CKERR(r);
     r = db_create(&dbb, env, 0);                                                            CKERR(r);
@@ -118,7 +118,7 @@ do_x1_recover_only (void) {
     int r;
 
     r = db_env_create(&env, 0);                                                             CKERR(r);
-    r = env->open(env, ENVDIR, envflags, S_IRWXU+S_IRWXG+S_IRWXO);                          CKERR(r);
+    r = env->open(env, ENVDIR, envflags|DB_RECOVER, S_IRWXU+S_IRWXG+S_IRWXO);                          CKERR(r);
     r = env->close(env, 0);                                                                 CKERR(r);
     exit(0);
 }

src/tests/recover-x1.c

@@ -15,7 +15,7 @@
 #include <sys/wait.h>
 #include "test.h"
 
-const int envflags = DB_INIT_MPOOL|DB_CREATE|DB_THREAD |DB_INIT_LOCK|DB_INIT_LOG|DB_INIT_TXN |DB_RECOVER;
+const int envflags = DB_INIT_MPOOL|DB_CREATE|DB_THREAD |DB_INIT_LOCK|DB_INIT_LOG|DB_INIT_TXN;
 char *namea="a.db";
 char *nameb="b.db";
 
@@ -62,7 +62,7 @@ do_x1_recover (BOOL did_commit) {
     DB *dba, *dbb;
     int r;
     r = db_env_create(&env, 0);                                                             CKERR(r);
-    r = env->open(env, ENVDIR, envflags, S_IRWXU+S_IRWXG+S_IRWXO);                          CKERR(r);
+    r = env->open(env, ENVDIR, envflags|DB_RECOVER, S_IRWXU+S_IRWXG+S_IRWXO);                          CKERR(r);
     r = db_create(&dba, env, 0);                                                            CKERR(r);
     r = dba->open(dba, NULL, namea, NULL, DB_BTREE, DB_AUTO_COMMIT|DB_CREATE, 0666);        CKERR(r);
     r = db_create(&dbb, env, 0);                                                            CKERR(r);
@@ -115,7 +115,7 @@ do_x1_recover_only (void) {
     int r;
 
     r = db_env_create(&env, 0);                                                             CKERR(r);
-    r = env->open(env, ENVDIR, envflags, S_IRWXU+S_IRWXG+S_IRWXO);                          CKERR(r);
+    r = env->open(env, ENVDIR, envflags|DB_RECOVER, S_IRWXU+S_IRWXG+S_IRWXO);                          CKERR(r);
     r = env->close(env, 0);                                                                 CKERR(r);
     exit(0);
 }
@@ -126,7 +126,8 @@ do_x1_no_recover (void) {
     int r;
 
     r = db_env_create(&env, 0);                                                             CKERR(r);
-    r = env->open(env, ENVDIR, envflags & ~DB_RECOVER, S_IRWXU+S_IRWXG+S_IRWXO);            CKERR(r);
+    r = env->open(env, ENVDIR, envflags & ~DB_RECOVER, S_IRWXU+S_IRWXG+S_IRWXO);
+    assert(r == DB_RUNRECOVERY);
     r = env->close(env, 0);                                                                 CKERR(r);
     exit(0);
 }

src/tests/recover-x2.c

@@ -19,7 +19,7 @@
 #include <sys/wait.h>
 #include "test.h"
 
-const int envflags = DB_INIT_MPOOL|DB_CREATE|DB_THREAD |DB_INIT_LOCK|DB_INIT_LOG|DB_INIT_TXN |DB_RECOVER;
+const int envflags = DB_INIT_MPOOL|DB_CREATE|DB_THREAD |DB_INIT_LOCK|DB_INIT_LOG|DB_INIT_TXN;
 char *namea="a.db";
 char *nameb="b.db";
 
@@ -79,7 +79,7 @@ do_x2_recover (BOOL did_commit) {
     DB_ENV *env;
     int r;
     r = db_env_create(&env, 0);                                                             CKERR(r);
-    r = env->open(env, ENVDIR, envflags, S_IRWXU+S_IRWXG+S_IRWXO);                          CKERR(r);
+    r = env->open(env, ENVDIR, envflags|DB_RECOVER, S_IRWXU+S_IRWXG+S_IRWXO);                          CKERR(r);
     DB_TXN *txn;
     r = env->txn_begin(env, NULL, &txn, 0);                                                 CKERR(r);
     {

src/ydb.c

@@ -314,7 +314,7 @@ static int needs_recovery (DB_ENV *env) {
     } else {
 	logdir = toku_strdup(env->i->dir);
     }
-    BOOL recovery_needed = tokudb_needs_recovery(logdir);
+    BOOL recovery_needed = tokudb_needs_recovery(logdir, TRUE);
     toku_free(logdir);
     return recovery_needed ? DB_RUNRECOVERY : 0;
 }