diff --git a/src/leveldb/db/db_test.cc b/src/leveldb/db/db_test.cc
index a0b08bc19..e8e11c201 100644
--- a/src/leveldb/db/db_test.cc
+++ b/src/leveldb/db/db_test.cc
@@ -1,2158 +1,2162 @@
 // Copyright (c) 2011 The LevelDB Authors. All rights reserved.
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file. See the AUTHORS file for names of contributors.
 
 #include "leveldb/db.h"
 #include "leveldb/filter_policy.h"
 #include "db/db_impl.h"
 #include "db/filename.h"
 #include "db/version_set.h"
 #include "db/write_batch_internal.h"
 #include "leveldb/cache.h"
 #include "leveldb/env.h"
 #include "leveldb/table.h"
 #include "util/hash.h"
 #include "util/logging.h"
 #include "util/mutexlock.h"
 #include "util/testharness.h"
 #include "util/testutil.h"
 
 namespace leveldb {
 
 static std::string RandomString(Random* rnd, int len) {
   std::string r;
   test::RandomString(rnd, len, &r);
   return r;
 }
 
 namespace {
 class AtomicCounter {
  private:
   port::Mutex mu_;
   int count_;
  public:
   AtomicCounter() : count_(0) { }
   void Increment() {
     IncrementBy(1);
   }
   void IncrementBy(int count) {
     MutexLock l(&mu_);
     count_ += count;
   }
   int Read() {
     MutexLock l(&mu_);
     return count_;
   }
   void Reset() {
     MutexLock l(&mu_);
     count_ = 0;
   }
 };
 
 void DelayMilliseconds(int millis) {
   Env::Default()->SleepForMicroseconds(millis * 1000);
 }
 }
 
 // Special Env used to delay background operations
 class SpecialEnv : public EnvWrapper {
  public:
   // sstable/log Sync() calls are blocked while this pointer is non-NULL.
   port::AtomicPointer delay_data_sync_;
 
   // sstable/log Sync() calls return an error.
   port::AtomicPointer data_sync_error_;
 
   // Simulate no-space errors while this pointer is non-NULL.
   port::AtomicPointer no_space_;
 
   // Simulate non-writable file system while this pointer is non-NULL
   port::AtomicPointer non_writable_;
 
   // Force sync of manifest files to fail while this pointer is non-NULL
   port::AtomicPointer manifest_sync_error_;
 
   // Force write to manifest files to fail while this pointer is non-NULL
   port::AtomicPointer manifest_write_error_;
 
   bool count_random_reads_;
   AtomicCounter random_read_counter_;
 
   explicit SpecialEnv(Env* base) : EnvWrapper(base) {
     delay_data_sync_.Release_Store(NULL);
     data_sync_error_.Release_Store(NULL);
     no_space_.Release_Store(NULL);
     non_writable_.Release_Store(NULL);
     count_random_reads_ = false;
     manifest_sync_error_.Release_Store(NULL);
     manifest_write_error_.Release_Store(NULL);
   }
 
   Status NewWritableFile(const std::string& f, WritableFile** r) {
     class DataFile : public WritableFile {
      private:
       SpecialEnv* env_;
       WritableFile* base_;
 
      public:
       DataFile(SpecialEnv* env, WritableFile* base)
           : env_(env),
             base_(base) {
       }
       ~DataFile() { delete base_; }
       Status Append(const Slice& data) {
         if (env_->no_space_.Acquire_Load() != NULL) {
           // Drop writes on the floor
           return Status::OK();
         } else {
           return base_->Append(data);
         }
       }
       Status Close() { return base_->Close(); }
       Status Flush() { return base_->Flush(); }
       Status Sync() {
         if (env_->data_sync_error_.Acquire_Load() != NULL) {
           return Status::IOError("simulated data sync error");
         }
         while (env_->delay_data_sync_.Acquire_Load() != NULL) {
           DelayMilliseconds(100);
         }
         return base_->Sync();
       }
+      std::string GetName() const override { return "[datafile]"; }
     };
     class ManifestFile : public WritableFile {
      private:
       SpecialEnv* env_;
       WritableFile* base_;
      public:
       ManifestFile(SpecialEnv* env, WritableFile* b) : env_(env), base_(b) { }
       ~ManifestFile() { delete base_; }
       Status Append(const Slice& data) {
         if (env_->manifest_write_error_.Acquire_Load() != NULL) {
           return Status::IOError("simulated writer error");
         } else {
           return base_->Append(data);
         }
       }
       Status Close() { return base_->Close(); }
       Status Flush() { return base_->Flush(); }
       Status Sync() {
         if (env_->manifest_sync_error_.Acquire_Load() != NULL) {
           return Status::IOError("simulated sync error");
         } else {
           return base_->Sync();
         }
       }
+      std::string GetName() const override { return "[manifestfile]"; }
+
     };
 
     if (non_writable_.Acquire_Load() != NULL) {
       return Status::IOError("simulated write error");
     }
 
     Status s = target()->NewWritableFile(f, r);
     if (s.ok()) {
       if (strstr(f.c_str(), ".ldb") != NULL ||
           strstr(f.c_str(), ".log") != NULL) {
         *r = new DataFile(this, *r);
       } else if (strstr(f.c_str(), "MANIFEST") != NULL) {
         *r = new ManifestFile(this, *r);
       }
     }
     return s;
   }
 
   Status NewRandomAccessFile(const std::string& f, RandomAccessFile** r) {
     class CountingFile : public RandomAccessFile {
      private:
       RandomAccessFile* target_;
       AtomicCounter* counter_;
      public:
       CountingFile(RandomAccessFile* target, AtomicCounter* counter)
           : target_(target), counter_(counter) {
       }
       virtual ~CountingFile() { delete target_; }
       virtual Status Read(uint64_t offset, size_t n, Slice* result,
                           char* scratch) const {
         counter_->Increment();
         return target_->Read(offset, n, result, scratch);
       }
+      std::string GetName() const override { return "[countingfile]"; }
     };
 
     Status s = target()->NewRandomAccessFile(f, r);
     if (s.ok() && count_random_reads_) {
       *r = new CountingFile(*r, &random_read_counter_);
     }
     return s;
   }
 };
 
 class DBTest {
  private:
   const FilterPolicy* filter_policy_;
 
   // Sequence of option configurations to try
   enum OptionConfig {
     kDefault,
     kReuse,
     kFilter,
     kUncompressed,
     kEnd
   };
   int option_config_;
 
  public:
   std::string dbname_;
   SpecialEnv* env_;
   DB* db_;
 
   Options last_options_;
 
   DBTest() : option_config_(kDefault),
              env_(new SpecialEnv(Env::Default())) {
     filter_policy_ = NewBloomFilterPolicy(10);
     dbname_ = test::TmpDir() + "/db_test";
     DestroyDB(dbname_, Options());
     db_ = NULL;
     Reopen();
   }
 
   ~DBTest() {
     delete db_;
     DestroyDB(dbname_, Options());
     delete env_;
     delete filter_policy_;
   }
 
   // Switch to a fresh database with the next option configuration to
   // test.  Return false if there are no more configurations to test.
   bool ChangeOptions() {
     option_config_++;
     if (option_config_ >= kEnd) {
       return false;
     } else {
       DestroyAndReopen();
       return true;
     }
   }
 
   // Return the current option configuration.
   Options CurrentOptions() {
     Options options;
     options.reuse_logs = false;
     switch (option_config_) {
       case kReuse:
         options.reuse_logs = true;
         break;
       case kFilter:
         options.filter_policy = filter_policy_;
         break;
       case kUncompressed:
         options.compression = kNoCompression;
         break;
       default:
         break;
     }
     return options;
   }
 
   DBImpl* dbfull() {
     return reinterpret_cast<DBImpl*>(db_);
   }
 
   void Reopen(Options* options = NULL) {
     ASSERT_OK(TryReopen(options));
   }
 
   void Close() {
     delete db_;
     db_ = NULL;
   }
 
   void DestroyAndReopen(Options* options = NULL) {
     delete db_;
     db_ = NULL;
     DestroyDB(dbname_, Options());
     ASSERT_OK(TryReopen(options));
   }
 
   Status TryReopen(Options* options) {
     delete db_;
     db_ = NULL;
     Options opts;
     if (options != NULL) {
       opts = *options;
     } else {
       opts = CurrentOptions();
       opts.create_if_missing = true;
     }
     last_options_ = opts;
 
     return DB::Open(opts, dbname_, &db_);
   }
 
   Status Put(const std::string& k, const std::string& v) {
     return db_->Put(WriteOptions(), k, v);
   }
 
   Status Delete(const std::string& k) {
     return db_->Delete(WriteOptions(), k);
   }
 
   std::string Get(const std::string& k, const Snapshot* snapshot = NULL) {
     ReadOptions options;
     options.snapshot = snapshot;
     std::string result;
     Status s = db_->Get(options, k, &result);
     if (s.IsNotFound()) {
       result = "NOT_FOUND";
     } else if (!s.ok()) {
       result = s.ToString();
     }
     return result;
   }
 
   // Return a string that contains all key,value pairs in order,
   // formatted like "(k1->v1)(k2->v2)".
   std::string Contents() {
     std::vector<std::string> forward;
     std::string result;
     Iterator* iter = db_->NewIterator(ReadOptions());
     for (iter->SeekToFirst(); iter->Valid(); iter->Next()) {
       std::string s = IterStatus(iter);
       result.push_back('(');
       result.append(s);
       result.push_back(')');
       forward.push_back(s);
     }
 
     // Check reverse iteration results are the reverse of forward results
     size_t matched = 0;
     for (iter->SeekToLast(); iter->Valid(); iter->Prev()) {
       ASSERT_LT(matched, forward.size());
       ASSERT_EQ(IterStatus(iter), forward[forward.size() - matched - 1]);
       matched++;
     }
     ASSERT_EQ(matched, forward.size());
 
     delete iter;
     return result;
   }
 
   std::string AllEntriesFor(const Slice& user_key) {
     Iterator* iter = dbfull()->TEST_NewInternalIterator();
     InternalKey target(user_key, kMaxSequenceNumber, kTypeValue);
     iter->Seek(target.Encode());
     std::string result;
     if (!iter->status().ok()) {
       result = iter->status().ToString();
     } else {
       result = "[ ";
       bool first = true;
       while (iter->Valid()) {
         ParsedInternalKey ikey;
         if (!ParseInternalKey(iter->key(), &ikey)) {
           result += "CORRUPTED";
         } else {
           if (last_options_.comparator->Compare(ikey.user_key, user_key) != 0) {
             break;
           }
           if (!first) {
             result += ", ";
           }
           first = false;
           switch (ikey.type) {
             case kTypeValue:
               result += iter->value().ToString();
               break;
             case kTypeDeletion:
               result += "DEL";
               break;
           }
         }
         iter->Next();
       }
       if (!first) {
         result += " ";
       }
       result += "]";
     }
     delete iter;
     return result;
   }
 
   int NumTableFilesAtLevel(int level) {
     std::string property;
     ASSERT_TRUE(
         db_->GetProperty("leveldb.num-files-at-level" + NumberToString(level),
                          &property));
     return atoi(property.c_str());
   }
 
   int TotalTableFiles() {
     int result = 0;
     for (int level = 0; level < config::kNumLevels; level++) {
       result += NumTableFilesAtLevel(level);
     }
     return result;
   }
 
   // Return spread of files per level
   std::string FilesPerLevel() {
     std::string result;
     int last_non_zero_offset = 0;
     for (int level = 0; level < config::kNumLevels; level++) {
       int f = NumTableFilesAtLevel(level);
       char buf[100];
       snprintf(buf, sizeof(buf), "%s%d", (level ? "," : ""), f);
       result += buf;
       if (f > 0) {
         last_non_zero_offset = result.size();
       }
     }
     result.resize(last_non_zero_offset);
     return result;
   }
 
   int CountFiles() {
     std::vector<std::string> files;
     env_->GetChildren(dbname_, &files);
     return static_cast<int>(files.size());
   }
 
   uint64_t Size(const Slice& start, const Slice& limit) {
     Range r(start, limit);
     uint64_t size;
     db_->GetApproximateSizes(&r, 1, &size);
     return size;
   }
 
   void Compact(const Slice& start, const Slice& limit) {
     db_->CompactRange(&start, &limit);
   }
 
   // Do n memtable compactions, each of which produces an sstable
   // covering the range [small,large].
   void MakeTables(int n, const std::string& small, const std::string& large) {
     for (int i = 0; i < n; i++) {
       Put(small, "begin");
       Put(large, "end");
       dbfull()->TEST_CompactMemTable();
     }
   }
 
   // Prevent pushing of new sstables into deeper levels by adding
   // tables that cover a specified range to all levels.
   void FillLevels(const std::string& smallest, const std::string& largest) {
     MakeTables(config::kNumLevels, smallest, largest);
   }
 
   void DumpFileCounts(const char* label) {
     fprintf(stderr, "---\n%s:\n", label);
     fprintf(stderr, "maxoverlap: %lld\n",
             static_cast<long long>(
                 dbfull()->TEST_MaxNextLevelOverlappingBytes()));
     for (int level = 0; level < config::kNumLevels; level++) {
       int num = NumTableFilesAtLevel(level);
       if (num > 0) {
         fprintf(stderr, "  level %3d : %d files\n", level, num);
       }
     }
   }
 
   std::string DumpSSTableList() {
     std::string property;
     db_->GetProperty("leveldb.sstables", &property);
     return property;
   }
 
   std::string IterStatus(Iterator* iter) {
     std::string result;
     if (iter->Valid()) {
       result = iter->key().ToString() + "->" + iter->value().ToString();
     } else {
       result = "(invalid)";
     }
     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;
   }
 
   // Returns number of files renamed.
   int RenameLDBToSST() {
     std::vector<std::string> filenames;
     ASSERT_OK(env_->GetChildren(dbname_, &filenames));
     uint64_t number;
     FileType type;
     int files_renamed = 0;
     for (size_t i = 0; i < filenames.size(); i++) {
       if (ParseFileName(filenames[i], &number, &type) && type == kTableFile) {
         const std::string from = TableFileName(dbname_, number);
         const std::string to = SSTTableFileName(dbname_, number);
         ASSERT_OK(env_->RenameFile(from, to));
         files_renamed++;
       }
     }
     return files_renamed;
   }
 };
 
 TEST(DBTest, Empty) {
   do {
     ASSERT_TRUE(db_ != NULL);
     ASSERT_EQ("NOT_FOUND", Get("foo"));
   } while (ChangeOptions());
 }
 
 TEST(DBTest, ReadWrite) {
   do {
     ASSERT_OK(Put("foo", "v1"));
     ASSERT_EQ("v1", Get("foo"));
     ASSERT_OK(Put("bar", "v2"));
     ASSERT_OK(Put("foo", "v3"));
     ASSERT_EQ("v3", Get("foo"));
     ASSERT_EQ("v2", Get("bar"));
   } while (ChangeOptions());
 }
 
 TEST(DBTest, PutDeleteGet) {
   do {
     ASSERT_OK(db_->Put(WriteOptions(), "foo", "v1"));
     ASSERT_EQ("v1", Get("foo"));
     ASSERT_OK(db_->Put(WriteOptions(), "foo", "v2"));
     ASSERT_EQ("v2", Get("foo"));
     ASSERT_OK(db_->Delete(WriteOptions(), "foo"));
     ASSERT_EQ("NOT_FOUND", Get("foo"));
   } while (ChangeOptions());
 }
 
 TEST(DBTest, GetFromImmutableLayer) {
   do {
     Options options = CurrentOptions();
     options.env = env_;
     options.write_buffer_size = 100000;  // Small write buffer
     Reopen(&options);
 
     ASSERT_OK(Put("foo", "v1"));
     ASSERT_EQ("v1", Get("foo"));
 
     env_->delay_data_sync_.Release_Store(env_);      // Block sync calls
     Put("k1", std::string(100000, 'x'));             // Fill memtable
     Put("k2", std::string(100000, 'y'));             // Trigger compaction
     ASSERT_EQ("v1", Get("foo"));
     env_->delay_data_sync_.Release_Store(NULL);      // Release sync calls
   } while (ChangeOptions());
 }
 
 TEST(DBTest, GetFromVersions) {
   do {
     ASSERT_OK(Put("foo", "v1"));
     dbfull()->TEST_CompactMemTable();
     ASSERT_EQ("v1", Get("foo"));
   } while (ChangeOptions());
 }
 
 TEST(DBTest, GetMemUsage) {
   do {
     ASSERT_OK(Put("foo", "v1"));
     std::string val;
     ASSERT_TRUE(db_->GetProperty("leveldb.approximate-memory-usage", &val));
     int mem_usage = atoi(val.c_str());
     ASSERT_GT(mem_usage, 0);
     ASSERT_LT(mem_usage, 5*1024*1024);
   } while (ChangeOptions());
 }
 
 TEST(DBTest, GetSnapshot) {
   do {
     // Try with both a short key and a long key
     for (int i = 0; i < 2; i++) {
       std::string key = (i == 0) ? std::string("foo") : std::string(200, 'x');
       ASSERT_OK(Put(key, "v1"));
       const Snapshot* s1 = db_->GetSnapshot();
       ASSERT_OK(Put(key, "v2"));
       ASSERT_EQ("v2", Get(key));
       ASSERT_EQ("v1", Get(key, s1));
       dbfull()->TEST_CompactMemTable();
       ASSERT_EQ("v2", Get(key));
       ASSERT_EQ("v1", Get(key, s1));
       db_->ReleaseSnapshot(s1);
     }
   } while (ChangeOptions());
 }
 
 TEST(DBTest, GetLevel0Ordering) {
   do {
     // Check that we process level-0 files in correct order.  The code
     // below generates two level-0 files where the earlier one comes
     // before the later one in the level-0 file list since the earlier
     // one has a smaller "smallest" key.
     ASSERT_OK(Put("bar", "b"));
     ASSERT_OK(Put("foo", "v1"));
     dbfull()->TEST_CompactMemTable();
     ASSERT_OK(Put("foo", "v2"));
     dbfull()->TEST_CompactMemTable();
     ASSERT_EQ("v2", Get("foo"));
   } while (ChangeOptions());
 }
 
 TEST(DBTest, GetOrderedByLevels) {
   do {
     ASSERT_OK(Put("foo", "v1"));
     Compact("a", "z");
     ASSERT_EQ("v1", Get("foo"));
     ASSERT_OK(Put("foo", "v2"));
     ASSERT_EQ("v2", Get("foo"));
     dbfull()->TEST_CompactMemTable();
     ASSERT_EQ("v2", Get("foo"));
   } while (ChangeOptions());
 }
 
 TEST(DBTest, GetPicksCorrectFile) {
   do {
     // Arrange to have multiple files in a non-level-0 level.
     ASSERT_OK(Put("a", "va"));
     Compact("a", "b");
     ASSERT_OK(Put("x", "vx"));
     Compact("x", "y");
     ASSERT_OK(Put("f", "vf"));
     Compact("f", "g");
     ASSERT_EQ("va", Get("a"));
     ASSERT_EQ("vf", Get("f"));
     ASSERT_EQ("vx", Get("x"));
   } while (ChangeOptions());
 }
 
 TEST(DBTest, GetEncountersEmptyLevel) {
   do {
     // Arrange for the following to happen:
     //   * sstable A in level 0
     //   * nothing in level 1
     //   * sstable B in level 2
     // Then do enough Get() calls to arrange for an automatic compaction
     // of sstable A.  A bug would cause the compaction to be marked as
     // occurring at level 1 (instead of the correct level 0).
 
     // Step 1: First place sstables in levels 0 and 2
     int compaction_count = 0;
     while (NumTableFilesAtLevel(0) == 0 ||
            NumTableFilesAtLevel(2) == 0) {
       ASSERT_LE(compaction_count, 100) << "could not fill levels 0 and 2";
       compaction_count++;
       Put("a", "begin");
       Put("z", "end");
       dbfull()->TEST_CompactMemTable();
     }
 
     // Step 2: clear level 1 if necessary.
     dbfull()->TEST_CompactRange(1, NULL, NULL);
     ASSERT_EQ(NumTableFilesAtLevel(0), 1);
     ASSERT_EQ(NumTableFilesAtLevel(1), 0);
     ASSERT_EQ(NumTableFilesAtLevel(2), 1);
 
     // Step 3: read a bunch of times
     for (int i = 0; i < 1000; i++) {
       ASSERT_EQ("NOT_FOUND", Get("missing"));
     }
 
     // Step 4: Wait for compaction to finish
     DelayMilliseconds(1000);
 
     ASSERT_EQ(NumTableFilesAtLevel(0), 0);
   } while (ChangeOptions());
 }
 
 TEST(DBTest, IterEmpty) {
   Iterator* iter = db_->NewIterator(ReadOptions());
 
   iter->SeekToFirst();
   ASSERT_EQ(IterStatus(iter), "(invalid)");
 
   iter->SeekToLast();
   ASSERT_EQ(IterStatus(iter), "(invalid)");
 
   iter->Seek("foo");
   ASSERT_EQ(IterStatus(iter), "(invalid)");
 
   delete iter;
 }
 
 TEST(DBTest, IterSingle) {
   ASSERT_OK(Put("a", "va"));
   Iterator* iter = db_->NewIterator(ReadOptions());
 
   iter->SeekToFirst();
   ASSERT_EQ(IterStatus(iter), "a->va");
   iter->Next();
   ASSERT_EQ(IterStatus(iter), "(invalid)");
   iter->SeekToFirst();
   ASSERT_EQ(IterStatus(iter), "a->va");
   iter->Prev();
   ASSERT_EQ(IterStatus(iter), "(invalid)");
 
   iter->SeekToLast();
   ASSERT_EQ(IterStatus(iter), "a->va");
   iter->Next();
   ASSERT_EQ(IterStatus(iter), "(invalid)");
   iter->SeekToLast();
   ASSERT_EQ(IterStatus(iter), "a->va");
   iter->Prev();
   ASSERT_EQ(IterStatus(iter), "(invalid)");
 
   iter->Seek("");
   ASSERT_EQ(IterStatus(iter), "a->va");
   iter->Next();
   ASSERT_EQ(IterStatus(iter), "(invalid)");
 
   iter->Seek("a");
   ASSERT_EQ(IterStatus(iter), "a->va");
   iter->Next();
   ASSERT_EQ(IterStatus(iter), "(invalid)");
 
   iter->Seek("b");
   ASSERT_EQ(IterStatus(iter), "(invalid)");
 
   delete iter;
 }
 
 TEST(DBTest, IterMulti) {
   ASSERT_OK(Put("a", "va"));
   ASSERT_OK(Put("b", "vb"));
   ASSERT_OK(Put("c", "vc"));
   Iterator* iter = db_->NewIterator(ReadOptions());
 
   iter->SeekToFirst();
   ASSERT_EQ(IterStatus(iter), "a->va");
   iter->Next();
   ASSERT_EQ(IterStatus(iter), "b->vb");
   iter->Next();
   ASSERT_EQ(IterStatus(iter), "c->vc");
   iter->Next();
   ASSERT_EQ(IterStatus(iter), "(invalid)");
   iter->SeekToFirst();
   ASSERT_EQ(IterStatus(iter), "a->va");
   iter->Prev();
   ASSERT_EQ(IterStatus(iter), "(invalid)");
 
   iter->SeekToLast();
   ASSERT_EQ(IterStatus(iter), "c->vc");
   iter->Prev();
   ASSERT_EQ(IterStatus(iter), "b->vb");
   iter->Prev();
   ASSERT_EQ(IterStatus(iter), "a->va");
   iter->Prev();
   ASSERT_EQ(IterStatus(iter), "(invalid)");
   iter->SeekToLast();
   ASSERT_EQ(IterStatus(iter), "c->vc");
   iter->Next();
   ASSERT_EQ(IterStatus(iter), "(invalid)");
 
   iter->Seek("");
   ASSERT_EQ(IterStatus(iter), "a->va");
   iter->Seek("a");
   ASSERT_EQ(IterStatus(iter), "a->va");
   iter->Seek("ax");
   ASSERT_EQ(IterStatus(iter), "b->vb");
   iter->Seek("b");
   ASSERT_EQ(IterStatus(iter), "b->vb");
   iter->Seek("z");
   ASSERT_EQ(IterStatus(iter), "(invalid)");
 
   // Switch from reverse to forward
   iter->SeekToLast();
   iter->Prev();
   iter->Prev();
   iter->Next();
   ASSERT_EQ(IterStatus(iter), "b->vb");
 
   // Switch from forward to reverse
   iter->SeekToFirst();
   iter->Next();
   iter->Next();
   iter->Prev();
   ASSERT_EQ(IterStatus(iter), "b->vb");
 
   // Make sure iter stays at snapshot
   ASSERT_OK(Put("a",  "va2"));
   ASSERT_OK(Put("a2", "va3"));
   ASSERT_OK(Put("b",  "vb2"));
   ASSERT_OK(Put("c",  "vc2"));
   ASSERT_OK(Delete("b"));
   iter->SeekToFirst();
   ASSERT_EQ(IterStatus(iter), "a->va");
   iter->Next();
   ASSERT_EQ(IterStatus(iter), "b->vb");
   iter->Next();
   ASSERT_EQ(IterStatus(iter), "c->vc");
   iter->Next();
   ASSERT_EQ(IterStatus(iter), "(invalid)");
   iter->SeekToLast();
   ASSERT_EQ(IterStatus(iter), "c->vc");
   iter->Prev();
   ASSERT_EQ(IterStatus(iter), "b->vb");
   iter->Prev();
   ASSERT_EQ(IterStatus(iter), "a->va");
   iter->Prev();
   ASSERT_EQ(IterStatus(iter), "(invalid)");
 
   delete iter;
 }
 
 TEST(DBTest, IterSmallAndLargeMix) {
   ASSERT_OK(Put("a", "va"));
   ASSERT_OK(Put("b", std::string(100000, 'b')));
   ASSERT_OK(Put("c", "vc"));
   ASSERT_OK(Put("d", std::string(100000, 'd')));
   ASSERT_OK(Put("e", std::string(100000, 'e')));
 
   Iterator* iter = db_->NewIterator(ReadOptions());
 
   iter->SeekToFirst();
   ASSERT_EQ(IterStatus(iter), "a->va");
   iter->Next();
   ASSERT_EQ(IterStatus(iter), "b->" + std::string(100000, 'b'));
   iter->Next();
   ASSERT_EQ(IterStatus(iter), "c->vc");
   iter->Next();
   ASSERT_EQ(IterStatus(iter), "d->" + std::string(100000, 'd'));
   iter->Next();
   ASSERT_EQ(IterStatus(iter), "e->" + std::string(100000, 'e'));
   iter->Next();
   ASSERT_EQ(IterStatus(iter), "(invalid)");
 
   iter->SeekToLast();
   ASSERT_EQ(IterStatus(iter), "e->" + std::string(100000, 'e'));
   iter->Prev();
   ASSERT_EQ(IterStatus(iter), "d->" + std::string(100000, 'd'));
   iter->Prev();
   ASSERT_EQ(IterStatus(iter), "c->vc");
   iter->Prev();
   ASSERT_EQ(IterStatus(iter), "b->" + std::string(100000, 'b'));
   iter->Prev();
   ASSERT_EQ(IterStatus(iter), "a->va");
   iter->Prev();
   ASSERT_EQ(IterStatus(iter), "(invalid)");
 
   delete iter;
 }
 
 TEST(DBTest, IterMultiWithDelete) {
   do {
     ASSERT_OK(Put("a", "va"));
     ASSERT_OK(Put("b", "vb"));
     ASSERT_OK(Put("c", "vc"));
     ASSERT_OK(Delete("b"));
     ASSERT_EQ("NOT_FOUND", Get("b"));
 
     Iterator* iter = db_->NewIterator(ReadOptions());
     iter->Seek("c");
     ASSERT_EQ(IterStatus(iter), "c->vc");
     iter->Prev();
     ASSERT_EQ(IterStatus(iter), "a->va");
     delete iter;
   } while (ChangeOptions());
 }
 
 TEST(DBTest, Recover) {
   do {
     ASSERT_OK(Put("foo", "v1"));
     ASSERT_OK(Put("baz", "v5"));
 
     Reopen();
     ASSERT_EQ("v1", Get("foo"));
 
     ASSERT_EQ("v1", Get("foo"));
     ASSERT_EQ("v5", Get("baz"));
     ASSERT_OK(Put("bar", "v2"));
     ASSERT_OK(Put("foo", "v3"));
 
     Reopen();
     ASSERT_EQ("v3", Get("foo"));
     ASSERT_OK(Put("foo", "v4"));
     ASSERT_EQ("v4", Get("foo"));
     ASSERT_EQ("v2", Get("bar"));
     ASSERT_EQ("v5", Get("baz"));
   } while (ChangeOptions());
 }
 
 TEST(DBTest, RecoveryWithEmptyLog) {
   do {
     ASSERT_OK(Put("foo", "v1"));
     ASSERT_OK(Put("foo", "v2"));
     Reopen();
     Reopen();
     ASSERT_OK(Put("foo", "v3"));
     Reopen();
     ASSERT_EQ("v3", Get("foo"));
   } while (ChangeOptions());
 }
 
 // Check that writes done during a memtable compaction are recovered
 // if the database is shutdown during the memtable compaction.
 TEST(DBTest, RecoverDuringMemtableCompaction) {
   do {
     Options options = CurrentOptions();
     options.env = env_;
     options.write_buffer_size = 1000000;
     Reopen(&options);
 
     // Trigger a long memtable compaction and reopen the database during it
     ASSERT_OK(Put("foo", "v1"));                         // Goes to 1st log file
     ASSERT_OK(Put("big1", std::string(10000000, 'x')));  // Fills memtable
     ASSERT_OK(Put("big2", std::string(1000, 'y')));      // Triggers compaction
     ASSERT_OK(Put("bar", "v2"));                         // Goes to new log file
 
     Reopen(&options);
     ASSERT_EQ("v1", Get("foo"));
     ASSERT_EQ("v2", Get("bar"));
     ASSERT_EQ(std::string(10000000, 'x'), Get("big1"));
     ASSERT_EQ(std::string(1000, 'y'), Get("big2"));
   } while (ChangeOptions());
 }
 
 static std::string Key(int i) {
   char buf[100];
   snprintf(buf, sizeof(buf), "key%06d", i);
   return std::string(buf);
 }
 
 TEST(DBTest, MinorCompactionsHappen) {
   Options options = CurrentOptions();
   options.write_buffer_size = 10000;
   Reopen(&options);
 
   const int N = 500;
 
   int starting_num_tables = TotalTableFiles();
   for (int i = 0; i < N; i++) {
     ASSERT_OK(Put(Key(i), Key(i) + std::string(1000, 'v')));
   }
   int ending_num_tables = TotalTableFiles();
   ASSERT_GT(ending_num_tables, starting_num_tables);
 
   for (int i = 0; i < N; i++) {
     ASSERT_EQ(Key(i) + std::string(1000, 'v'), Get(Key(i)));
   }
 
   Reopen();
 
   for (int i = 0; i < N; i++) {
     ASSERT_EQ(Key(i) + std::string(1000, 'v'), Get(Key(i)));
   }
 }
 
 TEST(DBTest, RecoverWithLargeLog) {
   {
     Options options = CurrentOptions();
     Reopen(&options);
     ASSERT_OK(Put("big1", std::string(200000, '1')));
     ASSERT_OK(Put("big2", std::string(200000, '2')));
     ASSERT_OK(Put("small3", std::string(10, '3')));
     ASSERT_OK(Put("small4", std::string(10, '4')));
     ASSERT_EQ(NumTableFilesAtLevel(0), 0);
   }
 
   // Make sure that if we re-open with a small write buffer size that
   // we flush table files in the middle of a large log file.
   Options options = CurrentOptions();
   options.write_buffer_size = 100000;
   Reopen(&options);
   ASSERT_EQ(NumTableFilesAtLevel(0), 3);
   ASSERT_EQ(std::string(200000, '1'), Get("big1"));
   ASSERT_EQ(std::string(200000, '2'), Get("big2"));
   ASSERT_EQ(std::string(10, '3'), Get("small3"));
   ASSERT_EQ(std::string(10, '4'), Get("small4"));
   ASSERT_GT(NumTableFilesAtLevel(0), 1);
 }
 
 TEST(DBTest, CompactionsGenerateMultipleFiles) {
   Options options = CurrentOptions();
   options.write_buffer_size = 100000000;        // Large write buffer
   Reopen(&options);
 
   Random rnd(301);
 
   // Write 8MB (80 values, each 100K)
   ASSERT_EQ(NumTableFilesAtLevel(0), 0);
   std::vector<std::string> values;
   for (int i = 0; i < 80; i++) {
     values.push_back(RandomString(&rnd, 100000));
     ASSERT_OK(Put(Key(i), values[i]));
   }
 
   // Reopening moves updates to level-0
   Reopen(&options);
   dbfull()->TEST_CompactRange(0, NULL, NULL);
 
   ASSERT_EQ(NumTableFilesAtLevel(0), 0);
   ASSERT_GT(NumTableFilesAtLevel(1), 1);
   for (int i = 0; i < 80; i++) {
     ASSERT_EQ(Get(Key(i)), values[i]);
   }
 }
 
 TEST(DBTest, RepeatedWritesToSameKey) {
   Options options = CurrentOptions();
   options.env = env_;
   options.write_buffer_size = 100000;  // Small write buffer
   Reopen(&options);
 
   // We must have at most one file per level except for level-0,
   // which may have up to kL0_StopWritesTrigger files.
   const int kMaxFiles = config::kNumLevels + config::kL0_StopWritesTrigger;
 
   Random rnd(301);
   std::string value = RandomString(&rnd, 2 * options.write_buffer_size);
   for (int i = 0; i < 5 * kMaxFiles; i++) {
     Put("key", value);
     ASSERT_LE(TotalTableFiles(), kMaxFiles);
     fprintf(stderr, "after %d: %d files\n", int(i+1), TotalTableFiles());
   }
 }
 
 TEST(DBTest, SparseMerge) {
   Options options = CurrentOptions();
   options.compression = kNoCompression;
   Reopen(&options);
 
   FillLevels("A", "Z");
 
   // Suppose there is:
   //    small amount of data with prefix A
   //    large amount of data with prefix B
   //    small amount of data with prefix C
   // and that recent updates have made small changes to all three prefixes.
   // Check that we do not do a compaction that merges all of B in one shot.
   const std::string value(1000, 'x');
   Put("A", "va");
   // Write approximately 100MB of "B" values
   for (int i = 0; i < 100000; i++) {
     char key[100];
     snprintf(key, sizeof(key), "B%010d", i);
     Put(key, value);
   }
   Put("C", "vc");
   dbfull()->TEST_CompactMemTable();
   dbfull()->TEST_CompactRange(0, NULL, NULL);
 
   // Make sparse update
   Put("A",    "va2");
   Put("B100", "bvalue2");
   Put("C",    "vc2");
   dbfull()->TEST_CompactMemTable();
 
   // Compactions should not cause us to create a situation where
   // a file overlaps too much data at the next level.
   ASSERT_LE(dbfull()->TEST_MaxNextLevelOverlappingBytes(), 20*1048576);
   dbfull()->TEST_CompactRange(0, NULL, NULL);
   ASSERT_LE(dbfull()->TEST_MaxNextLevelOverlappingBytes(), 20*1048576);
   dbfull()->TEST_CompactRange(1, NULL, NULL);
   ASSERT_LE(dbfull()->TEST_MaxNextLevelOverlappingBytes(), 20*1048576);
 }
 
 static bool Between(uint64_t val, uint64_t low, uint64_t high) {
   bool result = (val >= low) && (val <= high);
   if (!result) {
     fprintf(stderr, "Value %llu is not in range [%llu, %llu]\n",
             (unsigned long long)(val),
             (unsigned long long)(low),
             (unsigned long long)(high));
   }
   return result;
 }
 
 TEST(DBTest, ApproximateSizes) {
   do {
     Options options = CurrentOptions();
     options.write_buffer_size = 100000000;        // Large write buffer
     options.compression = kNoCompression;
     DestroyAndReopen();
 
     ASSERT_TRUE(Between(Size("", "xyz"), 0, 0));
     Reopen(&options);
     ASSERT_TRUE(Between(Size("", "xyz"), 0, 0));
 
     // Write 8MB (80 values, each 100K)
     ASSERT_EQ(NumTableFilesAtLevel(0), 0);
     const int N = 80;
     static const int S1 = 100000;
     static const int S2 = 105000;  // Allow some expansion from metadata
     Random rnd(301);
     for (int i = 0; i < N; i++) {
       ASSERT_OK(Put(Key(i), RandomString(&rnd, S1)));
     }
 
     // 0 because GetApproximateSizes() does not account for memtable space
     ASSERT_TRUE(Between(Size("", Key(50)), 0, 0));
 
     if (options.reuse_logs) {
       // Recovery will reuse memtable, and GetApproximateSizes() does not
       // account for memtable usage;
       Reopen(&options);
       ASSERT_TRUE(Between(Size("", Key(50)), 0, 0));
       continue;
     }
 
     // Check sizes across recovery by reopening a few times
     for (int run = 0; run < 3; run++) {
       Reopen(&options);
 
       for (int compact_start = 0; compact_start < N; compact_start += 10) {
         for (int i = 0; i < N; i += 10) {
           ASSERT_TRUE(Between(Size("", Key(i)), S1*i, S2*i));
           ASSERT_TRUE(Between(Size("", Key(i)+".suffix"), S1*(i+1), S2*(i+1)));
           ASSERT_TRUE(Between(Size(Key(i), Key(i+10)), S1*10, S2*10));
         }
         ASSERT_TRUE(Between(Size("", Key(50)), S1*50, S2*50));
         ASSERT_TRUE(Between(Size("", Key(50)+".suffix"), S1*50, S2*50));
 
         std::string cstart_str = Key(compact_start);
         std::string cend_str = Key(compact_start + 9);
         Slice cstart = cstart_str;
         Slice cend = cend_str;
         dbfull()->TEST_CompactRange(0, &cstart, &cend);
       }
 
       ASSERT_EQ(NumTableFilesAtLevel(0), 0);
       ASSERT_GT(NumTableFilesAtLevel(1), 0);
     }
   } while (ChangeOptions());
 }
 
 TEST(DBTest, ApproximateSizes_MixOfSmallAndLarge) {
   do {
     Options options = CurrentOptions();
     options.compression = kNoCompression;
     Reopen();
 
     Random rnd(301);
     std::string big1 = RandomString(&rnd, 100000);
     ASSERT_OK(Put(Key(0), RandomString(&rnd, 10000)));
     ASSERT_OK(Put(Key(1), RandomString(&rnd, 10000)));
     ASSERT_OK(Put(Key(2), big1));
     ASSERT_OK(Put(Key(3), RandomString(&rnd, 10000)));
     ASSERT_OK(Put(Key(4), big1));
     ASSERT_OK(Put(Key(5), RandomString(&rnd, 10000)));
     ASSERT_OK(Put(Key(6), RandomString(&rnd, 300000)));
     ASSERT_OK(Put(Key(7), RandomString(&rnd, 10000)));
 
     if (options.reuse_logs) {
       // Need to force a memtable compaction since recovery does not do so.
       ASSERT_OK(dbfull()->TEST_CompactMemTable());
     }
 
     // Check sizes across recovery by reopening a few times
     for (int run = 0; run < 3; run++) {
       Reopen(&options);
 
       ASSERT_TRUE(Between(Size("", Key(0)), 0, 0));
       ASSERT_TRUE(Between(Size("", Key(1)), 10000, 11000));
       ASSERT_TRUE(Between(Size("", Key(2)), 20000, 21000));
       ASSERT_TRUE(Between(Size("", Key(3)), 120000, 121000));
       ASSERT_TRUE(Between(Size("", Key(4)), 130000, 131000));
       ASSERT_TRUE(Between(Size("", Key(5)), 230000, 231000));
       ASSERT_TRUE(Between(Size("", Key(6)), 240000, 241000));
       ASSERT_TRUE(Between(Size("", Key(7)), 540000, 541000));
       ASSERT_TRUE(Between(Size("", Key(8)), 550000, 560000));
 
       ASSERT_TRUE(Between(Size(Key(3), Key(5)), 110000, 111000));
 
       dbfull()->TEST_CompactRange(0, NULL, NULL);
     }
   } while (ChangeOptions());
 }
 
 TEST(DBTest, IteratorPinsRef) {
   Put("foo", "hello");
 
   // Get iterator that will yield the current contents of the DB.
   Iterator* iter = db_->NewIterator(ReadOptions());
 
   // Write to force compactions
   Put("foo", "newvalue1");
   for (int i = 0; i < 100; i++) {
     ASSERT_OK(Put(Key(i), Key(i) + std::string(100000, 'v'))); // 100K values
   }
   Put("foo", "newvalue2");
 
   iter->SeekToFirst();
   ASSERT_TRUE(iter->Valid());
   ASSERT_EQ("foo", iter->key().ToString());
   ASSERT_EQ("hello", iter->value().ToString());
   iter->Next();
   ASSERT_TRUE(!iter->Valid());
   delete iter;
 }
 
 TEST(DBTest, Snapshot) {
   do {
     Put("foo", "v1");
     const Snapshot* s1 = db_->GetSnapshot();
     Put("foo", "v2");
     const Snapshot* s2 = db_->GetSnapshot();
     Put("foo", "v3");
     const Snapshot* s3 = db_->GetSnapshot();
 
     Put("foo", "v4");
     ASSERT_EQ("v1", Get("foo", s1));
     ASSERT_EQ("v2", Get("foo", s2));
     ASSERT_EQ("v3", Get("foo", s3));
     ASSERT_EQ("v4", Get("foo"));
 
     db_->ReleaseSnapshot(s3);
     ASSERT_EQ("v1", Get("foo", s1));
     ASSERT_EQ("v2", Get("foo", s2));
     ASSERT_EQ("v4", Get("foo"));
 
     db_->ReleaseSnapshot(s1);
     ASSERT_EQ("v2", Get("foo", s2));
     ASSERT_EQ("v4", Get("foo"));
 
     db_->ReleaseSnapshot(s2);
     ASSERT_EQ("v4", Get("foo"));
   } while (ChangeOptions());
 }
 
 TEST(DBTest, HiddenValuesAreRemoved) {
   do {
     Random rnd(301);
     FillLevels("a", "z");
 
     std::string big = RandomString(&rnd, 50000);
     Put("foo", big);
     Put("pastfoo", "v");
     const Snapshot* snapshot = db_->GetSnapshot();
     Put("foo", "tiny");
     Put("pastfoo2", "v2");        // Advance sequence number one more
 
     ASSERT_OK(dbfull()->TEST_CompactMemTable());
     ASSERT_GT(NumTableFilesAtLevel(0), 0);
 
     ASSERT_EQ(big, Get("foo", snapshot));
     ASSERT_TRUE(Between(Size("", "pastfoo"), 50000, 60000));
     db_->ReleaseSnapshot(snapshot);
     ASSERT_EQ(AllEntriesFor("foo"), "[ tiny, " + big + " ]");
     Slice x("x");
     dbfull()->TEST_CompactRange(0, NULL, &x);
     ASSERT_EQ(AllEntriesFor("foo"), "[ tiny ]");
     ASSERT_EQ(NumTableFilesAtLevel(0), 0);
     ASSERT_GE(NumTableFilesAtLevel(1), 1);
     dbfull()->TEST_CompactRange(1, NULL, &x);
     ASSERT_EQ(AllEntriesFor("foo"), "[ tiny ]");
 
     ASSERT_TRUE(Between(Size("", "pastfoo"), 0, 1000));
   } while (ChangeOptions());
 }
 
 TEST(DBTest, DeletionMarkers1) {
   Put("foo", "v1");
   ASSERT_OK(dbfull()->TEST_CompactMemTable());
   const int last = config::kMaxMemCompactLevel;
   ASSERT_EQ(NumTableFilesAtLevel(last), 1);   // foo => v1 is now in last level
 
   // Place a table at level last-1 to prevent merging with preceding mutation
   Put("a", "begin");
   Put("z", "end");
   dbfull()->TEST_CompactMemTable();
   ASSERT_EQ(NumTableFilesAtLevel(last), 1);
   ASSERT_EQ(NumTableFilesAtLevel(last-1), 1);
 
   Delete("foo");
   Put("foo", "v2");
   ASSERT_EQ(AllEntriesFor("foo"), "[ v2, DEL, v1 ]");
   ASSERT_OK(dbfull()->TEST_CompactMemTable());  // Moves to level last-2
   ASSERT_EQ(AllEntriesFor("foo"), "[ v2, DEL, v1 ]");
   Slice z("z");
   dbfull()->TEST_CompactRange(last-2, NULL, &z);
   // DEL eliminated, but v1 remains because we aren't compacting that level
   // (DEL can be eliminated because v2 hides v1).
   ASSERT_EQ(AllEntriesFor("foo"), "[ v2, v1 ]");
   dbfull()->TEST_CompactRange(last-1, NULL, NULL);
   // Merging last-1 w/ last, so we are the base level for "foo", so
   // DEL is removed.  (as is v1).
   ASSERT_EQ(AllEntriesFor("foo"), "[ v2 ]");
 }
 
 TEST(DBTest, DeletionMarkers2) {
   Put("foo", "v1");
   ASSERT_OK(dbfull()->TEST_CompactMemTable());
   const int last = config::kMaxMemCompactLevel;
   ASSERT_EQ(NumTableFilesAtLevel(last), 1);   // foo => v1 is now in last level
 
   // Place a table at level last-1 to prevent merging with preceding mutation
   Put("a", "begin");
   Put("z", "end");
   dbfull()->TEST_CompactMemTable();
   ASSERT_EQ(NumTableFilesAtLevel(last), 1);
   ASSERT_EQ(NumTableFilesAtLevel(last-1), 1);
 
   Delete("foo");
   ASSERT_EQ(AllEntriesFor("foo"), "[ DEL, v1 ]");
   ASSERT_OK(dbfull()->TEST_CompactMemTable());  // Moves to level last-2
   ASSERT_EQ(AllEntriesFor("foo"), "[ DEL, v1 ]");
   dbfull()->TEST_CompactRange(last-2, NULL, NULL);
   // DEL kept: "last" file overlaps
   ASSERT_EQ(AllEntriesFor("foo"), "[ DEL, v1 ]");
   dbfull()->TEST_CompactRange(last-1, NULL, NULL);
   // Merging last-1 w/ last, so we are the base level for "foo", so
   // DEL is removed.  (as is v1).
   ASSERT_EQ(AllEntriesFor("foo"), "[ ]");
 }
 
 TEST(DBTest, OverlapInLevel0) {
   do {
     ASSERT_EQ(config::kMaxMemCompactLevel, 2) << "Fix test to match config";
 
     // Fill levels 1 and 2 to disable the pushing of new memtables to levels > 0.
     ASSERT_OK(Put("100", "v100"));
     ASSERT_OK(Put("999", "v999"));
     dbfull()->TEST_CompactMemTable();
     ASSERT_OK(Delete("100"));
     ASSERT_OK(Delete("999"));
     dbfull()->TEST_CompactMemTable();
     ASSERT_EQ("0,1,1", FilesPerLevel());
 
     // Make files spanning the following ranges in level-0:
     //  files[0]  200 .. 900
     //  files[1]  300 .. 500
     // Note that files are sorted by smallest key.
     ASSERT_OK(Put("300", "v300"));
     ASSERT_OK(Put("500", "v500"));
     dbfull()->TEST_CompactMemTable();
     ASSERT_OK(Put("200", "v200"));
     ASSERT_OK(Put("600", "v600"));
     ASSERT_OK(Put("900", "v900"));
     dbfull()->TEST_CompactMemTable();
     ASSERT_EQ("2,1,1", FilesPerLevel());
 
     // Compact away the placeholder files we created initially
     dbfull()->TEST_CompactRange(1, NULL, NULL);
     dbfull()->TEST_CompactRange(2, NULL, NULL);
     ASSERT_EQ("2", FilesPerLevel());
 
     // Do a memtable compaction.  Before bug-fix, the compaction would
     // not detect the overlap with level-0 files and would incorrectly place
     // the deletion in a deeper level.
     ASSERT_OK(Delete("600"));
     dbfull()->TEST_CompactMemTable();
     ASSERT_EQ("3", FilesPerLevel());
     ASSERT_EQ("NOT_FOUND", Get("600"));
   } while (ChangeOptions());
 }
 
 TEST(DBTest, L0_CompactionBug_Issue44_a) {
   Reopen();
   ASSERT_OK(Put("b", "v"));
   Reopen();
   ASSERT_OK(Delete("b"));
   ASSERT_OK(Delete("a"));
   Reopen();
   ASSERT_OK(Delete("a"));
   Reopen();
   ASSERT_OK(Put("a", "v"));
   Reopen();
   Reopen();
   ASSERT_EQ("(a->v)", Contents());
   DelayMilliseconds(1000);  // Wait for compaction to finish
   ASSERT_EQ("(a->v)", Contents());
 }
 
 TEST(DBTest, L0_CompactionBug_Issue44_b) {
   Reopen();
   Put("","");
   Reopen();
   Delete("e");
   Put("","");
   Reopen();
   Put("c", "cv");
   Reopen();
   Put("","");
   Reopen();
   Put("","");
   DelayMilliseconds(1000);  // Wait for compaction to finish
   Reopen();
   Put("d","dv");
   Reopen();
   Put("","");
   Reopen();
   Delete("d");
   Delete("b");
   Reopen();
   ASSERT_EQ("(->)(c->cv)", Contents());
   DelayMilliseconds(1000);  // Wait for compaction to finish
   ASSERT_EQ("(->)(c->cv)", Contents());
 }
 
 TEST(DBTest, ComparatorCheck) {
   class NewComparator : public Comparator {
    public:
     virtual const char* Name() const { return "leveldb.NewComparator"; }
     virtual int Compare(const Slice& a, const Slice& b) const {
       return BytewiseComparator()->Compare(a, b);
     }
     virtual void FindShortestSeparator(std::string* s, const Slice& l) const {
       BytewiseComparator()->FindShortestSeparator(s, l);
     }
     virtual void FindShortSuccessor(std::string* key) const {
       BytewiseComparator()->FindShortSuccessor(key);
     }
   };
   NewComparator cmp;
   Options new_options = CurrentOptions();
   new_options.comparator = &cmp;
   Status s = TryReopen(&new_options);
   ASSERT_TRUE(!s.ok());
   ASSERT_TRUE(s.ToString().find("comparator") != std::string::npos)
       << s.ToString();
 }
 
 TEST(DBTest, CustomComparator) {
   class NumberComparator : public Comparator {
    public:
     virtual const char* Name() const { return "test.NumberComparator"; }
     virtual int Compare(const Slice& a, const Slice& b) const {
       return ToNumber(a) - ToNumber(b);
     }
     virtual void FindShortestSeparator(std::string* s, const Slice& l) const {
       ToNumber(*s);     // Check format
       ToNumber(l);      // Check format
     }
     virtual void FindShortSuccessor(std::string* key) const {
       ToNumber(*key);   // Check format
     }
    private:
     static int ToNumber(const Slice& x) {
       // Check that there are no extra characters.
       ASSERT_TRUE(x.size() >= 2 && x[0] == '[' && x[x.size()-1] == ']')
           << EscapeString(x);
       int val;
       char ignored;
       ASSERT_TRUE(sscanf(x.ToString().c_str(), "[%i]%c", &val, &ignored) == 1)
           << EscapeString(x);
       return val;
     }
   };
   NumberComparator cmp;
   Options new_options = CurrentOptions();
   new_options.create_if_missing = true;
   new_options.comparator = &cmp;
   new_options.filter_policy = NULL;     // Cannot use bloom filters
   new_options.write_buffer_size = 1000;  // Compact more often
   DestroyAndReopen(&new_options);
   ASSERT_OK(Put("[10]", "ten"));
   ASSERT_OK(Put("[0x14]", "twenty"));
   for (int i = 0; i < 2; i++) {
     ASSERT_EQ("ten", Get("[10]"));
     ASSERT_EQ("ten", Get("[0xa]"));
     ASSERT_EQ("twenty", Get("[20]"));
     ASSERT_EQ("twenty", Get("[0x14]"));
     ASSERT_EQ("NOT_FOUND", Get("[15]"));
     ASSERT_EQ("NOT_FOUND", Get("[0xf]"));
     Compact("[0]", "[9999]");
   }
 
   for (int run = 0; run < 2; run++) {
     for (int i = 0; i < 1000; i++) {
       char buf[100];
       snprintf(buf, sizeof(buf), "[%d]", i*10);
       ASSERT_OK(Put(buf, buf));
     }
     Compact("[0]", "[1000000]");
   }
 }
 
 TEST(DBTest, ManualCompaction) {
   ASSERT_EQ(config::kMaxMemCompactLevel, 2)
       << "Need to update this test to match kMaxMemCompactLevel";
 
   MakeTables(3, "p", "q");
   ASSERT_EQ("1,1,1", FilesPerLevel());
 
   // Compaction range falls before files
   Compact("", "c");
   ASSERT_EQ("1,1,1", FilesPerLevel());
 
   // Compaction range falls after files
   Compact("r", "z");
   ASSERT_EQ("1,1,1", FilesPerLevel());
 
   // Compaction range overlaps files
   Compact("p1", "p9");
   ASSERT_EQ("0,0,1", FilesPerLevel());
 
   // Populate a different range
   MakeTables(3, "c", "e");
   ASSERT_EQ("1,1,2", FilesPerLevel());
 
   // Compact just the new range
   Compact("b", "f");
   ASSERT_EQ("0,0,2", FilesPerLevel());
 
   // Compact all
   MakeTables(1, "a", "z");
   ASSERT_EQ("0,1,2", FilesPerLevel());
   db_->CompactRange(NULL, NULL);
   ASSERT_EQ("0,0,1", FilesPerLevel());
 }
 
 TEST(DBTest, DBOpen_Options) {
   std::string dbname = test::TmpDir() + "/db_options_test";
   DestroyDB(dbname, Options());
 
   // Does not exist, and create_if_missing == false: error
   DB* db = NULL;
   Options opts;
   opts.create_if_missing = false;
   Status s = DB::Open(opts, dbname, &db);
   ASSERT_TRUE(strstr(s.ToString().c_str(), "does not exist") != NULL);
   ASSERT_TRUE(db == NULL);
 
   // Does not exist, and create_if_missing == true: OK
   opts.create_if_missing = true;
   s = DB::Open(opts, dbname, &db);
   ASSERT_OK(s);
   ASSERT_TRUE(db != NULL);
 
   delete db;
   db = NULL;
 
   // Does exist, and error_if_exists == true: error
   opts.create_if_missing = false;
   opts.error_if_exists = true;
   s = DB::Open(opts, dbname, &db);
   ASSERT_TRUE(strstr(s.ToString().c_str(), "exists") != NULL);
   ASSERT_TRUE(db == NULL);
 
   // Does exist, and error_if_exists == false: OK
   opts.create_if_missing = true;
   opts.error_if_exists = false;
   s = DB::Open(opts, dbname, &db);
   ASSERT_OK(s);
   ASSERT_TRUE(db != NULL);
 
   delete db;
   db = NULL;
 }
 
 TEST(DBTest, Locking) {
   DB* db2 = NULL;
   Status s = DB::Open(CurrentOptions(), dbname_, &db2);
   ASSERT_TRUE(!s.ok()) << "Locking did not prevent re-opening db";
 }
 
 // Check that number of files does not grow when we are out of space
 TEST(DBTest, NoSpace) {
   Options options = CurrentOptions();
   options.env = env_;
   Reopen(&options);
 
   ASSERT_OK(Put("foo", "v1"));
   ASSERT_EQ("v1", Get("foo"));
   Compact("a", "z");
   const int num_files = CountFiles();
   env_->no_space_.Release_Store(env_);   // Force out-of-space errors
   for (int i = 0; i < 10; i++) {
     for (int level = 0; level < config::kNumLevels-1; level++) {
       dbfull()->TEST_CompactRange(level, NULL, NULL);
     }
   }
   env_->no_space_.Release_Store(NULL);
   ASSERT_LT(CountFiles(), num_files + 3);
 }
 
 TEST(DBTest, NonWritableFileSystem) {
   Options options = CurrentOptions();
   options.write_buffer_size = 1000;
   options.env = env_;
   Reopen(&options);
   ASSERT_OK(Put("foo", "v1"));
   env_->non_writable_.Release_Store(env_);  // Force errors for new files
   std::string big(100000, 'x');
   int errors = 0;
   for (int i = 0; i < 20; i++) {
     fprintf(stderr, "iter %d; errors %d\n", i, errors);
     if (!Put("foo", big).ok()) {
       errors++;
       DelayMilliseconds(100);
     }
   }
   ASSERT_GT(errors, 0);
   env_->non_writable_.Release_Store(NULL);
 }
 
 TEST(DBTest, WriteSyncError) {
   // Check that log sync errors cause the DB to disallow future writes.
 
   // (a) Cause log sync calls to fail
   Options options = CurrentOptions();
   options.env = env_;
   Reopen(&options);
   env_->data_sync_error_.Release_Store(env_);
 
   // (b) Normal write should succeed
   WriteOptions w;
   ASSERT_OK(db_->Put(w, "k1", "v1"));
   ASSERT_EQ("v1", Get("k1"));
 
   // (c) Do a sync write; should fail
   w.sync = true;
   ASSERT_TRUE(!db_->Put(w, "k2", "v2").ok());
   ASSERT_EQ("v1", Get("k1"));
   ASSERT_EQ("NOT_FOUND", Get("k2"));
 
   // (d) make sync behave normally
   env_->data_sync_error_.Release_Store(NULL);
 
   // (e) Do a non-sync write; should fail
   w.sync = false;
   ASSERT_TRUE(!db_->Put(w, "k3", "v3").ok());
   ASSERT_EQ("v1", Get("k1"));
   ASSERT_EQ("NOT_FOUND", Get("k2"));
   ASSERT_EQ("NOT_FOUND", Get("k3"));
 }
 
 TEST(DBTest, ManifestWriteError) {
   // Test for the following problem:
   // (a) Compaction produces file F
   // (b) Log record containing F is written to MANIFEST file, but Sync() fails
   // (c) GC deletes F
   // (d) After reopening DB, reads fail since deleted F is named in log record
 
   // We iterate twice.  In the second iteration, everything is the
   // same except the log record never makes it to the MANIFEST file.
   for (int iter = 0; iter < 2; iter++) {
     port::AtomicPointer* error_type = (iter == 0)
         ? &env_->manifest_sync_error_
         : &env_->manifest_write_error_;
 
     // Insert foo=>bar mapping
     Options options = CurrentOptions();
     options.env = env_;
     options.create_if_missing = true;
     options.error_if_exists = false;
     DestroyAndReopen(&options);
     ASSERT_OK(Put("foo", "bar"));
     ASSERT_EQ("bar", Get("foo"));
 
     // Memtable compaction (will succeed)
     dbfull()->TEST_CompactMemTable();
     ASSERT_EQ("bar", Get("foo"));
     const int last = config::kMaxMemCompactLevel;
     ASSERT_EQ(NumTableFilesAtLevel(last), 1);   // foo=>bar is now in last level
 
     // Merging compaction (will fail)
     error_type->Release_Store(env_);
     dbfull()->TEST_CompactRange(last, NULL, NULL);  // Should fail
     ASSERT_EQ("bar", Get("foo"));
 
     // Recovery: should not lose data
     error_type->Release_Store(NULL);
     Reopen(&options);
     ASSERT_EQ("bar", Get("foo"));
   }
 }
 
 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"));
 
   Close();
   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, StillReadSST) {
   ASSERT_OK(Put("foo", "bar"));
   ASSERT_EQ("bar", Get("foo"));
 
   // Dump the memtable to disk.
   dbfull()->TEST_CompactMemTable();
   ASSERT_EQ("bar", Get("foo"));
   Close();
   ASSERT_GT(RenameLDBToSST(), 0);
   Options options = CurrentOptions();
   options.paranoid_checks = true;
   Status s = TryReopen(&options);
   ASSERT_TRUE(s.ok());
   ASSERT_EQ("bar", Get("foo"));
 }
 
 TEST(DBTest, FilesDeletedAfterCompaction) {
   ASSERT_OK(Put("foo", "v2"));
   Compact("a", "z");
   const int num_files = CountFiles();
   for (int i = 0; i < 10; i++) {
     ASSERT_OK(Put("foo", "v2"));
     Compact("a", "z");
   }
   ASSERT_EQ(CountFiles(), num_files);
 }
 
 TEST(DBTest, BloomFilter) {
   env_->count_random_reads_ = true;
   Options options = CurrentOptions();
   options.env = env_;
   options.block_cache = NewLRUCache(0);  // Prevent cache hits
   options.filter_policy = NewBloomFilterPolicy(10);
   Reopen(&options);
 
   // Populate multiple layers
   const int N = 10000;
   for (int i = 0; i < N; i++) {
     ASSERT_OK(Put(Key(i), Key(i)));
   }
   Compact("a", "z");
   for (int i = 0; i < N; i += 100) {
     ASSERT_OK(Put(Key(i), Key(i)));
   }
   dbfull()->TEST_CompactMemTable();
 
   // Prevent auto compactions triggered by seeks
   env_->delay_data_sync_.Release_Store(env_);
 
   // Lookup present keys.  Should rarely read from small sstable.
   env_->random_read_counter_.Reset();
   for (int i = 0; i < N; i++) {
     ASSERT_EQ(Key(i), Get(Key(i)));
   }
   int reads = env_->random_read_counter_.Read();
   fprintf(stderr, "%d present => %d reads\n", N, reads);
   ASSERT_GE(reads, N);
   ASSERT_LE(reads, N + 2*N/100);
 
   // Lookup present keys.  Should rarely read from either sstable.
   env_->random_read_counter_.Reset();
   for (int i = 0; i < N; i++) {
     ASSERT_EQ("NOT_FOUND", Get(Key(i) + ".missing"));
   }
   reads = env_->random_read_counter_.Read();
   fprintf(stderr, "%d missing => %d reads\n", N, reads);
   ASSERT_LE(reads, 3*N/100);
 
   env_->delay_data_sync_.Release_Store(NULL);
   Close();
   delete options.block_cache;
   delete options.filter_policy;
 }
 
 // Multi-threaded test:
 namespace {
 
 static const int kNumThreads = 4;
 static const int kTestSeconds = 10;
 static const int kNumKeys = 1000;
 
 struct MTState {
   DBTest* test;
   port::AtomicPointer stop;
   port::AtomicPointer counter[kNumThreads];
   port::AtomicPointer thread_done[kNumThreads];
 };
 
 struct MTThread {
   MTState* state;
   int id;
 };
 
 static void MTThreadBody(void* arg) {
   MTThread* t = reinterpret_cast<MTThread*>(arg);
   int id = t->id;
   DB* db = t->state->test->db_;
   uintptr_t counter = 0;
   fprintf(stderr, "... starting thread %d\n", id);
   Random rnd(1000 + id);
   std::string value;
   char valbuf[1500];
   while (t->state->stop.Acquire_Load() == NULL) {
     t->state->counter[id].Release_Store(reinterpret_cast<void*>(counter));
 
     int key = rnd.Uniform(kNumKeys);
     char keybuf[20];
     snprintf(keybuf, sizeof(keybuf), "%016d", key);
 
     if (rnd.OneIn(2)) {
       // Write values of the form <key, my id, counter>.
       // We add some padding for force compactions.
       snprintf(valbuf, sizeof(valbuf), "%d.%d.%-1000d",
                key, id, static_cast<int>(counter));
       ASSERT_OK(db->Put(WriteOptions(), Slice(keybuf), Slice(valbuf)));
     } else {
       // Read a value and verify that it matches the pattern written above.
       Status s = db->Get(ReadOptions(), Slice(keybuf), &value);
       if (s.IsNotFound()) {
         // Key has not yet been written
       } else {
         // Check that the writer thread counter is >= the counter in the value
         ASSERT_OK(s);
         int k, w, c;
         ASSERT_EQ(3, sscanf(value.c_str(), "%d.%d.%d", &k, &w, &c)) << value;
         ASSERT_EQ(k, key);
         ASSERT_GE(w, 0);
         ASSERT_LT(w, kNumThreads);
         ASSERT_LE(static_cast<uintptr_t>(c), reinterpret_cast<uintptr_t>(
             t->state->counter[w].Acquire_Load()));
       }
     }
     counter++;
   }
   t->state->thread_done[id].Release_Store(t);
   fprintf(stderr, "... stopping thread %d after %d ops\n", id, int(counter));
 }
 
 }  // namespace
 
 TEST(DBTest, MultiThreaded) {
   do {
     // Initialize state
     MTState mt;
     mt.test = this;
     mt.stop.Release_Store(0);
     for (int id = 0; id < kNumThreads; id++) {
       mt.counter[id].Release_Store(0);
       mt.thread_done[id].Release_Store(0);
     }
 
     // Start threads
     MTThread thread[kNumThreads];
     for (int id = 0; id < kNumThreads; id++) {
       thread[id].state = &mt;
       thread[id].id = id;
       env_->StartThread(MTThreadBody, &thread[id]);
     }
 
     // Let them run for a while
     DelayMilliseconds(kTestSeconds * 1000);
 
     // Stop the threads and wait for them to finish
     mt.stop.Release_Store(&mt);
     for (int id = 0; id < kNumThreads; id++) {
       while (mt.thread_done[id].Acquire_Load() == NULL) {
         DelayMilliseconds(100);
       }
     }
   } while (ChangeOptions());
 }
 
 namespace {
 typedef std::map<std::string, std::string> KVMap;
 }
 
 class ModelDB: public DB {
  public:
   class ModelSnapshot : public Snapshot {
    public:
     KVMap map_;
   };
 
   explicit ModelDB(const Options& options): options_(options) { }
   ~ModelDB() { }
   virtual Status Put(const WriteOptions& o, const Slice& k, const Slice& v) {
     return DB::Put(o, k, v);
   }
   virtual Status Delete(const WriteOptions& o, const Slice& key) {
     return DB::Delete(o, key);
   }
   virtual Status Get(const ReadOptions& options,
                      const Slice& key, std::string* value) {
     assert(false);      // Not implemented
     return Status::NotFound(key);
   }
   virtual Iterator* NewIterator(const ReadOptions& options) {
     if (options.snapshot == NULL) {
       KVMap* saved = new KVMap;
       *saved = map_;
       return new ModelIter(saved, true);
     } else {
       const KVMap* snapshot_state =
           &(reinterpret_cast<const ModelSnapshot*>(options.snapshot)->map_);
       return new ModelIter(snapshot_state, false);
     }
   }
   virtual const Snapshot* GetSnapshot() {
     ModelSnapshot* snapshot = new ModelSnapshot;
     snapshot->map_ = map_;
     return snapshot;
   }
 
   virtual void ReleaseSnapshot(const Snapshot* snapshot) {
     delete reinterpret_cast<const ModelSnapshot*>(snapshot);
   }
   virtual Status Write(const WriteOptions& options, WriteBatch* batch) {
     class Handler : public WriteBatch::Handler {
      public:
       KVMap* map_;
       virtual void Put(const Slice& key, const Slice& value) {
         (*map_)[key.ToString()] = value.ToString();
       }
       virtual void Delete(const Slice& key) {
         map_->erase(key.ToString());
       }
     };
     Handler handler;
     handler.map_ = &map_;
     return batch->Iterate(&handler);
   }
 
   virtual bool GetProperty(const Slice& property, std::string* value) {
     return false;
   }
   virtual void GetApproximateSizes(const Range* r, int n, uint64_t* sizes) {
     for (int i = 0; i < n; i++) {
       sizes[i] = 0;
     }
   }
   virtual void CompactRange(const Slice* start, const Slice* end) {
   }
 
  private:
   class ModelIter: public Iterator {
    public:
     ModelIter(const KVMap* map, bool owned)
         : map_(map), owned_(owned), iter_(map_->end()) {
     }
     ~ModelIter() {
       if (owned_) delete map_;
     }
     virtual bool Valid() const { return iter_ != map_->end(); }
     virtual void SeekToFirst() { iter_ = map_->begin(); }
     virtual void SeekToLast() {
       if (map_->empty()) {
         iter_ = map_->end();
       } else {
         iter_ = map_->find(map_->rbegin()->first);
       }
     }
     virtual void Seek(const Slice& k) {
       iter_ = map_->lower_bound(k.ToString());
     }
     virtual void Next() { ++iter_; }
     virtual void Prev() { --iter_; }
     virtual Slice key() const { return iter_->first; }
     virtual Slice value() const { return iter_->second; }
     virtual Status status() const { return Status::OK(); }
    private:
     const KVMap* const map_;
     const bool owned_;  // Do we own map_
     KVMap::const_iterator iter_;
   };
   const Options options_;
   KVMap map_;
 };
 
 static std::string RandomKey(Random* rnd) {
   int len = (rnd->OneIn(3)
              ? 1                // Short sometimes to encourage collisions
              : (rnd->OneIn(100) ? rnd->Skewed(10) : rnd->Uniform(10)));
   return test::RandomKey(rnd, len);
 }
 
 static bool CompareIterators(int step,
                              DB* model,
                              DB* db,
                              const Snapshot* model_snap,
                              const Snapshot* db_snap) {
   ReadOptions options;
   options.snapshot = model_snap;
   Iterator* miter = model->NewIterator(options);
   options.snapshot = db_snap;
   Iterator* dbiter = db->NewIterator(options);
   bool ok = true;
   int count = 0;
   for (miter->SeekToFirst(), dbiter->SeekToFirst();
        ok && miter->Valid() && dbiter->Valid();
        miter->Next(), dbiter->Next()) {
     count++;
     if (miter->key().compare(dbiter->key()) != 0) {
       fprintf(stderr, "step %d: Key mismatch: '%s' vs. '%s'\n",
               step,
               EscapeString(miter->key()).c_str(),
               EscapeString(dbiter->key()).c_str());
       ok = false;
       break;
     }
 
     if (miter->value().compare(dbiter->value()) != 0) {
       fprintf(stderr, "step %d: Value mismatch for key '%s': '%s' vs. '%s'\n",
               step,
               EscapeString(miter->key()).c_str(),
               EscapeString(miter->value()).c_str(),
               EscapeString(miter->value()).c_str());
       ok = false;
     }
   }
 
   if (ok) {
     if (miter->Valid() != dbiter->Valid()) {
       fprintf(stderr, "step %d: Mismatch at end of iterators: %d vs. %d\n",
               step, miter->Valid(), dbiter->Valid());
       ok = false;
     }
   }
   fprintf(stderr, "%d entries compared: ok=%d\n", count, ok);
   delete miter;
   delete dbiter;
   return ok;
 }
 
 TEST(DBTest, Randomized) {
   Random rnd(test::RandomSeed());
   do {
     ModelDB model(CurrentOptions());
     const int N = 10000;
     const Snapshot* model_snap = NULL;
     const Snapshot* db_snap = NULL;
     std::string k, v;
     for (int step = 0; step < N; step++) {
       if (step % 100 == 0) {
         fprintf(stderr, "Step %d of %d\n", step, N);
       }
       // TODO(sanjay): Test Get() works
       int p = rnd.Uniform(100);
       if (p < 45) {                               // Put
         k = RandomKey(&rnd);
         v = RandomString(&rnd,
                          rnd.OneIn(20)
                          ? 100 + rnd.Uniform(100)
                          : rnd.Uniform(8));
         ASSERT_OK(model.Put(WriteOptions(), k, v));
         ASSERT_OK(db_->Put(WriteOptions(), k, v));
 
       } else if (p < 90) {                        // Delete
         k = RandomKey(&rnd);
         ASSERT_OK(model.Delete(WriteOptions(), k));
         ASSERT_OK(db_->Delete(WriteOptions(), k));
 
 
       } else {                                    // Multi-element batch
         WriteBatch b;
         const int num = rnd.Uniform(8);
         for (int i = 0; i < num; i++) {
           if (i == 0 || !rnd.OneIn(10)) {
             k = RandomKey(&rnd);
           } else {
             // Periodically re-use the same key from the previous iter, so
             // we have multiple entries in the write batch for the same key
           }
           if (rnd.OneIn(2)) {
             v = RandomString(&rnd, rnd.Uniform(10));
             b.Put(k, v);
           } else {
             b.Delete(k);
           }
         }
         ASSERT_OK(model.Write(WriteOptions(), &b));
         ASSERT_OK(db_->Write(WriteOptions(), &b));
       }
 
       if ((step % 100) == 0) {
         ASSERT_TRUE(CompareIterators(step, &model, db_, NULL, NULL));
         ASSERT_TRUE(CompareIterators(step, &model, db_, model_snap, db_snap));
         // Save a snapshot from each DB this time that we'll use next
         // time we compare things, to make sure the current state is
         // preserved with the snapshot
         if (model_snap != NULL) model.ReleaseSnapshot(model_snap);
         if (db_snap != NULL) db_->ReleaseSnapshot(db_snap);
 
         Reopen();
         ASSERT_TRUE(CompareIterators(step, &model, db_, NULL, NULL));
 
         model_snap = model.GetSnapshot();
         db_snap = db_->GetSnapshot();
       }
     }
     if (model_snap != NULL) model.ReleaseSnapshot(model_snap);
     if (db_snap != NULL) db_->ReleaseSnapshot(db_snap);
   } while (ChangeOptions());
 }
 
 std::string MakeKey(unsigned int num) {
   char buf[30];
   snprintf(buf, sizeof(buf), "%016u", num);
   return std::string(buf);
 }
 
 void BM_LogAndApply(int iters, int num_base_files) {
   std::string dbname = test::TmpDir() + "/leveldb_test_benchmark";
   DestroyDB(dbname, Options());
 
   DB* db = NULL;
   Options opts;
   opts.create_if_missing = true;
   Status s = DB::Open(opts, dbname, &db);
   ASSERT_OK(s);
   ASSERT_TRUE(db != NULL);
 
   delete db;
   db = NULL;
 
   Env* env = Env::Default();
 
   port::Mutex mu;
   MutexLock l(&mu);
 
   InternalKeyComparator cmp(BytewiseComparator());
   Options options;
   VersionSet vset(dbname, &options, NULL, &cmp);
   bool save_manifest;
   ASSERT_OK(vset.Recover(&save_manifest));
   VersionEdit vbase;
   uint64_t fnum = 1;
   for (int i = 0; i < num_base_files; i++) {
     InternalKey start(MakeKey(2*fnum), 1, kTypeValue);
     InternalKey limit(MakeKey(2*fnum+1), 1, kTypeDeletion);
     vbase.AddFile(2, fnum++, 1 /* file size */, start, limit);
   }
   ASSERT_OK(vset.LogAndApply(&vbase, &mu));
 
   uint64_t start_micros = env->NowMicros();
 
   for (int i = 0; i < iters; i++) {
     VersionEdit vedit;
     vedit.DeleteFile(2, fnum);
     InternalKey start(MakeKey(2*fnum), 1, kTypeValue);
     InternalKey limit(MakeKey(2*fnum+1), 1, kTypeDeletion);
     vedit.AddFile(2, fnum++, 1 /* file size */, start, limit);
     vset.LogAndApply(&vedit, &mu);
   }
   uint64_t stop_micros = env->NowMicros();
   unsigned int us = stop_micros - start_micros;
   char buf[16];
   snprintf(buf, sizeof(buf), "%d", num_base_files);
   fprintf(stderr,
           "BM_LogAndApply/%-6s   %8d iters : %9u us (%7.0f us / iter)\n",
           buf, iters, us, ((float)us) / iters);
 }
 
 }  // namespace leveldb
 
 int main(int argc, char** argv) {
   if (argc > 1 && std::string(argv[1]) == "--benchmark") {
     leveldb::BM_LogAndApply(1000, 1);
     leveldb::BM_LogAndApply(1000, 100);
     leveldb::BM_LogAndApply(1000, 10000);
     leveldb::BM_LogAndApply(100, 100000);
     return 0;
   }
 
   return leveldb::test::RunAllTests();
 }
diff --git a/src/leveldb/db/fault_injection_test.cc b/src/leveldb/db/fault_injection_test.cc
index 875dfe81e..236be71d6 100644
--- a/src/leveldb/db/fault_injection_test.cc
+++ b/src/leveldb/db/fault_injection_test.cc
@@ -1,554 +1,560 @@
 // Copyright 2014 The LevelDB Authors. All rights reserved.
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file. See the AUTHORS file for names of contributors.
 
 // This test uses a custom Env to keep track of the state of a filesystem as of
 // the last "sync". It then checks for data loss errors by purposely dropping
 // file data (or entire files) not protected by a "sync".
 
 #include "leveldb/db.h"
 
 #include <map>
 #include <set>
 #include "db/db_impl.h"
 #include "db/filename.h"
 #include "db/log_format.h"
 #include "db/version_set.h"
 #include "leveldb/cache.h"
 #include "leveldb/env.h"
 #include "leveldb/table.h"
 #include "leveldb/write_batch.h"
 #include "util/logging.h"
 #include "util/mutexlock.h"
 #include "util/testharness.h"
 #include "util/testutil.h"
 
 namespace leveldb {
 
 static const int kValueSize = 1000;
 static const int kMaxNumValues = 2000;
 static const size_t kNumIterations = 3;
 
 class FaultInjectionTestEnv;
 
 namespace {
 
 // Assume a filename, and not a directory name like "/foo/bar/"
 static std::string GetDirName(const std::string filename) {
   size_t found = filename.find_last_of("/\\");
   if (found == std::string::npos) {
     return "";
   } else {
     return filename.substr(0, found);
   }
 }
 
 Status SyncDir(const std::string& dir) {
   // As this is a test it isn't required to *actually* sync this directory.
   return Status::OK();
 }
 
 // A basic file truncation function suitable for this test.
 Status Truncate(const std::string& filename, uint64_t length) {
   leveldb::Env* env = leveldb::Env::Default();
 
   SequentialFile* orig_file;
   Status s = env->NewSequentialFile(filename, &orig_file);
   if (!s.ok())
     return s;
 
   char* scratch = new char[length];
   leveldb::Slice result;
   s = orig_file->Read(length, &result, scratch);
   delete orig_file;
   if (s.ok()) {
     std::string tmp_name = GetDirName(filename) + "/truncate.tmp";
     WritableFile* tmp_file;
     s = env->NewWritableFile(tmp_name, &tmp_file);
     if (s.ok()) {
       s = tmp_file->Append(result);
       delete tmp_file;
       if (s.ok()) {
         s = env->RenameFile(tmp_name, filename);
       } else {
         env->DeleteFile(tmp_name);
       }
     }
   }
 
   delete[] scratch;
 
   return s;
 }
 
 struct FileState {
   std::string filename_;
   ssize_t pos_;
   ssize_t pos_at_last_sync_;
   ssize_t pos_at_last_flush_;
 
   FileState(const std::string& filename)
       : filename_(filename),
         pos_(-1),
         pos_at_last_sync_(-1),
         pos_at_last_flush_(-1) { }
 
   FileState() : pos_(-1), pos_at_last_sync_(-1), pos_at_last_flush_(-1) {}
 
   bool IsFullySynced() const { return pos_ <= 0 || pos_ == pos_at_last_sync_; }
 
   Status DropUnsyncedData() const;
 };
 
 }  // anonymous namespace
 
 // A wrapper around WritableFile which informs another Env whenever this file
 // is written to or sync'ed.
 class TestWritableFile : public WritableFile {
  public:
   TestWritableFile(const FileState& state,
                    WritableFile* f,
                    FaultInjectionTestEnv* env);
   virtual ~TestWritableFile();
   virtual Status Append(const Slice& data);
   virtual Status Close();
   virtual Status Flush();
   virtual Status Sync();
+  std::string GetName() const override;
 
  private:
   FileState state_;
   WritableFile* target_;
   bool writable_file_opened_;
   FaultInjectionTestEnv* env_;
 
   Status SyncParent();
 };
 
 class FaultInjectionTestEnv : public EnvWrapper {
  public:
   FaultInjectionTestEnv() : EnvWrapper(Env::Default()), filesystem_active_(true) {}
   virtual ~FaultInjectionTestEnv() { }
   virtual Status NewWritableFile(const std::string& fname,
                                  WritableFile** result);
   virtual Status NewAppendableFile(const std::string& fname,
                                    WritableFile** result);
   virtual Status DeleteFile(const std::string& f);
   virtual Status RenameFile(const std::string& s, const std::string& t);
 
   void WritableFileClosed(const FileState& state);
   Status DropUnsyncedFileData();
   Status DeleteFilesCreatedAfterLastDirSync();
   void DirWasSynced();
   bool IsFileCreatedSinceLastDirSync(const std::string& filename);
   void ResetState();
   void UntrackFile(const std::string& f);
   // Setting the filesystem to inactive is the test equivalent to simulating a
   // system reset. Setting to inactive will freeze our saved filesystem state so
   // that it will stop being recorded. It can then be reset back to the state at
   // the time of the reset.
   bool IsFilesystemActive() const { return filesystem_active_; }
   void SetFilesystemActive(bool active) { filesystem_active_ = active; }
 
  private:
   port::Mutex mutex_;
   std::map<std::string, FileState> db_file_state_;
   std::set<std::string> new_files_since_last_dir_sync_;
   bool filesystem_active_;  // Record flushes, syncs, writes
 };
 
 TestWritableFile::TestWritableFile(const FileState& state,
                                    WritableFile* f,
                                    FaultInjectionTestEnv* env)
     : state_(state),
       target_(f),
       writable_file_opened_(true),
       env_(env) {
   assert(f != NULL);
 }
 
 TestWritableFile::~TestWritableFile() {
   if (writable_file_opened_) {
     Close();
   }
   delete target_;
 }
 
 Status TestWritableFile::Append(const Slice& data) {
   Status s = target_->Append(data);
   if (s.ok() && env_->IsFilesystemActive()) {
     state_.pos_ += data.size();
   }
   return s;
 }
 
 Status TestWritableFile::Close() {
   writable_file_opened_ = false;
   Status s = target_->Close();
   if (s.ok()) {
     env_->WritableFileClosed(state_);
   }
   return s;
 }
 
 Status TestWritableFile::Flush() {
   Status s = target_->Flush();
   if (s.ok() && env_->IsFilesystemActive()) {
     state_.pos_at_last_flush_ = state_.pos_;
   }
   return s;
 }
 
 Status TestWritableFile::SyncParent() {
   Status s = SyncDir(GetDirName(state_.filename_));
   if (s.ok()) {
     env_->DirWasSynced();
   }
   return s;
 }
 
 Status TestWritableFile::Sync() {
   if (!env_->IsFilesystemActive()) {
     return Status::OK();
   }
   // Ensure new files referred to by the manifest are in the filesystem.
   Status s = target_->Sync();
   if (s.ok()) {
     state_.pos_at_last_sync_ = state_.pos_;
   }
   if (env_->IsFileCreatedSinceLastDirSync(state_.filename_)) {
     Status ps = SyncParent();
     if (s.ok() && !ps.ok()) {
       s = ps;
     }
   }
   return s;
 }
 
+std::string TestWritableFile::GetName() const /*override*/ {
+    return "[testwritablefile]"; 
+}
+
+
 Status FaultInjectionTestEnv::NewWritableFile(const std::string& fname,
                                               WritableFile** result) {
   WritableFile* actual_writable_file;
   Status s = target()->NewWritableFile(fname, &actual_writable_file);
   if (s.ok()) {
     FileState state(fname);
     state.pos_ = 0;
     *result = new TestWritableFile(state, actual_writable_file, this);
     // NewWritableFile doesn't append to files, so if the same file is
     // opened again then it will be truncated - so forget our saved
     // state.
     UntrackFile(fname);
     MutexLock l(&mutex_);
     new_files_since_last_dir_sync_.insert(fname);
   }
   return s;
 }
 
 Status FaultInjectionTestEnv::NewAppendableFile(const std::string& fname,
                                                 WritableFile** result) {
   WritableFile* actual_writable_file;
   Status s = target()->NewAppendableFile(fname, &actual_writable_file);
   if (s.ok()) {
     FileState state(fname);
     state.pos_ = 0;
     {
       MutexLock l(&mutex_);
       if (db_file_state_.count(fname) == 0) {
         new_files_since_last_dir_sync_.insert(fname);
       } else {
         state = db_file_state_[fname];
       }
     }
     *result = new TestWritableFile(state, actual_writable_file, this);
   }
   return s;
 }
 
 Status FaultInjectionTestEnv::DropUnsyncedFileData() {
   Status s;
   MutexLock l(&mutex_);
   for (std::map<std::string, FileState>::const_iterator it =
            db_file_state_.begin();
        s.ok() && it != db_file_state_.end(); ++it) {
     const FileState& state = it->second;
     if (!state.IsFullySynced()) {
       s = state.DropUnsyncedData();
     }
   }
   return s;
 }
 
 void FaultInjectionTestEnv::DirWasSynced() {
   MutexLock l(&mutex_);
   new_files_since_last_dir_sync_.clear();
 }
 
 bool FaultInjectionTestEnv::IsFileCreatedSinceLastDirSync(
     const std::string& filename) {
   MutexLock l(&mutex_);
   return new_files_since_last_dir_sync_.find(filename) !=
          new_files_since_last_dir_sync_.end();
 }
 
 void FaultInjectionTestEnv::UntrackFile(const std::string& f) {
   MutexLock l(&mutex_);
   db_file_state_.erase(f);
   new_files_since_last_dir_sync_.erase(f);
 }
 
 Status FaultInjectionTestEnv::DeleteFile(const std::string& f) {
   Status s = EnvWrapper::DeleteFile(f);
   ASSERT_OK(s);
   if (s.ok()) {
     UntrackFile(f);
   }
   return s;
 }
 
 Status FaultInjectionTestEnv::RenameFile(const std::string& s,
                                          const std::string& t) {
   Status ret = EnvWrapper::RenameFile(s, t);
 
   if (ret.ok()) {
     MutexLock l(&mutex_);
     if (db_file_state_.find(s) != db_file_state_.end()) {
       db_file_state_[t] = db_file_state_[s];
       db_file_state_.erase(s);
     }
 
     if (new_files_since_last_dir_sync_.erase(s) != 0) {
       assert(new_files_since_last_dir_sync_.find(t) ==
              new_files_since_last_dir_sync_.end());
       new_files_since_last_dir_sync_.insert(t);
     }
   }
 
   return ret;
 }
 
 void FaultInjectionTestEnv::ResetState() {
   // Since we are not destroying the database, the existing files
   // should keep their recorded synced/flushed state. Therefore
   // we do not reset db_file_state_ and new_files_since_last_dir_sync_.
   MutexLock l(&mutex_);
   SetFilesystemActive(true);
 }
 
 Status FaultInjectionTestEnv::DeleteFilesCreatedAfterLastDirSync() {
   // Because DeleteFile access this container make a copy to avoid deadlock
   mutex_.Lock();
   std::set<std::string> new_files(new_files_since_last_dir_sync_.begin(),
                                   new_files_since_last_dir_sync_.end());
   mutex_.Unlock();
   Status s;
   std::set<std::string>::const_iterator it;
   for (it = new_files.begin(); s.ok() && it != new_files.end(); ++it) {
     s = DeleteFile(*it);
   }
   return s;
 }
 
 void FaultInjectionTestEnv::WritableFileClosed(const FileState& state) {
   MutexLock l(&mutex_);
   db_file_state_[state.filename_] = state;
 }
 
 Status FileState::DropUnsyncedData() const {
   ssize_t sync_pos = pos_at_last_sync_ == -1 ? 0 : pos_at_last_sync_;
   return Truncate(filename_, sync_pos);
 }
 
 class FaultInjectionTest {
  public:
   enum ExpectedVerifResult { VAL_EXPECT_NO_ERROR, VAL_EXPECT_ERROR };
   enum ResetMethod { RESET_DROP_UNSYNCED_DATA, RESET_DELETE_UNSYNCED_FILES };
 
   FaultInjectionTestEnv* env_;
   std::string dbname_;
   Cache* tiny_cache_;
   Options options_;
   DB* db_;
 
   FaultInjectionTest()
       : env_(new FaultInjectionTestEnv),
         tiny_cache_(NewLRUCache(100)),
         db_(NULL) {
     dbname_ = test::TmpDir() + "/fault_test";
     DestroyDB(dbname_, Options());  // Destroy any db from earlier run
     options_.reuse_logs = true;
     options_.env = env_;
     options_.paranoid_checks = true;
     options_.block_cache = tiny_cache_;
     options_.create_if_missing = true;
   }
 
   ~FaultInjectionTest() {
     CloseDB();
     DestroyDB(dbname_, Options());
     delete tiny_cache_;
     delete env_;
   }
 
   void ReuseLogs(bool reuse) {
     options_.reuse_logs = reuse;
   }
 
   void Build(int start_idx, int num_vals) {
     std::string key_space, value_space;
     WriteBatch batch;
     for (int i = start_idx; i < start_idx + num_vals; i++) {
       Slice key = Key(i, &key_space);
       batch.Clear();
       batch.Put(key, Value(i, &value_space));
       WriteOptions options;
       ASSERT_OK(db_->Write(options, &batch));
     }
   }
 
   Status ReadValue(int i, std::string* val) const {
     std::string key_space, value_space;
     Slice key = Key(i, &key_space);
     Value(i, &value_space);
     ReadOptions options;
     return db_->Get(options, key, val);
   }
 
   Status Verify(int start_idx, int num_vals,
                 ExpectedVerifResult expected) const {
     std::string val;
     std::string value_space;
     Status s;
     for (int i = start_idx; i < start_idx + num_vals && s.ok(); i++) {
       Value(i, &value_space);
       s = ReadValue(i, &val);
       if (expected == VAL_EXPECT_NO_ERROR) {
         if (s.ok()) {
           ASSERT_EQ(value_space, val);
         }
       } else if (s.ok()) {
         fprintf(stderr, "Expected an error at %d, but was OK\n", i);
         s = Status::IOError(dbname_, "Expected value error:");
       } else {
         s = Status::OK();  // An expected error
       }
     }
     return s;
   }
 
   // Return the ith key
   Slice Key(int i, std::string* storage) const {
     char buf[100];
     snprintf(buf, sizeof(buf), "%016d", i);
     storage->assign(buf, strlen(buf));
     return Slice(*storage);
   }
 
   // Return the value to associate with the specified key
   Slice Value(int k, std::string* storage) const {
     Random r(k);
     return test::RandomString(&r, kValueSize, storage);
   }
 
   Status OpenDB() {
     delete db_;
     db_ = NULL;
     env_->ResetState();
     return DB::Open(options_, dbname_, &db_);
   }
 
   void CloseDB() {
     delete db_;
     db_ = NULL;
   }
 
   void DeleteAllData() {
     Iterator* iter = db_->NewIterator(ReadOptions());
     WriteOptions options;
     for (iter->SeekToFirst(); iter->Valid(); iter->Next()) {
       ASSERT_OK(db_->Delete(WriteOptions(), iter->key()));
     }
 
     delete iter;
   }
 
   void ResetDBState(ResetMethod reset_method) {
     switch (reset_method) {
       case RESET_DROP_UNSYNCED_DATA:
         ASSERT_OK(env_->DropUnsyncedFileData());
         break;
       case RESET_DELETE_UNSYNCED_FILES:
         ASSERT_OK(env_->DeleteFilesCreatedAfterLastDirSync());
         break;
       default:
         assert(false);
     }
   }
 
   void PartialCompactTestPreFault(int num_pre_sync, int num_post_sync) {
     DeleteAllData();
     Build(0, num_pre_sync);
     db_->CompactRange(NULL, NULL);
     Build(num_pre_sync, num_post_sync);
   }
 
   void PartialCompactTestReopenWithFault(ResetMethod reset_method,
                                          int num_pre_sync,
                                          int num_post_sync) {
     env_->SetFilesystemActive(false);
     CloseDB();
     ResetDBState(reset_method);
     ASSERT_OK(OpenDB());
     ASSERT_OK(Verify(0, num_pre_sync, FaultInjectionTest::VAL_EXPECT_NO_ERROR));
     ASSERT_OK(Verify(num_pre_sync, num_post_sync, FaultInjectionTest::VAL_EXPECT_ERROR));
   }
 
   void NoWriteTestPreFault() {
   }
 
   void NoWriteTestReopenWithFault(ResetMethod reset_method) {
     CloseDB();
     ResetDBState(reset_method);
     ASSERT_OK(OpenDB());
   }
 
   void DoTest() {
     Random rnd(0);
     ASSERT_OK(OpenDB());
     for (size_t idx = 0; idx < kNumIterations; idx++) {
       int num_pre_sync = rnd.Uniform(kMaxNumValues);
       int num_post_sync = rnd.Uniform(kMaxNumValues);
 
       PartialCompactTestPreFault(num_pre_sync, num_post_sync);
       PartialCompactTestReopenWithFault(RESET_DROP_UNSYNCED_DATA,
                                         num_pre_sync,
                                         num_post_sync);
 
       NoWriteTestPreFault();
       NoWriteTestReopenWithFault(RESET_DROP_UNSYNCED_DATA);
 
       PartialCompactTestPreFault(num_pre_sync, num_post_sync);
       // No new files created so we expect all values since no files will be
       // dropped.
       PartialCompactTestReopenWithFault(RESET_DELETE_UNSYNCED_FILES,
                                         num_pre_sync + num_post_sync,
                                         0);
 
       NoWriteTestPreFault();
       NoWriteTestReopenWithFault(RESET_DELETE_UNSYNCED_FILES);
     }
   }
 };
 
 TEST(FaultInjectionTest, FaultTestNoLogReuse) {
   ReuseLogs(false);
   DoTest();
 }
 
 TEST(FaultInjectionTest, FaultTestWithLogReuse) {
   ReuseLogs(true);
   DoTest();
 }
 
 }  // namespace leveldb
 
 int main(int argc, char** argv) {
   return leveldb::test::RunAllTests();
 }
diff --git a/src/leveldb/db/log_test.cc b/src/leveldb/db/log_test.cc
index 48a592865..c750284a0 100644
--- a/src/leveldb/db/log_test.cc
+++ b/src/leveldb/db/log_test.cc
@@ -1,591 +1,594 @@
 // Copyright (c) 2011 The LevelDB Authors. All rights reserved.
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file. See the AUTHORS file for names of contributors.
 
 #include "db/log_reader.h"
 #include "db/log_writer.h"
 #include "leveldb/env.h"
 #include "util/coding.h"
 #include "util/crc32c.h"
 #include "util/random.h"
 #include "util/testharness.h"
 
 namespace leveldb {
 namespace log {
 
 // Construct a string of the specified length made out of the supplied
 // partial string.
 static std::string BigString(const std::string& partial_string, size_t n) {
   std::string result;
   while (result.size() < n) {
     result.append(partial_string);
   }
   result.resize(n);
   return result;
 }
 
 // Construct a string from a number
 static std::string NumberString(int n) {
   char buf[50];
   snprintf(buf, sizeof(buf), "%d.", n);
   return std::string(buf);
 }
 
 // Return a skewed potentially long string
 static std::string RandomSkewedString(int i, Random* rnd) {
   return BigString(NumberString(i), rnd->Skewed(17));
 }
 
 class LogTest {
  private:
   class StringDest : public WritableFile {
    public:
     std::string contents_;
 
     virtual Status Close() { return Status::OK(); }
     virtual Status Flush() { return Status::OK(); }
     virtual Status Sync() { return Status::OK(); }
     virtual Status Append(const Slice& slice) {
       contents_.append(slice.data(), slice.size());
       return Status::OK();
     }
+    std::string GetName() const override { return "[stringdest]"; }
   };
 
   class StringSource : public SequentialFile {
    public:
     Slice contents_;
     bool force_error_;
     bool returned_partial_;
     StringSource() : force_error_(false), returned_partial_(false) { }
 
     virtual Status Read(size_t n, Slice* result, char* scratch) {
       ASSERT_TRUE(!returned_partial_) << "must not Read() after eof/error";
 
       if (force_error_) {
         force_error_ = false;
         returned_partial_ = true;
         return Status::Corruption("read error");
       }
 
       if (contents_.size() < n) {
         n = contents_.size();
         returned_partial_ = true;
       }
       *result = Slice(contents_.data(), n);
       contents_.remove_prefix(n);
       return Status::OK();
     }
 
     virtual Status Skip(uint64_t n) {
       if (n > contents_.size()) {
         contents_.clear();
         return Status::NotFound("in-memory file skipped past end");
       }
 
       contents_.remove_prefix(n);
 
       return Status::OK();
     }
+
+    std::string GetName() const override { return "[stringsource]"; }
   };
 
   class ReportCollector : public Reader::Reporter {
    public:
     size_t dropped_bytes_;
     std::string message_;
 
     ReportCollector() : dropped_bytes_(0) { }
     virtual void Corruption(size_t bytes, const Status& status) {
       dropped_bytes_ += bytes;
       message_.append(status.ToString());
     }
   };
 
   StringDest dest_;
   StringSource source_;
   ReportCollector report_;
   bool reading_;
   Writer* writer_;
   Reader* reader_;
 
   // Record metadata for testing initial offset functionality
   static size_t initial_offset_record_sizes_[];
   static uint64_t initial_offset_last_record_offsets_[];
   static int num_initial_offset_records_;
 
  public:
   LogTest() : reading_(false),
               writer_(new Writer(&dest_)),
               reader_(new Reader(&source_, &report_, true/*checksum*/,
                       0/*initial_offset*/)) {
   }
 
   ~LogTest() {
     delete writer_;
     delete reader_;
   }
 
   void ReopenForAppend() {
     delete writer_;
     writer_ = new Writer(&dest_, dest_.contents_.size());
   }
 
   void Write(const std::string& msg) {
     ASSERT_TRUE(!reading_) << "Write() after starting to read";
     writer_->AddRecord(Slice(msg));
   }
 
   size_t WrittenBytes() const {
     return dest_.contents_.size();
   }
 
   std::string Read() {
     if (!reading_) {
       reading_ = true;
       source_.contents_ = Slice(dest_.contents_);
     }
     std::string scratch;
     Slice record;
     if (reader_->ReadRecord(&record, &scratch)) {
       return record.ToString();
     } else {
       return "EOF";
     }
   }
 
   void IncrementByte(int offset, int delta) {
     dest_.contents_[offset] += delta;
   }
 
   void SetByte(int offset, char new_byte) {
     dest_.contents_[offset] = new_byte;
   }
 
   void ShrinkSize(int bytes) {
     dest_.contents_.resize(dest_.contents_.size() - bytes);
   }
 
   void FixChecksum(int header_offset, int len) {
     // Compute crc of type/len/data
     uint32_t crc = crc32c::Value(&dest_.contents_[header_offset+6], 1 + len);
     crc = crc32c::Mask(crc);
     EncodeFixed32(&dest_.contents_[header_offset], crc);
   }
 
   void ForceError() {
     source_.force_error_ = true;
   }
 
   size_t DroppedBytes() const {
     return report_.dropped_bytes_;
   }
 
   std::string ReportMessage() const {
     return report_.message_;
   }
 
   // Returns OK iff recorded error message contains "msg"
   std::string MatchError(const std::string& msg) const {
     if (report_.message_.find(msg) == std::string::npos) {
       return report_.message_;
     } else {
       return "OK";
     }
   }
 
   void WriteInitialOffsetLog() {
     for (int i = 0; i < num_initial_offset_records_; i++) {
       std::string record(initial_offset_record_sizes_[i],
                          static_cast<char>('a' + i));
       Write(record);
     }
   }
 
   void StartReadingAt(uint64_t initial_offset) {
     delete reader_;
     reader_ = new Reader(&source_, &report_, true/*checksum*/, initial_offset);
   }
 
   void CheckOffsetPastEndReturnsNoRecords(uint64_t offset_past_end) {
     WriteInitialOffsetLog();
     reading_ = true;
     source_.contents_ = Slice(dest_.contents_);
     Reader* offset_reader = new Reader(&source_, &report_, true/*checksum*/,
                                        WrittenBytes() + offset_past_end);
     Slice record;
     std::string scratch;
     ASSERT_TRUE(!offset_reader->ReadRecord(&record, &scratch));
     delete offset_reader;
   }
 
   void CheckInitialOffsetRecord(uint64_t initial_offset,
                                 int expected_record_offset) {
     WriteInitialOffsetLog();
     reading_ = true;
     source_.contents_ = Slice(dest_.contents_);
     Reader* offset_reader = new Reader(&source_, &report_, true/*checksum*/,
                                        initial_offset);
 
     // Read all records from expected_record_offset through the last one.
     ASSERT_LT(expected_record_offset, num_initial_offset_records_);
     for (; expected_record_offset < num_initial_offset_records_;
          ++expected_record_offset) {
       Slice record;
       std::string scratch;
       ASSERT_TRUE(offset_reader->ReadRecord(&record, &scratch));
       ASSERT_EQ(initial_offset_record_sizes_[expected_record_offset],
                 record.size());
       ASSERT_EQ(initial_offset_last_record_offsets_[expected_record_offset],
                 offset_reader->LastRecordOffset());
       ASSERT_EQ((char)('a' + expected_record_offset), record.data()[0]);
     }
     delete offset_reader;
   }
 };
 
 size_t LogTest::initial_offset_record_sizes_[] =
     {10000,  // Two sizable records in first block
      10000,
      2 * log::kBlockSize - 1000,  // Span three blocks
      1,
      13716,  // Consume all but two bytes of block 3.
      log::kBlockSize - kHeaderSize, // Consume the entirety of block 4.
     };
 
 uint64_t LogTest::initial_offset_last_record_offsets_[] =
     {0,
      kHeaderSize + 10000,
      2 * (kHeaderSize + 10000),
      2 * (kHeaderSize + 10000) +
          (2 * log::kBlockSize - 1000) + 3 * kHeaderSize,
      2 * (kHeaderSize + 10000) +
          (2 * log::kBlockSize - 1000) + 3 * kHeaderSize
          + kHeaderSize + 1,
      3 * log::kBlockSize,
     };
 
 // LogTest::initial_offset_last_record_offsets_ must be defined before this.
 int LogTest::num_initial_offset_records_ =
     sizeof(LogTest::initial_offset_last_record_offsets_)/sizeof(uint64_t);
 
 TEST(LogTest, Empty) {
   ASSERT_EQ("EOF", Read());
 }
 
 TEST(LogTest, ReadWrite) {
   Write("foo");
   Write("bar");
   Write("");
   Write("xxxx");
   ASSERT_EQ("foo", Read());
   ASSERT_EQ("bar", Read());
   ASSERT_EQ("", Read());
   ASSERT_EQ("xxxx", Read());
   ASSERT_EQ("EOF", Read());
   ASSERT_EQ("EOF", Read());  // Make sure reads at eof work
 }
 
 TEST(LogTest, ManyBlocks) {
   for (int i = 0; i < 100000; i++) {
     Write(NumberString(i));
   }
   for (int i = 0; i < 100000; i++) {
     ASSERT_EQ(NumberString(i), Read());
   }
   ASSERT_EQ("EOF", Read());
 }
 
 TEST(LogTest, Fragmentation) {
   Write("small");
   Write(BigString("medium", 50000));
   Write(BigString("large", 100000));
   ASSERT_EQ("small", Read());
   ASSERT_EQ(BigString("medium", 50000), Read());
   ASSERT_EQ(BigString("large", 100000), Read());
   ASSERT_EQ("EOF", Read());
 }
 
 TEST(LogTest, MarginalTrailer) {
   // Make a trailer that is exactly the same length as an empty record.
   const int n = kBlockSize - 2*kHeaderSize;
   Write(BigString("foo", n));
   ASSERT_EQ(kBlockSize - kHeaderSize, WrittenBytes());
   Write("");
   Write("bar");
   ASSERT_EQ(BigString("foo", n), Read());
   ASSERT_EQ("", Read());
   ASSERT_EQ("bar", Read());
   ASSERT_EQ("EOF", Read());
 }
 
 TEST(LogTest, MarginalTrailer2) {
   // Make a trailer that is exactly the same length as an empty record.
   const int n = kBlockSize - 2*kHeaderSize;
   Write(BigString("foo", n));
   ASSERT_EQ(kBlockSize - kHeaderSize, WrittenBytes());
   Write("bar");
   ASSERT_EQ(BigString("foo", n), Read());
   ASSERT_EQ("bar", Read());
   ASSERT_EQ("EOF", Read());
   ASSERT_EQ(0, DroppedBytes());
   ASSERT_EQ("", ReportMessage());
 }
 
 TEST(LogTest, ShortTrailer) {
   const int n = kBlockSize - 2*kHeaderSize + 4;
   Write(BigString("foo", n));
   ASSERT_EQ(kBlockSize - kHeaderSize + 4, WrittenBytes());
   Write("");
   Write("bar");
   ASSERT_EQ(BigString("foo", n), Read());
   ASSERT_EQ("", Read());
   ASSERT_EQ("bar", Read());
   ASSERT_EQ("EOF", Read());
 }
 
 TEST(LogTest, AlignedEof) {
   const int n = kBlockSize - 2*kHeaderSize + 4;
   Write(BigString("foo", n));
   ASSERT_EQ(kBlockSize - kHeaderSize + 4, WrittenBytes());
   ASSERT_EQ(BigString("foo", n), Read());
   ASSERT_EQ("EOF", Read());
 }
 
 TEST(LogTest, OpenForAppend) {
   Write("hello");
   ReopenForAppend();
   Write("world");
   ASSERT_EQ("hello", Read());
   ASSERT_EQ("world", Read());
   ASSERT_EQ("EOF", Read());
 }
 
 TEST(LogTest, RandomRead) {
   const int N = 500;
   Random write_rnd(301);
   for (int i = 0; i < N; i++) {
     Write(RandomSkewedString(i, &write_rnd));
   }
   Random read_rnd(301);
   for (int i = 0; i < N; i++) {
     ASSERT_EQ(RandomSkewedString(i, &read_rnd), Read());
   }
   ASSERT_EQ("EOF", Read());
 }
 
 // Tests of all the error paths in log_reader.cc follow:
 
 TEST(LogTest, ReadError) {
   Write("foo");
   ForceError();
   ASSERT_EQ("EOF", Read());
   ASSERT_EQ(kBlockSize, DroppedBytes());
   ASSERT_EQ("OK", MatchError("read error"));
 }
 
 TEST(LogTest, BadRecordType) {
   Write("foo");
   // Type is stored in header[6]
   IncrementByte(6, 100);
   FixChecksum(0, 3);
   ASSERT_EQ("EOF", Read());
   ASSERT_EQ(3, DroppedBytes());
   ASSERT_EQ("OK", MatchError("unknown record type"));
 }
 
 TEST(LogTest, TruncatedTrailingRecordIsIgnored) {
   Write("foo");
   ShrinkSize(4);   // Drop all payload as well as a header byte
   ASSERT_EQ("EOF", Read());
   // Truncated last record is ignored, not treated as an error.
   ASSERT_EQ(0, DroppedBytes());
   ASSERT_EQ("", ReportMessage());
 }
 
 TEST(LogTest, BadLength) {
   const int kPayloadSize = kBlockSize - kHeaderSize;
   Write(BigString("bar", kPayloadSize));
   Write("foo");
   // Least significant size byte is stored in header[4].
   IncrementByte(4, 1);
   ASSERT_EQ("foo", Read());
   ASSERT_EQ(kBlockSize, DroppedBytes());
   ASSERT_EQ("OK", MatchError("bad record length"));
 }
 
 TEST(LogTest, BadLengthAtEndIsIgnored) {
   Write("foo");
   ShrinkSize(1);
   ASSERT_EQ("EOF", Read());
   ASSERT_EQ(0, DroppedBytes());
   ASSERT_EQ("", ReportMessage());
 }
 
 TEST(LogTest, ChecksumMismatch) {
   Write("foo");
   IncrementByte(0, 10);
   ASSERT_EQ("EOF", Read());
   ASSERT_EQ(10, DroppedBytes());
   ASSERT_EQ("OK", MatchError("checksum mismatch"));
 }
 
 TEST(LogTest, UnexpectedMiddleType) {
   Write("foo");
   SetByte(6, kMiddleType);
   FixChecksum(0, 3);
   ASSERT_EQ("EOF", Read());
   ASSERT_EQ(3, DroppedBytes());
   ASSERT_EQ("OK", MatchError("missing start"));
 }
 
 TEST(LogTest, UnexpectedLastType) {
   Write("foo");
   SetByte(6, kLastType);
   FixChecksum(0, 3);
   ASSERT_EQ("EOF", Read());
   ASSERT_EQ(3, DroppedBytes());
   ASSERT_EQ("OK", MatchError("missing start"));
 }
 
 TEST(LogTest, UnexpectedFullType) {
   Write("foo");
   Write("bar");
   SetByte(6, kFirstType);
   FixChecksum(0, 3);
   ASSERT_EQ("bar", Read());
   ASSERT_EQ("EOF", Read());
   ASSERT_EQ(3, DroppedBytes());
   ASSERT_EQ("OK", MatchError("partial record without end"));
 }
 
 TEST(LogTest, UnexpectedFirstType) {
   Write("foo");
   Write(BigString("bar", 100000));
   SetByte(6, kFirstType);
   FixChecksum(0, 3);
   ASSERT_EQ(BigString("bar", 100000), Read());
   ASSERT_EQ("EOF", Read());
   ASSERT_EQ(3, DroppedBytes());
   ASSERT_EQ("OK", MatchError("partial record without end"));
 }
 
 TEST(LogTest, MissingLastIsIgnored) {
   Write(BigString("bar", kBlockSize));
   // Remove the LAST block, including header.
   ShrinkSize(14);
   ASSERT_EQ("EOF", Read());
   ASSERT_EQ("", ReportMessage());
   ASSERT_EQ(0, DroppedBytes());
 }
 
 TEST(LogTest, PartialLastIsIgnored) {
   Write(BigString("bar", kBlockSize));
   // Cause a bad record length in the LAST block.
   ShrinkSize(1);
   ASSERT_EQ("EOF", Read());
   ASSERT_EQ("", ReportMessage());
   ASSERT_EQ(0, DroppedBytes());
 }
 
 TEST(LogTest, SkipIntoMultiRecord) {
   // Consider a fragmented record:
   //    first(R1), middle(R1), last(R1), first(R2)
   // If initial_offset points to a record after first(R1) but before first(R2)
   // incomplete fragment errors are not actual errors, and must be suppressed
   // until a new first or full record is encountered.
   Write(BigString("foo", 3*kBlockSize));
   Write("correct");
   StartReadingAt(kBlockSize);
 
   ASSERT_EQ("correct", Read());
   ASSERT_EQ("", ReportMessage());
   ASSERT_EQ(0, DroppedBytes());
   ASSERT_EQ("EOF", Read());
 }
 
 TEST(LogTest, ErrorJoinsRecords) {
   // Consider two fragmented records:
   //    first(R1) last(R1) first(R2) last(R2)
   // where the middle two fragments disappear.  We do not want
   // first(R1),last(R2) to get joined and returned as a valid record.
 
   // Write records that span two blocks
   Write(BigString("foo", kBlockSize));
   Write(BigString("bar", kBlockSize));
   Write("correct");
 
   // Wipe the middle block
   for (int offset = kBlockSize; offset < 2*kBlockSize; offset++) {
     SetByte(offset, 'x');
   }
 
   ASSERT_EQ("correct", Read());
   ASSERT_EQ("EOF", Read());
   const size_t dropped = DroppedBytes();
   ASSERT_LE(dropped, 2*kBlockSize + 100);
   ASSERT_GE(dropped, 2*kBlockSize);
 }
 
 TEST(LogTest, ReadStart) {
   CheckInitialOffsetRecord(0, 0);
 }
 
 TEST(LogTest, ReadSecondOneOff) {
   CheckInitialOffsetRecord(1, 1);
 }
 
 TEST(LogTest, ReadSecondTenThousand) {
   CheckInitialOffsetRecord(10000, 1);
 }
 
 TEST(LogTest, ReadSecondStart) {
   CheckInitialOffsetRecord(10007, 1);
 }
 
 TEST(LogTest, ReadThirdOneOff) {
   CheckInitialOffsetRecord(10008, 2);
 }
 
 TEST(LogTest, ReadThirdStart) {
   CheckInitialOffsetRecord(20014, 2);
 }
 
 TEST(LogTest, ReadFourthOneOff) {
   CheckInitialOffsetRecord(20015, 3);
 }
 
 TEST(LogTest, ReadFourthFirstBlockTrailer) {
   CheckInitialOffsetRecord(log::kBlockSize - 4, 3);
 }
 
 TEST(LogTest, ReadFourthMiddleBlock) {
   CheckInitialOffsetRecord(log::kBlockSize + 1, 3);
 }
 
 TEST(LogTest, ReadFourthLastBlock) {
   CheckInitialOffsetRecord(2 * log::kBlockSize + 1, 3);
 }
 
 TEST(LogTest, ReadFourthStart) {
   CheckInitialOffsetRecord(
       2 * (kHeaderSize + 1000) + (2 * log::kBlockSize - 1000) + 3 * kHeaderSize,
       3);
 }
 
 TEST(LogTest, ReadInitialOffsetIntoBlockPadding) {
   CheckInitialOffsetRecord(3 * log::kBlockSize - 3, 5);
 }
 
 TEST(LogTest, ReadEnd) {
   CheckOffsetPastEndReturnsNoRecords(0);
 }
 
 TEST(LogTest, ReadPastEnd) {
   CheckOffsetPastEndReturnsNoRecords(5);
 }
 
 }  // namespace log
 }  // namespace leveldb
 
 int main(int argc, char** argv) {
   return leveldb::test::RunAllTests();
 }
diff --git a/src/leveldb/table/table_test.cc b/src/leveldb/table/table_test.cc
index abf6e246f..d1ab277e1 100644
--- a/src/leveldb/table/table_test.cc
+++ b/src/leveldb/table/table_test.cc
@@ -1,876 +1,881 @@
 // Copyright (c) 2011 The LevelDB Authors. All rights reserved.
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file. See the AUTHORS file for names of contributors.
 
 #include "leveldb/table.h"
 
 #include <map>
 #include <string>
 #include "db/dbformat.h"
 #include "db/memtable.h"
 #include "db/write_batch_internal.h"
 #include "leveldb/db.h"
 #include "leveldb/env.h"
 #include "leveldb/iterator.h"
 #include "leveldb/table_builder.h"
 #include "table/block.h"
 #include "table/block_builder.h"
 #include "table/format.h"
 #include "util/random.h"
 #include "util/testharness.h"
 #include "util/testutil.h"
 
 namespace leveldb {
 
 // Return reverse of "key".
 // Used to test non-lexicographic comparators.
 static std::string Reverse(const Slice& key) {
   std::string str(key.ToString());
   std::string rev("");
   for (std::string::reverse_iterator rit = str.rbegin();
        rit != str.rend(); ++rit) {
     rev.push_back(*rit);
   }
   return rev;
 }
 
 namespace {
 class ReverseKeyComparator : public Comparator {
  public:
   virtual const char* Name() const {
     return "leveldb.ReverseBytewiseComparator";
   }
 
   virtual int Compare(const Slice& a, const Slice& b) const {
     return BytewiseComparator()->Compare(Reverse(a), Reverse(b));
   }
 
   virtual void FindShortestSeparator(
       std::string* start,
       const Slice& limit) const {
     std::string s = Reverse(*start);
     std::string l = Reverse(limit);
     BytewiseComparator()->FindShortestSeparator(&s, l);
     *start = Reverse(s);
   }
 
   virtual void FindShortSuccessor(std::string* key) const {
     std::string s = Reverse(*key);
     BytewiseComparator()->FindShortSuccessor(&s);
     *key = Reverse(s);
   }
 };
 }  // namespace
 static ReverseKeyComparator reverse_key_comparator;
 
 static void Increment(const Comparator* cmp, std::string* key) {
   if (cmp == BytewiseComparator()) {
     key->push_back('\0');
   } else {
     assert(cmp == &reverse_key_comparator);
     std::string rev = Reverse(*key);
     rev.push_back('\0');
     *key = Reverse(rev);
   }
 }
 
 // An STL comparator that uses a Comparator
 namespace {
 struct STLLessThan {
   const Comparator* cmp;
 
   STLLessThan() : cmp(BytewiseComparator()) { }
   STLLessThan(const Comparator* c) : cmp(c) { }
   bool operator()(const std::string& a, const std::string& b) const {
     return cmp->Compare(Slice(a), Slice(b)) < 0;
   }
 };
 }  // namespace
 
 class StringSink: public WritableFile {
  public:
   ~StringSink() { }
 
   const std::string& contents() const { return contents_; }
 
   virtual Status Close() { return Status::OK(); }
   virtual Status Flush() { return Status::OK(); }
   virtual Status Sync() { return Status::OK(); }
 
   virtual Status Append(const Slice& data) {
     contents_.append(data.data(), data.size());
     return Status::OK();
   }
 
+  std::string GetName() const override { return "[stringsink]"; }
+
+
  private:
   std::string contents_;
 };
 
 
 class StringSource: public RandomAccessFile {
  public:
   StringSource(const Slice& contents)
       : contents_(contents.data(), contents.size()) {
   }
 
   virtual ~StringSource() { }
 
   uint64_t Size() const { return contents_.size(); }
 
   virtual Status Read(uint64_t offset, size_t n, Slice* result,
                        char* scratch) const {
     if (offset > contents_.size()) {
       return Status::InvalidArgument("invalid Read offset");
     }
     if (offset + n > contents_.size()) {
       n = contents_.size() - offset;
     }
     memcpy(scratch, &contents_[offset], n);
     *result = Slice(scratch, n);
     return Status::OK();
   }
 
+  std::string GetName() const override { return "[stringsource]"; }
+
  private:
   std::string contents_;
 };
 
 typedef std::map<std::string, std::string, STLLessThan> KVMap;
 
 // Helper class for tests to unify the interface between
 // BlockBuilder/TableBuilder and Block/Table.
 class Constructor {
  public:
   explicit Constructor(const Comparator* cmp) : data_(STLLessThan(cmp)) { }
   virtual ~Constructor() { }
 
   void Add(const std::string& key, const Slice& value) {
     data_[key] = value.ToString();
   }
 
   // Finish constructing the data structure with all the keys that have
   // been added so far.  Returns the keys in sorted order in "*keys"
   // and stores the key/value pairs in "*kvmap"
   void Finish(const Options& options,
               std::vector<std::string>* keys,
               KVMap* kvmap) {
     *kvmap = data_;
     keys->clear();
     for (KVMap::const_iterator it = data_.begin();
          it != data_.end();
          ++it) {
       keys->push_back(it->first);
     }
     data_.clear();
     Status s = FinishImpl(options, *kvmap);
     ASSERT_TRUE(s.ok()) << s.ToString();
   }
 
   // Construct the data structure from the data in "data"
   virtual Status FinishImpl(const Options& options, const KVMap& data) = 0;
 
   virtual Iterator* NewIterator() const = 0;
 
   virtual const KVMap& data() { return data_; }
 
   virtual DB* db() const { return NULL; }  // Overridden in DBConstructor
 
  private:
   KVMap data_;
 };
 
 class BlockConstructor: public Constructor {
  public:
   explicit BlockConstructor(const Comparator* cmp)
       : Constructor(cmp),
         comparator_(cmp),
         block_(NULL) { }
   ~BlockConstructor() {
     delete block_;
   }
   virtual Status FinishImpl(const Options& options, const KVMap& data) {
     delete block_;
     block_ = NULL;
     BlockBuilder builder(&options);
 
     for (KVMap::const_iterator it = data.begin();
          it != data.end();
          ++it) {
       builder.Add(it->first, it->second);
     }
     // Open the block
     data_ = builder.Finish().ToString();
     BlockContents contents;
     contents.data = data_;
     contents.cachable = false;
     contents.heap_allocated = false;
     block_ = new Block(contents);
     return Status::OK();
   }
   virtual Iterator* NewIterator() const {
     return block_->NewIterator(comparator_);
   }
 
  private:
   const Comparator* comparator_;
   std::string data_;
   Block* block_;
 
   BlockConstructor();
 };
 
 class TableConstructor: public Constructor {
  public:
   TableConstructor(const Comparator* cmp)
       : Constructor(cmp),
         source_(NULL), table_(NULL) {
   }
   ~TableConstructor() {
     Reset();
   }
   virtual Status FinishImpl(const Options& options, const KVMap& data) {
     Reset();
     StringSink sink;
     TableBuilder builder(options, &sink);
 
     for (KVMap::const_iterator it = data.begin();
          it != data.end();
          ++it) {
       builder.Add(it->first, it->second);
       ASSERT_TRUE(builder.status().ok());
     }
     Status s = builder.Finish();
     ASSERT_TRUE(s.ok()) << s.ToString();
 
     ASSERT_EQ(sink.contents().size(), builder.FileSize());
 
     // Open the table
     source_ = new StringSource(sink.contents());
     Options table_options;
     table_options.comparator = options.comparator;
     return Table::Open(table_options, source_, sink.contents().size(), &table_);
   }
 
   virtual Iterator* NewIterator() const {
     return table_->NewIterator(ReadOptions());
   }
 
   uint64_t ApproximateOffsetOf(const Slice& key) const {
     return table_->ApproximateOffsetOf(key);
   }
 
  private:
   void Reset() {
     delete table_;
     delete source_;
     table_ = NULL;
     source_ = NULL;
   }
 
   StringSource* source_;
   Table* table_;
 
   TableConstructor();
 };
 
 // A helper class that converts internal format keys into user keys
 class KeyConvertingIterator: public Iterator {
  public:
   explicit KeyConvertingIterator(Iterator* iter) : iter_(iter) { }
   virtual ~KeyConvertingIterator() { delete iter_; }
   virtual bool Valid() const { return iter_->Valid(); }
   virtual void Seek(const Slice& target) {
     ParsedInternalKey ikey(target, kMaxSequenceNumber, kTypeValue);
     std::string encoded;
     AppendInternalKey(&encoded, ikey);
     iter_->Seek(encoded);
   }
   virtual void SeekToFirst() { iter_->SeekToFirst(); }
   virtual void SeekToLast() { iter_->SeekToLast(); }
   virtual void Next() { iter_->Next(); }
   virtual void Prev() { iter_->Prev(); }
 
   virtual Slice key() const {
     assert(Valid());
     ParsedInternalKey key;
     if (!ParseInternalKey(iter_->key(), &key)) {
       status_ = Status::Corruption("malformed internal key");
       return Slice("corrupted key");
     }
     return key.user_key;
   }
 
   virtual Slice value() const { return iter_->value(); }
   virtual Status status() const {
     return status_.ok() ? iter_->status() : status_;
   }
 
  private:
   mutable Status status_;
   Iterator* iter_;
 
   // No copying allowed
   KeyConvertingIterator(const KeyConvertingIterator&);
   void operator=(const KeyConvertingIterator&);
 };
 
 class MemTableConstructor: public Constructor {
  public:
   explicit MemTableConstructor(const Comparator* cmp)
       : Constructor(cmp),
         internal_comparator_(cmp) {
     memtable_ = new MemTable(internal_comparator_);
     memtable_->Ref();
   }
   ~MemTableConstructor() {
     memtable_->Unref();
   }
   virtual Status FinishImpl(const Options& options, const KVMap& data) {
     memtable_->Unref();
     memtable_ = new MemTable(internal_comparator_);
     memtable_->Ref();
     int seq = 1;
     for (KVMap::const_iterator it = data.begin();
          it != data.end();
          ++it) {
       memtable_->Add(seq, kTypeValue, it->first, it->second);
       seq++;
     }
     return Status::OK();
   }
   virtual Iterator* NewIterator() const {
     return new KeyConvertingIterator(memtable_->NewIterator());
   }
 
  private:
   InternalKeyComparator internal_comparator_;
   MemTable* memtable_;
 };
 
 class DBConstructor: public Constructor {
  public:
   explicit DBConstructor(const Comparator* cmp)
       : Constructor(cmp),
         comparator_(cmp) {
     db_ = NULL;
     NewDB();
   }
   ~DBConstructor() {
     delete db_;
   }
   virtual Status FinishImpl(const Options& options, const KVMap& data) {
     delete db_;
     db_ = NULL;
     NewDB();
     for (KVMap::const_iterator it = data.begin();
          it != data.end();
          ++it) {
       WriteBatch batch;
       batch.Put(it->first, it->second);
       ASSERT_TRUE(db_->Write(WriteOptions(), &batch).ok());
     }
     return Status::OK();
   }
   virtual Iterator* NewIterator() const {
     return db_->NewIterator(ReadOptions());
   }
 
   virtual DB* db() const { return db_; }
 
  private:
   void NewDB() {
     std::string name = test::TmpDir() + "/table_testdb";
 
     Options options;
     options.comparator = comparator_;
     Status status = DestroyDB(name, options);
     ASSERT_TRUE(status.ok()) << status.ToString();
 
     options.create_if_missing = true;
     options.error_if_exists = true;
     options.write_buffer_size = 10000;  // Something small to force merging
     status = DB::Open(options, name, &db_);
     ASSERT_TRUE(status.ok()) << status.ToString();
   }
 
   const Comparator* comparator_;
   DB* db_;
 };
 
 enum TestType {
   TABLE_TEST,
   BLOCK_TEST,
   MEMTABLE_TEST,
   DB_TEST
 };
 
 struct TestArgs {
   TestType type;
   bool reverse_compare;
   int restart_interval;
 };
 
 static const TestArgs kTestArgList[] = {
   { TABLE_TEST, false, 16 },
   { TABLE_TEST, false, 1 },
   { TABLE_TEST, false, 1024 },
   { TABLE_TEST, true, 16 },
   { TABLE_TEST, true, 1 },
   { TABLE_TEST, true, 1024 },
 
   { BLOCK_TEST, false, 16 },
   { BLOCK_TEST, false, 1 },
   { BLOCK_TEST, false, 1024 },
   { BLOCK_TEST, true, 16 },
   { BLOCK_TEST, true, 1 },
   { BLOCK_TEST, true, 1024 },
 
   // Restart interval does not matter for memtables
   { MEMTABLE_TEST, false, 16 },
   { MEMTABLE_TEST, true, 16 },
 
   // Do not bother with restart interval variations for DB
   { DB_TEST, false, 16 },
   { DB_TEST, true, 16 },
 };
 static const int kNumTestArgs = sizeof(kTestArgList) / sizeof(kTestArgList[0]);
 
 class Harness {
  public:
   Harness() : constructor_(NULL) { }
 
   void Init(const TestArgs& args) {
     delete constructor_;
     constructor_ = NULL;
     options_ = Options();
 
     options_.block_restart_interval = args.restart_interval;
     // Use shorter block size for tests to exercise block boundary
     // conditions more.
     options_.block_size = 256;
     if (args.reverse_compare) {
       options_.comparator = &reverse_key_comparator;
     }
     switch (args.type) {
       case TABLE_TEST:
         constructor_ = new TableConstructor(options_.comparator);
         break;
       case BLOCK_TEST:
         constructor_ = new BlockConstructor(options_.comparator);
         break;
       case MEMTABLE_TEST:
         constructor_ = new MemTableConstructor(options_.comparator);
         break;
       case DB_TEST:
         constructor_ = new DBConstructor(options_.comparator);
         break;
     }
   }
 
   ~Harness() {
     delete constructor_;
   }
 
   void Add(const std::string& key, const std::string& value) {
     constructor_->Add(key, value);
   }
 
   void Test(Random* rnd) {
     std::vector<std::string> keys;
     KVMap data;
     constructor_->Finish(options_, &keys, &data);
 
     TestForwardScan(keys, data);
     TestBackwardScan(keys, data);
     TestRandomAccess(rnd, keys, data);
   }
 
   void TestForwardScan(const std::vector<std::string>& keys,
                        const KVMap& data) {
     Iterator* iter = constructor_->NewIterator();
     ASSERT_TRUE(!iter->Valid());
     iter->SeekToFirst();
     for (KVMap::const_iterator model_iter = data.begin();
          model_iter != data.end();
          ++model_iter) {
       ASSERT_EQ(ToString(data, model_iter), ToString(iter));
       iter->Next();
     }
     ASSERT_TRUE(!iter->Valid());
     delete iter;
   }
 
   void TestBackwardScan(const std::vector<std::string>& keys,
                         const KVMap& data) {
     Iterator* iter = constructor_->NewIterator();
     ASSERT_TRUE(!iter->Valid());
     iter->SeekToLast();
     for (KVMap::const_reverse_iterator model_iter = data.rbegin();
          model_iter != data.rend();
          ++model_iter) {
       ASSERT_EQ(ToString(data, model_iter), ToString(iter));
       iter->Prev();
     }
     ASSERT_TRUE(!iter->Valid());
     delete iter;
   }
 
   void TestRandomAccess(Random* rnd,
                         const std::vector<std::string>& keys,
                         const KVMap& data) {
     static const bool kVerbose = false;
     Iterator* iter = constructor_->NewIterator();
     ASSERT_TRUE(!iter->Valid());
     KVMap::const_iterator model_iter = data.begin();
     if (kVerbose) fprintf(stderr, "---\n");
     for (int i = 0; i < 200; i++) {
       const int toss = rnd->Uniform(5);
       switch (toss) {
         case 0: {
           if (iter->Valid()) {
             if (kVerbose) fprintf(stderr, "Next\n");
             iter->Next();
             ++model_iter;
             ASSERT_EQ(ToString(data, model_iter), ToString(iter));
           }
           break;
         }
 
         case 1: {
           if (kVerbose) fprintf(stderr, "SeekToFirst\n");
           iter->SeekToFirst();
           model_iter = data.begin();
           ASSERT_EQ(ToString(data, model_iter), ToString(iter));
           break;
         }
 
         case 2: {
           std::string key = PickRandomKey(rnd, keys);
           model_iter = data.lower_bound(key);
           if (kVerbose) fprintf(stderr, "Seek '%s'\n",
                                 EscapeString(key).c_str());
           iter->Seek(Slice(key));
           ASSERT_EQ(ToString(data, model_iter), ToString(iter));
           break;
         }
 
         case 3: {
           if (iter->Valid()) {
             if (kVerbose) fprintf(stderr, "Prev\n");
             iter->Prev();
             if (model_iter == data.begin()) {
               model_iter = data.end();   // Wrap around to invalid value
             } else {
               --model_iter;
             }
             ASSERT_EQ(ToString(data, model_iter), ToString(iter));
           }
           break;
         }
 
         case 4: {
           if (kVerbose) fprintf(stderr, "SeekToLast\n");
           iter->SeekToLast();
           if (keys.empty()) {
             model_iter = data.end();
           } else {
             std::string last = data.rbegin()->first;
             model_iter = data.lower_bound(last);
           }
           ASSERT_EQ(ToString(data, model_iter), ToString(iter));
           break;
         }
       }
     }
     delete iter;
   }
 
   std::string ToString(const KVMap& data, const KVMap::const_iterator& it) {
     if (it == data.end()) {
       return "END";
     } else {
       return "'" + it->first + "->" + it->second + "'";
     }
   }
 
   std::string ToString(const KVMap& data,
                        const KVMap::const_reverse_iterator& it) {
     if (it == data.rend()) {
       return "END";
     } else {
       return "'" + it->first + "->" + it->second + "'";
     }
   }
 
   std::string ToString(const Iterator* it) {
     if (!it->Valid()) {
       return "END";
     } else {
       return "'" + it->key().ToString() + "->" + it->value().ToString() + "'";
     }
   }
 
   std::string PickRandomKey(Random* rnd, const std::vector<std::string>& keys) {
     if (keys.empty()) {
       return "foo";
     } else {
       const int index = rnd->Uniform(keys.size());
       std::string result = keys[index];
       switch (rnd->Uniform(3)) {
         case 0:
           // Return an existing key
           break;
         case 1: {
           // Attempt to return something smaller than an existing key
           if (result.size() > 0 && result[result.size()-1] > '\0') {
             result[result.size()-1]--;
           }
           break;
         }
         case 2: {
           // Return something larger than an existing key
           Increment(options_.comparator, &result);
           break;
         }
       }
       return result;
     }
   }
 
   // Returns NULL if not running against a DB
   DB* db() const { return constructor_->db(); }
 
  private:
   Options options_;
   Constructor* constructor_;
 };
 
 // Test empty table/block.
 TEST(Harness, Empty) {
   for (int i = 0; i < kNumTestArgs; i++) {
     Init(kTestArgList[i]);
     Random rnd(test::RandomSeed() + 1);
     Test(&rnd);
   }
 }
 
 // Special test for a block with no restart entries.  The C++ leveldb
 // code never generates such blocks, but the Java version of leveldb
 // seems to.
 TEST(Harness, ZeroRestartPointsInBlock) {
   char data[sizeof(uint32_t)];
   memset(data, 0, sizeof(data));
   BlockContents contents;
   contents.data = Slice(data, sizeof(data));
   contents.cachable = false;
   contents.heap_allocated = false;
   Block block(contents);
   Iterator* iter = block.NewIterator(BytewiseComparator());
   iter->SeekToFirst();
   ASSERT_TRUE(!iter->Valid());
   iter->SeekToLast();
   ASSERT_TRUE(!iter->Valid());
   iter->Seek("foo");
   ASSERT_TRUE(!iter->Valid());
   delete iter;
 }
 
 // Test the empty key
 TEST(Harness, SimpleEmptyKey) {
   for (int i = 0; i < kNumTestArgs; i++) {
     Init(kTestArgList[i]);
     Random rnd(test::RandomSeed() + 1);
     Add("", "v");
     Test(&rnd);
   }
 }
 
 TEST(Harness, SimpleSingle) {
   for (int i = 0; i < kNumTestArgs; i++) {
     Init(kTestArgList[i]);
     Random rnd(test::RandomSeed() + 2);
     Add("abc", "v");
     Test(&rnd);
   }
 }
 
 TEST(Harness, SimpleMulti) {
   for (int i = 0; i < kNumTestArgs; i++) {
     Init(kTestArgList[i]);
     Random rnd(test::RandomSeed() + 3);
     Add("abc", "v");
     Add("abcd", "v");
     Add("ac", "v2");
     Test(&rnd);
   }
 }
 
 TEST(Harness, SimpleSpecialKey) {
   for (int i = 0; i < kNumTestArgs; i++) {
     Init(kTestArgList[i]);
     Random rnd(test::RandomSeed() + 4);
     Add("\xff\xff", "v3");
     Test(&rnd);
   }
 }
 
 TEST(Harness, Randomized) {
   for (int i = 0; i < kNumTestArgs; i++) {
     Init(kTestArgList[i]);
     Random rnd(test::RandomSeed() + 5);
     for (int num_entries = 0; num_entries < 2000;
          num_entries += (num_entries < 50 ? 1 : 200)) {
       if ((num_entries % 10) == 0) {
         fprintf(stderr, "case %d of %d: num_entries = %d\n",
                 (i + 1), int(kNumTestArgs), num_entries);
       }
       for (int e = 0; e < num_entries; e++) {
         std::string v;
         Add(test::RandomKey(&rnd, rnd.Skewed(4)),
             test::RandomString(&rnd, rnd.Skewed(5), &v).ToString());
       }
       Test(&rnd);
     }
   }
 }
 
 TEST(Harness, RandomizedLongDB) {
   Random rnd(test::RandomSeed());
   TestArgs args = { DB_TEST, false, 16 };
   Init(args);
   int num_entries = 100000;
   for (int e = 0; e < num_entries; e++) {
     std::string v;
     Add(test::RandomKey(&rnd, rnd.Skewed(4)),
         test::RandomString(&rnd, rnd.Skewed(5), &v).ToString());
   }
   Test(&rnd);
 
   // We must have created enough data to force merging
   int files = 0;
   for (int level = 0; level < config::kNumLevels; level++) {
     std::string value;
     char name[100];
     snprintf(name, sizeof(name), "leveldb.num-files-at-level%d", level);
     ASSERT_TRUE(db()->GetProperty(name, &value));
     files += atoi(value.c_str());
   }
   ASSERT_GT(files, 0);
 }
 
 class MemTableTest { };
 
 TEST(MemTableTest, Simple) {
   InternalKeyComparator cmp(BytewiseComparator());
   MemTable* memtable = new MemTable(cmp);
   memtable->Ref();
   WriteBatch batch;
   WriteBatchInternal::SetSequence(&batch, 100);
   batch.Put(std::string("k1"), std::string("v1"));
   batch.Put(std::string("k2"), std::string("v2"));
   batch.Put(std::string("k3"), std::string("v3"));
   batch.Put(std::string("largekey"), std::string("vlarge"));
   ASSERT_TRUE(WriteBatchInternal::InsertInto(&batch, memtable).ok());
 
   Iterator* iter = memtable->NewIterator();
   iter->SeekToFirst();
   while (iter->Valid()) {
     fprintf(stderr, "key: '%s' -> '%s'\n",
             iter->key().ToString().c_str(),
             iter->value().ToString().c_str());
     iter->Next();
   }
 
   delete iter;
   memtable->Unref();
 }
 
 static bool Between(uint64_t val, uint64_t low, uint64_t high) {
   bool result = (val >= low) && (val <= high);
   if (!result) {
     fprintf(stderr, "Value %llu is not in range [%llu, %llu]\n",
             (unsigned long long)(val),
             (unsigned long long)(low),
             (unsigned long long)(high));
   }
   return result;
 }
 
 class TableTest { };
 
 TEST(TableTest, ApproximateOffsetOfPlain) {
   TableConstructor c(BytewiseComparator());
   c.Add("k01", "hello");
   c.Add("k02", "hello2");
   c.Add("k03", std::string(10000, 'x'));
   c.Add("k04", std::string(200000, 'x'));
   c.Add("k05", std::string(300000, 'x'));
   c.Add("k06", "hello3");
   c.Add("k07", std::string(100000, 'x'));
   std::vector<std::string> keys;
   KVMap kvmap;
   Options options;
   options.block_size = 1024;
   options.compression = kNoCompression;
   c.Finish(options, &keys, &kvmap);
 
   ASSERT_TRUE(Between(c.ApproximateOffsetOf("abc"),       0,      0));
   ASSERT_TRUE(Between(c.ApproximateOffsetOf("k01"),       0,      0));
   ASSERT_TRUE(Between(c.ApproximateOffsetOf("k01a"),      0,      0));
   ASSERT_TRUE(Between(c.ApproximateOffsetOf("k02"),       0,      0));
   ASSERT_TRUE(Between(c.ApproximateOffsetOf("k03"),       0,      0));
   ASSERT_TRUE(Between(c.ApproximateOffsetOf("k04"),   10000,  11000));
   ASSERT_TRUE(Between(c.ApproximateOffsetOf("k04a"), 210000, 211000));
   ASSERT_TRUE(Between(c.ApproximateOffsetOf("k05"),  210000, 211000));
   ASSERT_TRUE(Between(c.ApproximateOffsetOf("k06"),  510000, 511000));
   ASSERT_TRUE(Between(c.ApproximateOffsetOf("k07"),  510000, 511000));
   ASSERT_TRUE(Between(c.ApproximateOffsetOf("xyz"),  610000, 612000));
 
 }
 
 static bool SnappyCompressionSupported() {
   std::string out;
   Slice in = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa";
   return port::Snappy_Compress(in.data(), in.size(), &out);
 }
 
 TEST(TableTest, ApproximateOffsetOfCompressed) {
   if (!SnappyCompressionSupported()) {
     fprintf(stderr, "skipping compression tests\n");
     return;
   }
 
   Random rnd(301);
   TableConstructor c(BytewiseComparator());
   std::string tmp;
   c.Add("k01", "hello");
   c.Add("k02", test::CompressibleString(&rnd, 0.25, 10000, &tmp));
   c.Add("k03", "hello3");
   c.Add("k04", test::CompressibleString(&rnd, 0.25, 10000, &tmp));
   std::vector<std::string> keys;
   KVMap kvmap;
   Options options;
   options.block_size = 1024;
   options.compression = kSnappyCompression;
   c.Finish(options, &keys, &kvmap);
 
   // Expected upper and lower bounds of space used by compressible strings.
   static const int kSlop = 1000;  // Compressor effectiveness varies.
   const int expected = 2500;  // 10000 * compression ratio (0.25)
   const int min_z = expected - kSlop;
   const int max_z = expected + kSlop;
 
   ASSERT_TRUE(Between(c.ApproximateOffsetOf("abc"), 0, kSlop));
   ASSERT_TRUE(Between(c.ApproximateOffsetOf("k01"), 0, kSlop));
   ASSERT_TRUE(Between(c.ApproximateOffsetOf("k02"), 0, kSlop));
   // Have now emitted a large compressible string, so adjust expected offset.
   ASSERT_TRUE(Between(c.ApproximateOffsetOf("k03"), min_z, max_z));
   ASSERT_TRUE(Between(c.ApproximateOffsetOf("k04"), min_z, max_z));
   // Have now emitted two large compressible strings, so adjust expected offset.
   ASSERT_TRUE(Between(c.ApproximateOffsetOf("xyz"), 2 * min_z, 2 * max_z));
 }
 
 }  // namespace leveldb
 
 int main(int argc, char** argv) {
   return leveldb::test::RunAllTests();
 }