Added meta-database support.

Summary:
Added kMetaDatabase for meta-databases in db/filename.h along with supporting
fuctions.
Fixed switch in DBImpl so that it also handles kMetaDatabase.
Fixed DestroyDB() that it can handle destroying meta-databases.

Test Plan: make check

Reviewers: sheki, emayanke, vamsi, dhruba

Reviewed By: dhruba

Differential Revision: https://reviews.facebook.net/D7245

db/db_impl.cc

@@ -413,6 +413,7 @@ void DBImpl::PurgeObsoleteFiles(DeletionState& state) {
           break;
         case kCurrentFile:
         case kDBLockFile:
+        case kMetaDatabase:
           keep = true;
           break;
       }
@@ -2330,7 +2331,12 @@ Status DestroyDB(const std::string& dbname, const Options& options) {
     for (size_t i = 0; i < filenames.size(); i++) {
       if (ParseFileName(filenames[i], &number, &type) &&
           type != kDBLockFile) {  // Lock file will be deleted at end
-        Status del = env->DeleteFile(dbname + "/" + filenames[i]);
+        Status del;
+        if (type == kMetaDatabase) {
+          del = DestroyDB(dbname + "/" + filenames[i], options);
+        } else {
+          del = env->DeleteFile(dbname + "/" + filenames[i]);
+        }
         if (result.ok() && !del.ok()) {
           result = del;
         }

db/db_test.cc

@@ -1962,6 +1962,41 @@ TEST(DBTest, DBOpen_Change_NumLevels) {
   ASSERT_TRUE(db == NULL);
 }
 
+TEST(DBTest, DestroyDBMetaDatabase) {
+  std::string dbname = test::TmpDir() + "/db_meta";
+  std::string metadbname = MetaDatabaseName(dbname, 0);
+  std::string metametadbname = MetaDatabaseName(metadbname, 0);
+
+  // Destroy previous versions if they exist. Using the long way.
+  DestroyDB(metametadbname, Options());
+  DestroyDB(metadbname, Options());
+  DestroyDB(dbname, Options());
+
+  // Setup databases
+  Options opts;
+  opts.create_if_missing = true;
+  DB* db = NULL;
+  ASSERT_OK(DB::Open(opts, dbname, &db));
+  delete db;
+  db = NULL;
+  ASSERT_OK(DB::Open(opts, metadbname, &db));
+  delete db;
+  db = NULL;
+  ASSERT_OK(DB::Open(opts, metametadbname, &db));
+  delete db;
+  db = NULL;
+
+  // Delete databases
+  DestroyDB(dbname, Options());
+
+  // Check if deletion worked.
+  opts.create_if_missing = false;
+  ASSERT_TRUE(!DB::Open(opts, dbname, &db).ok());
+  ASSERT_TRUE(!DB::Open(opts, metadbname, &db).ok());
+  ASSERT_TRUE(!DB::Open(opts, metametadbname, &db).ok());
+}
+
+
 // Check that number of files does not grow when we are out of space
 TEST(DBTest, NoSpace) {
   Options options = CurrentOptions();

db/filename.cc

@@ -115,6 +115,12 @@ std::string OldInfoLogFileName(const std::string& dbname, uint64_t ts,
   return log_dir + "/" + flatten_db_path + "_LOG.old." + buf;
 }
 
+std::string MetaDatabaseName(const std::string& dbname, uint64_t number) {
+  char buf[100];
+  snprintf(buf, sizeof(buf), "/METADB-%llu",
+           static_cast<unsigned long long>(number));
+  return dbname + buf;
+}
 
 // Owned filenames have the form:
 //    dbname/CURRENT
@@ -123,6 +129,7 @@ std::string OldInfoLogFileName(const std::string& dbname, uint64_t ts,
 //    dbname/LOG.old.[0-9]+
 //    dbname/MANIFEST-[0-9]+
 //    dbname/[0-9]+.(log|sst)
+//    dbname/METADB-[0-9]+
 bool ParseFileName(const std::string& fname,
                    uint64_t* number,
                    FileType* type) {
@@ -155,6 +162,17 @@ bool ParseFileName(const std::string& fname,
     }
     *type = kDescriptorFile;
     *number = num;
+  } else if (rest.starts_with("METADB-")) {
+    rest.remove_prefix(strlen("METADB-"));
+    uint64_t num;
+    if (!ConsumeDecimalNumber(&rest, &num)) {
+      return false;
+    }
+    if (!rest.empty()) {
+      return false;
+    }
+    *type = kMetaDatabase;
+    *number = num;
   } else {
     // Avoid strtoull() to keep filename format independent of the
     // current locale

db/filename.h

@@ -24,7 +24,8 @@ enum FileType {
   kDescriptorFile,
   kCurrentFile,
   kTempFile,
-  kInfoLogFile  // Either the current one, or an old one
+  kInfoLogFile,  // Either the current one, or an old one
+  kMetaDatabase
 };
 
 // Return the name of the log file with the specified number
@@ -73,6 +74,11 @@ extern std::string InfoLogFileName(const std::string& dbname,
 extern std::string OldInfoLogFileName(const std::string& dbname, uint64_t ts,
     const std::string& db_path="", const std::string& log_dir="");
 
+// Return the name to use for a metadatabase. The result will be prefixed with
+// "dbname".
+extern std::string MetaDatabaseName(const std::string& dbname,
+                                    uint64_t number);
+
 // If filename is a leveldb file, store the type of the file in *type.
 // The number encoded in the filename is stored in *number.  If the
 // filename was successfully parsed, returns true.  Else return false.

db/filename_test.cc

@@ -31,6 +31,8 @@ TEST(FileNameTest, Parse) {
     { "LOCK",               0,     kDBLockFile },
     { "MANIFEST-2",         2,     kDescriptorFile },
     { "MANIFEST-7",         7,     kDescriptorFile },
+    { "METADB-2",           2,     kMetaDatabase },
+    { "METADB-7",           7,     kMetaDatabase },
     { "LOG",                0,     kInfoLogFile },
     { "LOG.old",            0,     kInfoLogFile },
     { "18446744073709551615.log", 18446744073709551615ull, kLogFile },
@@ -57,6 +59,11 @@ TEST(FileNameTest, Parse) {
     "MANIFEST-",
     "XMANIFEST-3",
     "MANIFEST-3x",
+    "META",
+    "METADB",
+    "METADB-",
+    "XMETADB-3",
+    "METADB-3x",
     "LOC",
     "LOCKx",
     "LO",
@@ -113,6 +120,12 @@ TEST(FileNameTest, Construction) {
   ASSERT_TRUE(ParseFileName(fname.c_str() + 4, &number, &type));
   ASSERT_EQ(999U, number);
   ASSERT_EQ(kTempFile, type);
+
+  fname = MetaDatabaseName("met", 100);
+  ASSERT_EQ("met/", std::string(fname.data(), 4));
+  ASSERT_TRUE(ParseFileName(fname.c_str() + 4, &number, &type));
+  ASSERT_EQ(100U, number);
+  ASSERT_EQ(kMetaDatabase, type);
 }
 
 }  // namespace leveldb