Make DB::Open fail if sst files are missing.

Also, cleanup for Clang's -Wimplicit-fallthrough warning.

db/db_impl.cc

@@ -310,16 +310,24 @@ Status DBImpl::Recover(VersionEdit* edit) {
     if (!s.ok()) {
       return s;
     }
+    std::set<uint64_t> expected;
+    versions_->AddLiveFiles(&expected);
     uint64_t number;
     FileType type;
     std::vector<uint64_t> logs;
     for (size_t i = 0; i < filenames.size(); i++) {
-      if (ParseFileName(filenames[i], &number, &type)
+      if (ParseFileName(filenames[i], &number, &type)) {
-          && type == kLogFile
+        expected.erase(number);
-          && ((number >= min_log) || (number == prev_log))) {
+        if (type == kLogFile && ((number >= min_log) || (number == prev_log)))
-        logs.push_back(number);
+          logs.push_back(number);
       }
     }
+    if (!expected.empty()) {
+      char buf[50];
+      snprintf(buf, sizeof(buf), "%d missing files; e.g.",
+               static_cast<int>(expected.size()));
+      return Status::Corruption(buf, TableFileName(dbname_, *(expected.begin())));
+    }
 
     // Recover in the order in which the logs were generated
     std::sort(logs.begin(), logs.end());

db/db_test.cc

@@ -461,6 +461,20 @@ class DBTest {
     }
     return result;
   }
+
+  bool DeleteAnSSTFile() {
+    std::vector<std::string> filenames;
+    ASSERT_OK(env_->GetChildren(dbname_, &filenames));
+    uint64_t number;
+    FileType type;
+    for (size_t i = 0; i < filenames.size(); i++) {
+      if (ParseFileName(filenames[i], &number, &type) && type == kTableFile) {
+        ASSERT_OK(env_->DeleteFile(TableFileName(dbname_, number)));
+        return true;
+      }
+    }
+    return false;
+  }
 };
 
 TEST(DBTest, Empty) {
@@ -1567,6 +1581,23 @@ TEST(DBTest, ManifestWriteError) {
   }
 }
 
+TEST(DBTest, MissingSSTFile) {
+  ASSERT_OK(Put("foo", "bar"));
+  ASSERT_EQ("bar", Get("foo"));
+
+  // Dump the memtable to disk.
+  dbfull()->TEST_CompactMemTable();
+  ASSERT_EQ("bar", Get("foo"));
+
+  ASSERT_TRUE(DeleteAnSSTFile());
+  Options options = CurrentOptions();
+  options.paranoid_checks = true;
+  Status s = TryReopen(&options);
+  ASSERT_TRUE(!s.ok());
+  ASSERT_TRUE(s.ToString().find("issing") != std::string::npos)
+      << s.ToString();
+}
+
 TEST(DBTest, FilesDeletedAfterCompaction) {
   ASSERT_OK(Put("foo", "v2"));
   Compact("a", "z");

util/hash.cc

@@ -6,6 +6,13 @@
 #include "util/coding.h"
 #include "util/hash.h"
 
+// The FALLTHROUGH_INTENDED macro can be used to annotate implicit fall-through
+// between switch labels. The real definition should be provided externally.
+// This one is a fallback version for unsupported compilers.
+#ifndef FALLTHROUGH_INTENDED
+#define FALLTHROUGH_INTENDED do { } while (0)
+#endif
+
 namespace leveldb {
 
 uint32_t Hash(const char* data, size_t n, uint32_t seed) {
@@ -28,10 +35,10 @@ uint32_t Hash(const char* data, size_t n, uint32_t seed) {
   switch (limit - data) {
     case 3:
       h += data[2] << 16;
-      // fall through
+      FALLTHROUGH_INTENDED;
     case 2:
       h += data[1] << 8;
-      // fall through
+      FALLTHROUGH_INTENDED;
     case 1:
       h += data[0];
       h *= m;