Closes #2003 closes [t:2003] add 'table lock on empty' log entry, and recovery logic for it.

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

newbrt/brt.c

@@ -4905,15 +4905,22 @@ brt_is_empty (BRT brt) {
 int
 toku_brt_note_table_lock (BRT brt, TOKUTXN txn)
 {
+    int r = 0;
     if (brt->h->txnid_that_created_or_locked_when_empty != toku_txn_get_txnid(txn) &&
         brt_is_empty(brt) &&
         brt->h->txnid_that_created_or_locked_when_empty == 0) {
         brt->h->txnid_that_created_or_locked_when_empty = toku_txn_get_txnid(txn);
-        int r = toku_txn_note_brt(txn, brt);
+        r = toku_txn_note_brt(txn, brt);
         assert(r==0);
-        return toku_logger_save_rollback_tablelock_on_empty_table(txn, toku_cachefile_filenum(brt->cf));
+        r = toku_logger_save_rollback_tablelock_on_empty_table(txn, toku_cachefile_filenum(brt->cf));
+        if (r==0) {
+            TOKULOGGER logger = toku_txn_logger(txn);
+            TXNID xid = toku_txn_get_txnid(txn);
+            r = toku_log_tablelock_on_empty_table(logger, (LSN*)NULL,
+                                                  0, toku_cachefile_filenum(brt->cf), xid);
+        }
     }
-    return 0;
+    return r;
 }
 
 LSN toku_brt_checkpoint_lsn(BRT brt) {

newbrt/logformat.c

@@ -116,6 +116,9 @@ const struct logtype logtypes[] = {
     {"fclose",   'e', FA{{"BYTESTRING", "fname", 0},
                          {"FILENUM",    "filenum", 0},
                          NULLFIELD}},
+    {"tablelock_on_empty_table", 'L', FA{{"FILENUM",    "filenum", 0},
+                           {"TXNID",      "xid", 0},
+                           NULLFIELD}},
     {"enq_insert", 'I', FA{{"FILENUM",    "filenum", 0},
                            {"TXNID",      "xid", 0},
                            {"BYTESTRING", "key", 0},

newbrt/recover.c

@@ -193,6 +193,11 @@ static int toku_recover_backward_xabort (struct logtype_xabort *UU(l), RECOVER_E
     return 0;
 }
 
+static int toku_recover_backward_tablelock_on_empty_table(struct logtype_tablelock_on_empty_table *UU(l), RECOVER_ENV UU(renv)) {
+    // nothing
+    return 0;
+}
+
 static void create_dir_from_file (const char *fname) {
     char *tmp=toku_strdup(fname);
     char ch;
@@ -313,6 +318,22 @@ static void toku_recover_enq_insert (LSN lsn, FILENUM filenum, TXNID xid, BYTEST
     assert(r == 0);
 }
 
+static int toku_recover_tablelock_on_empty_table(LSN UU(lsn), FILENUM filenum, TXNID xid, RECOVER_ENV renv) {
+    struct cf_pair *pair = NULL;
+    int r = find_cachefile(&renv->fmap, filenum, &pair);
+    if (r!=0) {
+	// if we didn't find a cachefile, then we don't have to do anything.
+	return 0;
+    }    
+
+    TOKUTXN txn;
+    r = toku_txnid2txn(renv->logger, xid, &txn);
+    assert(r == 0);
+    r = toku_brt_note_table_lock(pair->brt, txn);
+    assert(r == 0);
+    return 0;
+}
+
 static int toku_recover_backward_enq_insert (struct logtype_enq_insert *UU(l), RECOVER_ENV UU(renv)) {
     // nothing
     return 0;