src/leveldb/db/version_set.cc

   1 // Copyright (c) 2011 The LevelDB Authors. All rights reserved.
   2 // Use of this source code is governed by a BSD-style license that can be
   3 // found in the LICENSE file. See the AUTHORS file for names of contributors.
   4
   5 #include "db/version_set.h"
   6
   7 #include <algorithm>
   8 #include <stdio.h>
   9 #include "db/filename.h"
  10 #include "db/log_reader.h"
  11 #include "db/log_writer.h"
  12 #include "db/memtable.h"
  13 #include "db/table_cache.h"
  14 #include "leveldb/env.h"
  15 #include "leveldb/table_builder.h"
  16 #include "table/merger.h"
  17 #include "table/two_level_iterator.h"
  18 #include "util/coding.h"
  19 #include "util/logging.h"
  20
  21 namespace leveldb {
  22
  23 static const int kTargetFileSize = 2 * 1048576;
  24
  25 // Maximum bytes of overlaps in grandparent (i.e., level+2) before we
  26 // stop building a single file in a level->level+1 compaction.
  27 static const int64_t kMaxGrandParentOverlapBytes = 10 * kTargetFileSize;
  28
  29 // Maximum number of bytes in all compacted files.  We avoid expanding
  30 // the lower level file set of a compaction if it would make the
  31 // total compaction cover more than this many bytes.
  32 static const int64_t kExpandedCompactionByteSizeLimit = 25 * kTargetFileSize;
  33
  34 static double MaxBytesForLevel(int level) {
  35   // Note: the result for level zero is not really used since we set
  36   // the level-0 compaction threshold based on number of files.
  37   double result = 10 * 1048576.0;  // Result for both level-0 and level-1
  38   while (level > 1) {
  39     result *= 10;
  40     level--;
  41   }
  42   return result;
  43 }
  44
  45 static uint64_t MaxFileSizeForLevel(int level) {
  46   return kTargetFileSize;  // We could vary per level to reduce number of files?
  47 }
  48
  49 static int64_t TotalFileSize(const std::vector<FileMetaData*>& files) {
  50   int64_t sum = 0;
  51   for (size_t i = 0; i < files.size(); i++) {
  52     sum += files[i]->file_size;
  53   }
  54   return sum;
  55 }
  56
  57 Version::~Version() {
  58   assert(refs_ == 0);
  59
  60   // Remove from linked list
  61   prev_->next_ = next_;
  62   next_->prev_ = prev_;
  63
  64   // Drop references to files
  65   for (int level = 0; level < config::kNumLevels; level++) {
  66     for (size_t i = 0; i < files_[level].size(); i++) {
  67       FileMetaData* f = files_[level][i];
  68       assert(f->refs > 0);
  69       f->refs--;
  70       if (f->refs <= 0) {
  71         delete f;
  72       }
  73     }
  74   }
  75 }
  76
  77 int FindFile(const InternalKeyComparator& icmp,
  78              const std::vector<FileMetaData*>& files,
  79              const Slice& key) {
  80   uint32_t left = 0;
  81   uint32_t right = files.size();
  82   while (left < right) {
  83     uint32_t mid = (left + right) / 2;
  84     const FileMetaData* f = files[mid];
  85     if (icmp.InternalKeyComparator::Compare(f->largest.Encode(), key) < 0) {
  86       // Key at "mid.largest" is < "target".  Therefore all
  87       // files at or before "mid" are uninteresting.
  88       left = mid + 1;
  89     } else {
  90       // Key at "mid.largest" is >= "target".  Therefore all files
  91       // after "mid" are uninteresting.
  92       right = mid;
  93     }
  94   }
  95   return right;
  96 }
  97
  98 static bool AfterFile(const Comparator* ucmp,
  99                       const Slice* user_key, const FileMetaData* f) {
 100   // NULL user_key occurs before all keys and is therefore never after *f
 101   return (user_key != NULL &&
 102           ucmp->Compare(*user_key, f->largest.user_key()) > 0);
 103 }
 104
 105 static bool BeforeFile(const Comparator* ucmp,
 106                        const Slice* user_key, const FileMetaData* f) {
 107   // NULL user_key occurs after all keys and is therefore never before *f
 108   return (user_key != NULL &&
 109           ucmp->Compare(*user_key, f->smallest.user_key()) < 0);
 110 }
 111
 112 bool SomeFileOverlapsRange(
 113     const InternalKeyComparator& icmp,
 114     bool disjoint_sorted_files,
 115     const std::vector<FileMetaData*>& files,
 116     const Slice* smallest_user_key,
 117     const Slice* largest_user_key) {
 118   const Comparator* ucmp = icmp.user_comparator();
 119   if (!disjoint_sorted_files) {
 120     // Need to check against all files
 121     for (size_t i = 0; i < files.size(); i++) {
 122       const FileMetaData* f = files[i];
 123       if (AfterFile(ucmp, smallest_user_key, f) ||
 124           BeforeFile(ucmp, largest_user_key, f)) {
 125         // No overlap
 126       } else {
 127         return true;  // Overlap
 128       }
 129     }
 130     return false;
 131   }
 132
 133   // Binary search over file list
 134   uint32_t index = 0;
 135   if (smallest_user_key != NULL) {
 136     // Find the earliest possible internal key for smallest_user_key
 137     InternalKey small(*smallest_user_key, kMaxSequenceNumber,kValueTypeForSeek);
 138     index = FindFile(icmp, files, small.Encode());
 139   }
 140
 141   if (index >= files.size()) {
 142     // beginning of range is after all files, so no overlap.
 143     return false;
 144   }
 145
 146   return !BeforeFile(ucmp, largest_user_key, files[index]);
 147 }
 148
 149 // An internal iterator.  For a given version/level pair, yields
 150 // information about the files in the level.  For a given entry, key()
 151 // is the largest key that occurs in the file, and value() is an
 152 // 16-byte value containing the file number and file size, both
 153 // encoded using EncodeFixed64.
 154 class Version::LevelFileNumIterator : public Iterator {
 155  public:
 156   LevelFileNumIterator(const InternalKeyComparator& icmp,
 157                        const std::vector<FileMetaData*>* flist)
 158       : icmp_(icmp),
 159         flist_(flist),
 160         index_(flist->size()) {        // Marks as invalid
 161   }
 162   virtual bool Valid() const {
 163     return index_ < flist_->size();
 164   }
 165   virtual void Seek(const Slice& target) {
 166     index_ = FindFile(icmp_, *flist_, target);
 167   }
 168   virtual void SeekToFirst() { index_ = 0; }
 169   virtual void SeekToLast() {
 170     index_ = flist_->empty() ? 0 : flist_->size() - 1;
 171   }
 172   virtual void Next() {
 173     assert(Valid());
 174     index_++;
 175   }
 176   virtual void Prev() {
 177     assert(Valid());
 178     if (index_ == 0) {
 179       index_ = flist_->size();  // Marks as invalid
 180     } else {
 181       index_--;
 182     }
 183   }
 184   Slice key() const {
 185     assert(Valid());
 186     return (*flist_)[index_]->largest.Encode();
 187   }
 188   Slice value() const {
 189     assert(Valid());
 190     EncodeFixed64(value_buf_, (*flist_)[index_]->number);
 191     EncodeFixed64(value_buf_+8, (*flist_)[index_]->file_size);
 192     return Slice(value_buf_, sizeof(value_buf_));
 193   }
 194   virtual Status status() const { return Status::OK(); }
 195  private:
 196   const InternalKeyComparator icmp_;
 197   const std::vector<FileMetaData*>* const flist_;
 198   uint32_t index_;
 199
 200   // Backing store for value().  Holds the file number and size.
 201   mutable char value_buf_[16];
 202 };
 203
 204 static Iterator* GetFileIterator(void* arg,
 205                                  const ReadOptions& options,
 206                                  const Slice& file_value) {
 207   TableCache* cache = reinterpret_cast<TableCache*>(arg);
 208   if (file_value.size() != 16) {
 209     return NewErrorIterator(
 210         Status::Corruption("FileReader invoked with unexpected value"));
 211   } else {
 212     return cache->NewIterator(options,
 213                               DecodeFixed64(file_value.data()),
 214                               DecodeFixed64(file_value.data() + 8));
 215   }
 216 }
 217
 218 Iterator* Version::NewConcatenatingIterator(const ReadOptions& options,
 219                                             int level) const {
 220   return NewTwoLevelIterator(
 221       new LevelFileNumIterator(vset_->icmp_, &files_[level]),
 222       &GetFileIterator, vset_->table_cache_, options);
 223 }
 224
 225 void Version::AddIterators(const ReadOptions& options,
 226                            std::vector<Iterator*>* iters) {
 227   // Merge all level zero files together since they may overlap
 228   for (size_t i = 0; i < files_[0].size(); i++) {
 229     iters->push_back(
 230         vset_->table_cache_->NewIterator(
 231             options, files_[0][i]->number, files_[0][i]->file_size));
 232   }
 233
 234   // For levels > 0, we can use a concatenating iterator that sequentially
 235   // walks through the non-overlapping files in the level, opening them
 236   // lazily.
 237   for (int level = 1; level < config::kNumLevels; level++) {
 238     if (!files_[level].empty()) {
 239       iters->push_back(NewConcatenatingIterator(options, level));
 240     }
 241   }
 242 }
 243
 244 // Callback from TableCache::Get()
 245 namespace {
 246 enum SaverState {
 247   kNotFound,
 248   kFound,
 249   kDeleted,
 250   kCorrupt,
 251 };
 252 struct Saver {
 253   SaverState state;
 254   const Comparator* ucmp;
 255   Slice user_key;
 256   std::string* value;
 257 };
 258 }
 259 static void SaveValue(void* arg, const Slice& ikey, const Slice& v) {
 260   Saver* s = reinterpret_cast<Saver*>(arg);
 261   ParsedInternalKey parsed_key;
 262   if (!ParseInternalKey(ikey, &parsed_key)) {
 263     s->state = kCorrupt;
 264   } else {
 265     if (s->ucmp->Compare(parsed_key.user_key, s->user_key) == 0) {
 266       s->state = (parsed_key.type == kTypeValue) ? kFound : kDeleted;
 267       if (s->state == kFound) {
 268         s->value->assign(v.data(), v.size());
 269       }
 270     }
 271   }
 272 }
 273
 274 static bool NewestFirst(FileMetaData* a, FileMetaData* b) {
 275   return a->number > b->number;
 276 }
 277
 278 void Version::ForEachOverlapping(Slice user_key, Slice internal_key,
 279                                  void* arg,
 280                                  bool (*func)(void*, int, FileMetaData*)) {
 281   // TODO(sanjay): Change Version::Get() to use this function.
 282   const Comparator* ucmp = vset_->icmp_.user_comparator();
 283
 284   // Search level-0 in order from newest to oldest.
 285   std::vector<FileMetaData*> tmp;
 286   tmp.reserve(files_[0].size());
 287   for (uint32_t i = 0; i < files_[0].size(); i++) {
 288     FileMetaData* f = files_[0][i];
 289     if (ucmp->Compare(user_key, f->smallest.user_key()) >= 0 &&
 290         ucmp->Compare(user_key, f->largest.user_key()) <= 0) {
 291       tmp.push_back(f);
 292     }
 293   }
 294   if (!tmp.empty()) {
 295     std::sort(tmp.begin(), tmp.end(), NewestFirst);
 296     for (uint32_t i = 0; i < tmp.size(); i++) {
 297       if (!(*func)(arg, 0, tmp[i])) {
 298         return;
 299       }
 300     }
 301   }
 302
 303   // Search other levels.
 304   for (int level = 1; level < config::kNumLevels; level++) {
 305     size_t num_files = files_[level].size();
 306     if (num_files == 0) continue;
 307
 308     // Binary search to find earliest index whose largest key >= internal_key.
 309     uint32_t index = FindFile(vset_->icmp_, files_[level], internal_key);
 310     if (index < num_files) {
 311       FileMetaData* f = files_[level][index];
 312       if (ucmp->Compare(user_key, f->smallest.user_key()) < 0) {
 313         // All of "f" is past any data for user_key
 314       } else {
 315         if (!(*func)(arg, level, f)) {
 316           return;
 317         }
 318       }
 319     }
 320   }
 321 }
 322
 323 Status Version::Get(const ReadOptions& options,
 324                     const LookupKey& k,
 325                     std::string* value,
 326                     GetStats* stats) {
 327   Slice ikey = k.internal_key();
 328   Slice user_key = k.user_key();
 329   const Comparator* ucmp = vset_->icmp_.user_comparator();
 330   Status s;
 331
 332   stats->seek_file = NULL;
 333   stats->seek_file_level = -1;
 334   FileMetaData* last_file_read = NULL;
 335   int last_file_read_level = -1;
 336
 337   // We can search level-by-level since entries never hop across
 338   // levels.  Therefore we are guaranteed that if we find data
 339   // in an smaller level, later levels are irrelevant.
 340   std::vector<FileMetaData*> tmp;
 341   FileMetaData* tmp2;
 342   for (int level = 0; level < config::kNumLevels; level++) {
 343     size_t num_files = files_[level].size();
 344     if (num_files == 0) continue;
 345
 346     // Get the list of files to search in this level
 347     FileMetaData* const* files = &files_[level][0];
 348     if (level == 0) {
 349       // Level-0 files may overlap each other.  Find all files that
 350       // overlap user_key and process them in order from newest to oldest.
 351       tmp.reserve(num_files);
 352       for (uint32_t i = 0; i < num_files; i++) {
 353         FileMetaData* f = files[i];
 354         if (ucmp->Compare(user_key, f->smallest.user_key()) >= 0 &&
 355             ucmp->Compare(user_key, f->largest.user_key()) <= 0) {
 356           tmp.push_back(f);
 357         }
 358       }
 359       if (tmp.empty()) continue;
 360
 361       std::sort(tmp.begin(), tmp.end(), NewestFirst);
 362       files = &tmp[0];
 363       num_files = tmp.size();
 364     } else {
 365       // Binary search to find earliest index whose largest key >= ikey.
 366       uint32_t index = FindFile(vset_->icmp_, files_[level], ikey);
 367       if (index >= num_files) {
 368         files = NULL;
 369         num_files = 0;
 370       } else {
 371         tmp2 = files[index];
 372         if (ucmp->Compare(user_key, tmp2->smallest.user_key()) < 0) {
 373           // All of "tmp2" is past any data for user_key
 374           files = NULL;
 375           num_files = 0;
 376         } else {
 377           files = &tmp2;
 378           num_files = 1;
 379         }
 380       }
 381     }
 382
 383     for (uint32_t i = 0; i < num_files; ++i) {
 384       if (last_file_read != NULL && stats->seek_file == NULL) {
 385         // We have had more than one seek for this read.  Charge the 1st file.
 386         stats->seek_file = last_file_read;
 387         stats->seek_file_level = last_file_read_level;
 388       }
 389
 390       FileMetaData* f = files[i];
 391       last_file_read = f;
 392       last_file_read_level = level;
 393
 394       Saver saver;
 395       saver.state = kNotFound;
 396       saver.ucmp = ucmp;
 397       saver.user_key = user_key;
 398       saver.value = value;
 399       s = vset_->table_cache_->Get(options, f->number, f->file_size,
 400                                    ikey, &saver, SaveValue);
 401       if (!s.ok()) {
 402         return s;
 403       }
 404       switch (saver.state) {
 405         case kNotFound:
 406           break;      // Keep searching in other files
 407         case kFound:
 408           return s;
 409         case kDeleted:
 410           s = Status::NotFound(Slice());  // Use empty error message for speed
 411           return s;
 412         case kCorrupt:
 413           s = Status::Corruption("corrupted key for ", user_key);
 414           return s;
 415       }
 416     }
 417   }
 418
 419   return Status::NotFound(Slice());  // Use an empty error message for speed
 420 }
 421
 422 bool Version::UpdateStats(const GetStats& stats) {
 423   FileMetaData* f = stats.seek_file;
 424   if (f != NULL) {
 425     f->allowed_seeks--;
 426     if (f->allowed_seeks <= 0 && file_to_compact_ == NULL) {
 427       file_to_compact_ = f;
 428       file_to_compact_level_ = stats.seek_file_level;
 429       return true;
 430     }
 431   }
 432   return false;
 433 }
 434
 435 bool Version::RecordReadSample(Slice internal_key) {
 436   ParsedInternalKey ikey;
 437   if (!ParseInternalKey(internal_key, &ikey)) {
 438     return false;
 439   }
 440
 441   struct State {
 442     GetStats stats;  // Holds first matching file
 443     int matches;
 444
 445     static bool Match(void* arg, int level, FileMetaData* f) {
 446       State* state = reinterpret_cast<State*>(arg);
 447       state->matches++;
 448       if (state->matches == 1) {
 449         // Remember first match.
 450         state->stats.seek_file = f;
 451         state->stats.seek_file_level = level;
 452       }
 453       // We can stop iterating once we have a second match.
 454       return state->matches < 2;
 455     }
 456   };
 457
 458   State state;
 459   state.matches = 0;
 460   ForEachOverlapping(ikey.user_key, internal_key, &state, &State::Match);
 461
 462   // Must have at least two matches since we want to merge across
 463   // files. But what if we have a single file that contains many
 464   // overwrites and deletions?  Should we have another mechanism for
 465   // finding such files?
 466   if (state.matches >= 2) {
 467     // 1MB cost is about 1 seek (see comment in Builder::Apply).
 468     return UpdateStats(state.stats);
 469   }
 470   return false;
 471 }
 472
 473 void Version::Ref() {
 474   ++refs_;
 475 }
 476
 477 void Version::Unref() {
 478   assert(this != &vset_->dummy_versions_);
 479   assert(refs_ >= 1);
 480   --refs_;
 481   if (refs_ == 0) {
 482     delete this;
 483   }
 484 }
 485
 486 bool Version::OverlapInLevel(int level,
 487                              const Slice* smallest_user_key,
 488                              const Slice* largest_user_key) {
 489   return SomeFileOverlapsRange(vset_->icmp_, (level > 0), files_[level],
 490                                smallest_user_key, largest_user_key);
 491 }
 492
 493 int Version::PickLevelForMemTableOutput(
 494     const Slice& smallest_user_key,
 495     const Slice& largest_user_key) {
 496   int level = 0;
 497   if (!OverlapInLevel(0, &smallest_user_key, &largest_user_key)) {
 498     // Push to next level if there is no overlap in next level,
 499     // and the #bytes overlapping in the level after that are limited.
 500     InternalKey start(smallest_user_key, kMaxSequenceNumber, kValueTypeForSeek);
 501     InternalKey limit(largest_user_key, 0, static_cast<ValueType>(0));
 502     std::vector<FileMetaData*> overlaps;
 503     while (level < config::kMaxMemCompactLevel) {
 504       if (OverlapInLevel(level + 1, &smallest_user_key, &largest_user_key)) {
 505         break;
 506       }
 507       if (level + 2 < config::kNumLevels) {
 508         // Check that file does not overlap too many grandparent bytes.
 509         GetOverlappingInputs(level + 2, &start, &limit, &overlaps);
 510         const int64_t sum = TotalFileSize(overlaps);
 511         if (sum > kMaxGrandParentOverlapBytes) {
 512           break;
 513         }
 514       }
 515       level++;
 516     }
 517   }
 518   return level;
 519 }
 520
 521 // Store in "*inputs" all files in "level" that overlap [begin,end]
 522 void Version::GetOverlappingInputs(
 523     int level,
 524     const InternalKey* begin,
 525     const InternalKey* end,
 526     std::vector<FileMetaData*>* inputs) {
 527   assert(level >= 0);
 528   assert(level < config::kNumLevels);
 529   inputs->clear();
 530   Slice user_begin, user_end;
 531   if (begin != NULL) {
 532     user_begin = begin->user_key();
 533   }
 534   if (end != NULL) {
 535     user_end = end->user_key();
 536   }
 537   const Comparator* user_cmp = vset_->icmp_.user_comparator();
 538   for (size_t i = 0; i < files_[level].size(); ) {
 539     FileMetaData* f = files_[level][i++];
 540     const Slice file_start = f->smallest.user_key();
 541     const Slice file_limit = f->largest.user_key();
 542     if (begin != NULL && user_cmp->Compare(file_limit, user_begin) < 0) {
 543       // "f" is completely before specified range; skip it
 544     } else if (end != NULL && user_cmp->Compare(file_start, user_end) > 0) {
 545       // "f" is completely after specified range; skip it
 546     } else {
 547       inputs->push_back(f);
 548       if (level == 0) {
 549         // Level-0 files may overlap each other.  So check if the newly
 550         // added file has expanded the range.  If so, restart search.
 551         if (begin != NULL && user_cmp->Compare(file_start, user_begin) < 0) {
 552           user_begin = file_start;
 553           inputs->clear();
 554           i = 0;
 555         } else if (end != NULL && user_cmp->Compare(file_limit, user_end) > 0) {
 556           user_end = file_limit;
 557           inputs->clear();
 558           i = 0;
 559         }
 560       }
 561     }
 562   }
 563 }
 564
 565 std::string Version::DebugString() const {
 566   std::string r;
 567   for (int level = 0; level < config::kNumLevels; level++) {
 568     // E.g.,
 569     //   --- level 1 ---
 570     //   17:123['a' .. 'd']
 571     //   20:43['e' .. 'g']
 572     r.append("--- level ");
 573     AppendNumberTo(&r, level);
 574     r.append(" ---\n");
 575     const std::vector<FileMetaData*>& files = files_[level];
 576     for (size_t i = 0; i < files.size(); i++) {
 577       r.push_back(' ');
 578       AppendNumberTo(&r, files[i]->number);
 579       r.push_back(':');
 580       AppendNumberTo(&r, files[i]->file_size);
 581       r.append("[");
 582       r.append(files[i]->smallest.DebugString());
 583       r.append(" .. ");
 584       r.append(files[i]->largest.DebugString());
 585       r.append("]\n");
 586     }
 587   }
 588   return r;
 589 }
 590
 591 // A helper class so we can efficiently apply a whole sequence
 592 // of edits to a particular state without creating intermediate
 593 // Versions that contain full copies of the intermediate state.
 594 class VersionSet::Builder {
 595  private:
 596   // Helper to sort by v->files_[file_number].smallest
 597   struct BySmallestKey {
 598     const InternalKeyComparator* internal_comparator;
 599
 600     bool operator()(FileMetaData* f1, FileMetaData* f2) const {
 601       int r = internal_comparator->Compare(f1->smallest, f2->smallest);
 602       if (r != 0) {
 603         return (r < 0);
 604       } else {
 605         // Break ties by file number
 606         return (f1->number < f2->number);
 607       }
 608     }
 609   };
 610
 611   typedef std::set<FileMetaData*, BySmallestKey> FileSet;
 612   struct LevelState {
 613     std::set<uint64_t> deleted_files;
 614     FileSet* added_files;
 615   };
 616
 617   VersionSet* vset_;
 618   Version* base_;
 619   LevelState levels_[config::kNumLevels];
 620
 621  public:
 622   // Initialize a builder with the files from *base and other info from *vset
 623   Builder(VersionSet* vset, Version* base)
 624       : vset_(vset),
 625         base_(base) {
 626     base_->Ref();
 627     BySmallestKey cmp;
 628     cmp.internal_comparator = &vset_->icmp_;
 629     for (int level = 0; level < config::kNumLevels; level++) {
 630       levels_[level].added_files = new FileSet(cmp);
 631     }
 632   }
 633
 634   ~Builder() {
 635     for (int level = 0; level < config::kNumLevels; level++) {
 636       const FileSet* added = levels_[level].added_files;
 637       std::vector<FileMetaData*> to_unref;
 638       to_unref.reserve(added->size());
 639       for (FileSet::const_iterator it = added->begin();
 640           it != added->end(); ++it) {
 641         to_unref.push_back(*it);
 642       }
 643       delete added;
 644       for (uint32_t i = 0; i < to_unref.size(); i++) {
 645         FileMetaData* f = to_unref[i];
 646         f->refs--;
 647         if (f->refs <= 0) {
 648           delete f;
 649         }
 650       }
 651     }
 652     base_->Unref();
 653   }
 654
 655   // Apply all of the edits in *edit to the current state.
 656   void Apply(VersionEdit* edit) {
 657     // Update compaction pointers
 658     for (size_t i = 0; i < edit->compact_pointers_.size(); i++) {
 659       const int level = edit->compact_pointers_[i].first;
 660       vset_->compact_pointer_[level] =
 661           edit->compact_pointers_[i].second.Encode().ToString();
 662     }
 663
 664     // Delete files
 665     const VersionEdit::DeletedFileSet& del = edit->deleted_files_;
 666     for (VersionEdit::DeletedFileSet::const_iterator iter = del.begin();
 667          iter != del.end();
 668          ++iter) {
 669       const int level = iter->first;
 670       const uint64_t number = iter->second;
 671       levels_[level].deleted_files.insert(number);
 672     }
 673
 674     // Add new files
 675     for (size_t i = 0; i < edit->new_files_.size(); i++) {
 676       const int level = edit->new_files_[i].first;
 677       FileMetaData* f = new FileMetaData(edit->new_files_[i].second);
 678       f->refs = 1;
 679
 680       // We arrange to automatically compact this file after
 681       // a certain number of seeks.  Let's assume:
 682       //   (1) One seek costs 10ms
 683       //   (2) Writing or reading 1MB costs 10ms (100MB/s)
 684       //   (3) A compaction of 1MB does 25MB of IO:
 685       //         1MB read from this level
 686       //         10-12MB read from next level (boundaries may be misaligned)
 687       //         10-12MB written to next level
 688       // This implies that 25 seeks cost the same as the compaction
 689       // of 1MB of data.  I.e., one seek costs approximately the
 690       // same as the compaction of 40KB of data.  We are a little
 691       // conservative and allow approximately one seek for every 16KB
 692       // of data before triggering a compaction.
 693       f->allowed_seeks = (f->file_size / 16384);
 694       if (f->allowed_seeks < 100) f->allowed_seeks = 100;
 695
 696       levels_[level].deleted_files.erase(f->number);
 697       levels_[level].added_files->insert(f);
 698     }
 699   }
 700
 701   // Save the current state in *v.
 702   void SaveTo(Version* v) {
 703     BySmallestKey cmp;
 704     cmp.internal_comparator = &vset_->icmp_;
 705     for (int level = 0; level < config::kNumLevels; level++) {
 706       // Merge the set of added files with the set of pre-existing files.
 707       // Drop any deleted files.  Store the result in *v.
 708       const std::vector<FileMetaData*>& base_files = base_->files_[level];
 709       std::vector<FileMetaData*>::const_iterator base_iter = base_files.begin();
 710       std::vector<FileMetaData*>::const_iterator base_end = base_files.end();
 711       const FileSet* added = levels_[level].added_files;
 712       v->files_[level].reserve(base_files.size() + added->size());
 713       for (FileSet::const_iterator added_iter = added->begin();
 714            added_iter != added->end();
 715            ++added_iter) {
 716         // Add all smaller files listed in base_
 717         for (std::vector<FileMetaData*>::const_iterator bpos
 718                  = std::upper_bound(base_iter, base_end, *added_iter, cmp);
 719              base_iter != bpos;
 720              ++base_iter) {
 721           MaybeAddFile(v, level, *base_iter);
 722         }
 723
 724         MaybeAddFile(v, level, *added_iter);
 725       }
 726
 727       // Add remaining base files
 728       for (; base_iter != base_end; ++base_iter) {
 729         MaybeAddFile(v, level, *base_iter);
 730       }
 731
 732 #ifndef NDEBUG
 733       // Make sure there is no overlap in levels > 0
 734       if (level > 0) {
 735         for (uint32_t i = 1; i < v->files_[level].size(); i++) {
 736           const InternalKey& prev_end = v->files_[level][i-1]->largest;
 737           const InternalKey& this_begin = v->files_[level][i]->smallest;
 738           if (vset_->icmp_.Compare(prev_end, this_begin) >= 0) {
 739             fprintf(stderr, "overlapping ranges in same level %s vs. %s\n",
 740                     prev_end.DebugString().c_str(),
 741                     this_begin.DebugString().c_str());
 742             abort();
 743           }
 744         }
 745       }
 746 #endif
 747     }
 748   }
 749
 750   void MaybeAddFile(Version* v, int level, FileMetaData* f) {
 751     if (levels_[level].deleted_files.count(f->number) > 0) {
 752       // File is deleted: do nothing
 753     } else {
 754       std::vector<FileMetaData*>* files = &v->files_[level];
 755       if (level > 0 && !files->empty()) {
 756         // Must not overlap
 757         assert(vset_->icmp_.Compare((*files)[files->size()-1]->largest,
 758                                     f->smallest) < 0);
 759       }
 760       f->refs++;
 761       files->push_back(f);
 762     }
 763   }
 764 };
 765
 766 VersionSet::VersionSet(const std::string& dbname,
 767                        const Options* options,
 768                        TableCache* table_cache,
 769                        const InternalKeyComparator* cmp)
 770     : env_(options->env),
 771       dbname_(dbname),
 772       options_(options),
 773       table_cache_(table_cache),
 774       icmp_(*cmp),
 775       next_file_number_(2),
 776       manifest_file_number_(0),  // Filled by Recover()
 777       last_sequence_(0),
 778       log_number_(0),
 779       prev_log_number_(0),
 780       descriptor_file_(NULL),
 781       descriptor_log_(NULL),
 782       dummy_versions_(this),
 783       current_(NULL) {
 784   AppendVersion(new Version(this));
 785 }
 786
 787 VersionSet::~VersionSet() {
 788   current_->Unref();
 789   assert(dummy_versions_.next_ == &dummy_versions_);  // List must be empty
 790   delete descriptor_log_;
 791   delete descriptor_file_;
 792 }
 793
 794 void VersionSet::AppendVersion(Version* v) {
 795   // Make "v" current
 796   assert(v->refs_ == 0);
 797   assert(v != current_);
 798   if (current_ != NULL) {
 799     current_->Unref();
 800   }
 801   current_ = v;
 802   v->Ref();
 803
 804   // Append to linked list
 805   v->prev_ = dummy_versions_.prev_;
 806   v->next_ = &dummy_versions_;
 807   v->prev_->next_ = v;
 808   v->next_->prev_ = v;
 809 }
 810
 811 Status VersionSet::LogAndApply(VersionEdit* edit, port::Mutex* mu) {
 812   if (edit->has_log_number_) {
 813     assert(edit->log_number_ >= log_number_);
 814     assert(edit->log_number_ < next_file_number_);
 815   } else {
 816     edit->SetLogNumber(log_number_);
 817   }
 818
 819   if (!edit->has_prev_log_number_) {
 820     edit->SetPrevLogNumber(prev_log_number_);
 821   }
 822
 823   edit->SetNextFile(next_file_number_);
 824   edit->SetLastSequence(last_sequence_);
 825
 826   Version* v = new Version(this);
 827   {
 828     Builder builder(this, current_);
 829     builder.Apply(edit);
 830     builder.SaveTo(v);
 831   }
 832   Finalize(v);
 833
 834   // Initialize new descriptor log file if necessary by creating
 835   // a temporary file that contains a snapshot of the current version.
 836   std::string new_manifest_file;
 837   Status s;
 838   if (descriptor_log_ == NULL) {
 839     // No reason to unlock *mu here since we only hit this path in the
 840     // first call to LogAndApply (when opening the database).
 841     assert(descriptor_file_ == NULL);
 842     new_manifest_file = DescriptorFileName(dbname_, manifest_file_number_);
 843     edit->SetNextFile(next_file_number_);
 844     s = env_->NewWritableFile(new_manifest_file, &descriptor_file_);
 845     if (s.ok()) {
 846       descriptor_log_ = new log::Writer(descriptor_file_);
 847       s = WriteSnapshot(descriptor_log_);
 848     }
 849   }
 850
 851   // Unlock during expensive MANIFEST log write
 852   {
 853     mu->Unlock();
 854
 855     // Write new record to MANIFEST log
 856     if (s.ok()) {
 857       std::string record;
 858       edit->EncodeTo(&record);
 859       s = descriptor_log_->AddRecord(record);
 860       if (s.ok()) {
 861         s = descriptor_file_->Sync();
 862       }
 863       if (!s.ok()) {
 864         Log(options_->info_log, "MANIFEST write: %s\n", s.ToString().c_str());
 865       }
 866     }
 867
 868     // If we just created a new descriptor file, install it by writing a
 869     // new CURRENT file that points to it.
 870     if (s.ok() && !new_manifest_file.empty()) {
 871       s = SetCurrentFile(env_, dbname_, manifest_file_number_);
 872     }
 873
 874     mu->Lock();
 875   }
 876
 877   // Install the new version
 878   if (s.ok()) {
 879     AppendVersion(v);
 880     log_number_ = edit->log_number_;
 881     prev_log_number_ = edit->prev_log_number_;
 882   } else {
 883     delete v;
 884     if (!new_manifest_file.empty()) {
 885       delete descriptor_log_;
 886       delete descriptor_file_;
 887       descriptor_log_ = NULL;
 888       descriptor_file_ = NULL;
 889       env_->DeleteFile(new_manifest_file);
 890     }
 891   }
 892
 893   return s;
 894 }
 895
 896 Status VersionSet::Recover() {
 897   struct LogReporter : public log::Reader::Reporter {
 898     Status* status;
 899     virtual void Corruption(size_t bytes, const Status& s) {
 900       if (this->status->ok()) *this->status = s;
 901     }
 902   };
 903
 904   // Read "CURRENT" file, which contains a pointer to the current manifest file
 905   std::string current;
 906   Status s = ReadFileToString(env_, CurrentFileName(dbname_), &current);
 907   if (!s.ok()) {
 908     return s;
 909   }
 910   if (current.empty() || current[current.size()-1] != '\n') {
 911     return Status::Corruption("CURRENT file does not end with newline");
 912   }
 913   current.resize(current.size() - 1);
 914
 915   std::string dscname = dbname_ + "/" + current;
 916   SequentialFile* file;
 917   s = env_->NewSequentialFile(dscname, &file);
 918   if (!s.ok()) {
 919     return s;
 920   }
 921
 922   bool have_log_number = false;
 923   bool have_prev_log_number = false;
 924   bool have_next_file = false;
 925   bool have_last_sequence = false;
 926   uint64_t next_file = 0;
 927   uint64_t last_sequence = 0;
 928   uint64_t log_number = 0;
 929   uint64_t prev_log_number = 0;
 930   Builder builder(this, current_);
 931
 932   {
 933     LogReporter reporter;
 934     reporter.status = &s;
 935     log::Reader reader(file, &reporter, true/*checksum*/, 0/*initial_offset*/);
 936     Slice record;
 937     std::string scratch;
 938     while (reader.ReadRecord(&record, &scratch) && s.ok()) {
 939       VersionEdit edit;
 940       s = edit.DecodeFrom(record);
 941       if (s.ok()) {
 942         if (edit.has_comparator_ &&
 943             edit.comparator_ != icmp_.user_comparator()->Name()) {
 944           s = Status::InvalidArgument(
 945               edit.comparator_ + " does not match existing comparator ",
 946               icmp_.user_comparator()->Name());
 947         }
 948       }
 949
 950       if (s.ok()) {
 951         builder.Apply(&edit);
 952       }
 953
 954       if (edit.has_log_number_) {
 955         log_number = edit.log_number_;
 956         have_log_number = true;
 957       }
 958
 959       if (edit.has_prev_log_number_) {
 960         prev_log_number = edit.prev_log_number_;
 961         have_prev_log_number = true;
 962       }
 963
 964       if (edit.has_next_file_number_) {
 965         next_file = edit.next_file_number_;
 966         have_next_file = true;
 967       }
 968
 969       if (edit.has_last_sequence_) {
 970         last_sequence = edit.last_sequence_;
 971         have_last_sequence = true;
 972       }
 973     }
 974   }
 975   delete file;
 976   file = NULL;
 977
 978   if (s.ok()) {
 979     if (!have_next_file) {
 980       s = Status::Corruption("no meta-nextfile entry in descriptor");
 981     } else if (!have_log_number) {
 982       s = Status::Corruption("no meta-lognumber entry in descriptor");
 983     } else if (!have_last_sequence) {
 984       s = Status::Corruption("no last-sequence-number entry in descriptor");
 985     }
 986
 987     if (!have_prev_log_number) {
 988       prev_log_number = 0;
 989     }
 990
 991     MarkFileNumberUsed(prev_log_number);
 992     MarkFileNumberUsed(log_number);
 993   }
 994
 995   if (s.ok()) {
 996     Version* v = new Version(this);
 997     builder.SaveTo(v);
 998     // Install recovered version
 999     Finalize(v);
1000     AppendVersion(v);
1001     manifest_file_number_ = next_file;
1002     next_file_number_ = next_file + 1;
1003     last_sequence_ = last_sequence;
1004     log_number_ = log_number;
1005     prev_log_number_ = prev_log_number;
1006   }
1007
1008   return s;
1009 }
1010
1011 void VersionSet::MarkFileNumberUsed(uint64_t number) {
1012   if (next_file_number_ <= number) {
1013     next_file_number_ = number + 1;
1014   }
1015 }
1016
1017 void VersionSet::Finalize(Version* v) {
1018   // Precomputed best level for next compaction
1019   int best_level = -1;
1020   double best_score = -1;
1021
1022   for (int level = 0; level < config::kNumLevels-1; level++) {
1023     double score;
1024     if (level == 0) {
1025       // We treat level-0 specially by bounding the number of files
1026       // instead of number of bytes for two reasons:
1027       //
1028       // (1) With larger write-buffer sizes, it is nice not to do too
1029       // many level-0 compactions.
1030       //
1031       // (2) The files in level-0 are merged on every read and
1032       // therefore we wish to avoid too many files when the individual
1033       // file size is small (perhaps because of a small write-buffer
1034       // setting, or very high compression ratios, or lots of
1035       // overwrites/deletions).
1036       score = v->files_[level].size() /
1037           static_cast<double>(config::kL0_CompactionTrigger);
1038     } else {
1039       // Compute the ratio of current size to size limit.
1040       const uint64_t level_bytes = TotalFileSize(v->files_[level]);
1041       score = static_cast<double>(level_bytes) / MaxBytesForLevel(level);
1042     }
1043
1044     if (score > best_score) {
1045       best_level = level;
1046       best_score = score;
1047     }
1048   }
1049
1050   v->compaction_level_ = best_level;
1051   v->compaction_score_ = best_score;
1052 }
1053
1054 Status VersionSet::WriteSnapshot(log::Writer* log) {
1055   // TODO: Break up into multiple records to reduce memory usage on recovery?
1056
1057   // Save metadata
1058   VersionEdit edit;
1059   edit.SetComparatorName(icmp_.user_comparator()->Name());
1060
1061   // Save compaction pointers
1062   for (int level = 0; level < config::kNumLevels; level++) {
1063     if (!compact_pointer_[level].empty()) {
1064       InternalKey key;
1065       key.DecodeFrom(compact_pointer_[level]);
1066       edit.SetCompactPointer(level, key);
1067     }
1068   }
1069
1070   // Save files
1071   for (int level = 0; level < config::kNumLevels; level++) {
1072     const std::vector<FileMetaData*>& files = current_->files_[level];
1073     for (size_t i = 0; i < files.size(); i++) {
1074       const FileMetaData* f = files[i];
1075       edit.AddFile(level, f->number, f->file_size, f->smallest, f->largest);
1076     }
1077   }
1078
1079   std::string record;
1080   edit.EncodeTo(&record);
1081   return log->AddRecord(record);
1082 }
1083
1084 int VersionSet::NumLevelFiles(int level) const {
1085   assert(level >= 0);
1086   assert(level < config::kNumLevels);
1087   return current_->files_[level].size();
1088 }
1089
1090 const char* VersionSet::LevelSummary(LevelSummaryStorage* scratch) const {
1091   // Update code if kNumLevels changes
1092   assert(config::kNumLevels == 7);
1093   snprintf(scratch->buffer, sizeof(scratch->buffer),
1094            "files[ %d %d %d %d %d %d %d ]",
1095            int(current_->files_[0].size()),
1096            int(current_->files_[1].size()),
1097            int(current_->files_[2].size()),
1098            int(current_->files_[3].size()),
1099            int(current_->files_[4].size()),
1100            int(current_->files_[5].size()),
1101            int(current_->files_[6].size()));
1102   return scratch->buffer;
1103 }
1104
1105 uint64_t VersionSet::ApproximateOffsetOf(Version* v, const InternalKey& ikey) {
1106   uint64_t result = 0;
1107   for (int level = 0; level < config::kNumLevels; level++) {
1108     const std::vector<FileMetaData*>& files = v->files_[level];
1109     for (size_t i = 0; i < files.size(); i++) {
1110       if (icmp_.Compare(files[i]->largest, ikey) <= 0) {
1111         // Entire file is before "ikey", so just add the file size
1112         result += files[i]->file_size;
1113       } else if (icmp_.Compare(files[i]->smallest, ikey) > 0) {
1114         // Entire file is after "ikey", so ignore
1115         if (level > 0) {
1116           // Files other than level 0 are sorted by meta->smallest, so
1117           // no further files in this level will contain data for
1118           // "ikey".
1119           break;
1120         }
1121       } else {
1122         // "ikey" falls in the range for this table.  Add the
1123         // approximate offset of "ikey" within the table.
1124         Table* tableptr;
1125         Iterator* iter = table_cache_->NewIterator(
1126             ReadOptions(), files[i]->number, files[i]->file_size, &tableptr);
1127         if (tableptr != NULL) {
1128           result += tableptr->ApproximateOffsetOf(ikey.Encode());
1129         }
1130         delete iter;
1131       }
1132     }
1133   }
1134   return result;
1135 }
1136
1137 void VersionSet::AddLiveFiles(std::set<uint64_t>* live) {
1138   for (Version* v = dummy_versions_.next_;
1139        v != &dummy_versions_;
1140        v = v->next_) {
1141     for (int level = 0; level < config::kNumLevels; level++) {
1142       const std::vector<FileMetaData*>& files = v->files_[level];
1143       for (size_t i = 0; i < files.size(); i++) {
1144         live->insert(files[i]->number);
1145       }
1146     }
1147   }
1148 }
1149
1150 int64_t VersionSet::NumLevelBytes(int level) const {
1151   assert(level >= 0);
1152   assert(level < config::kNumLevels);
1153   return TotalFileSize(current_->files_[level]);
1154 }
1155
1156 int64_t VersionSet::MaxNextLevelOverlappingBytes() {
1157   int64_t result = 0;
1158   std::vector<FileMetaData*> overlaps;
1159   for (int level = 1; level < config::kNumLevels - 1; level++) {
1160     for (size_t i = 0; i < current_->files_[level].size(); i++) {
1161       const FileMetaData* f = current_->files_[level][i];
1162       current_->GetOverlappingInputs(level+1, &f->smallest, &f->largest,
1163                                      &overlaps);
1164       const int64_t sum = TotalFileSize(overlaps);
1165       if (sum > result) {
1166         result = sum;
1167       }
1168     }
1169   }
1170   return result;
1171 }
1172
1173 // Stores the minimal range that covers all entries in inputs in
1174 // *smallest, *largest.
1175 // REQUIRES: inputs is not empty
1176 void VersionSet::GetRange(const std::vector<FileMetaData*>& inputs,
1177                           InternalKey* smallest,
1178                           InternalKey* largest) {
1179   assert(!inputs.empty());
1180   smallest->Clear();
1181   largest->Clear();
1182   for (size_t i = 0; i < inputs.size(); i++) {
1183     FileMetaData* f = inputs[i];
1184     if (i == 0) {
1185       *smallest = f->smallest;
1186       *largest = f->largest;
1187     } else {
1188       if (icmp_.Compare(f->smallest, *smallest) < 0) {
1189         *smallest = f->smallest;
1190       }
1191       if (icmp_.Compare(f->largest, *largest) > 0) {
1192         *largest = f->largest;
1193       }
1194     }
1195   }
1196 }
1197
1198 // Stores the minimal range that covers all entries in inputs1 and inputs2
1199 // in *smallest, *largest.
1200 // REQUIRES: inputs is not empty
1201 void VersionSet::GetRange2(const std::vector<FileMetaData*>& inputs1,
1202                            const std::vector<FileMetaData*>& inputs2,
1203                            InternalKey* smallest,
1204                            InternalKey* largest) {
1205   std::vector<FileMetaData*> all = inputs1;
1206   all.insert(all.end(), inputs2.begin(), inputs2.end());
1207   GetRange(all, smallest, largest);
1208 }
1209
1210 Iterator* VersionSet::MakeInputIterator(Compaction* c) {
1211   ReadOptions options;
1212   options.verify_checksums = options_->paranoid_checks;
1213   options.fill_cache = false;
1214
1215   // Level-0 files have to be merged together.  For other levels,
1216   // we will make a concatenating iterator per level.
1217   // TODO(opt): use concatenating iterator for level-0 if there is no overlap
1218   const int space = (c->level() == 0 ? c->inputs_[0].size() + 1 : 2);
1219   Iterator** list = new Iterator*[space];
1220   int num = 0;
1221   for (int which = 0; which < 2; which++) {
1222     if (!c->inputs_[which].empty()) {
1223       if (c->level() + which == 0) {
1224         const std::vector<FileMetaData*>& files = c->inputs_[which];
1225         for (size_t i = 0; i < files.size(); i++) {
1226           list[num++] = table_cache_->NewIterator(
1227               options, files[i]->number, files[i]->file_size);
1228         }
1229       } else {
1230         // Create concatenating iterator for the files from this level
1231         list[num++] = NewTwoLevelIterator(
1232             new Version::LevelFileNumIterator(icmp_, &c->inputs_[which]),
1233             &GetFileIterator, table_cache_, options);
1234       }
1235     }
1236   }
1237   assert(num <= space);
1238   Iterator* result = NewMergingIterator(&icmp_, list, num);
1239   delete[] list;
1240   return result;
1241 }
1242
1243 Compaction* VersionSet::PickCompaction() {
1244   Compaction* c;
1245   int level;
1246
1247   // We prefer compactions triggered by too much data in a level over
1248   // the compactions triggered by seeks.
1249   const bool size_compaction = (current_->compaction_score_ >= 1);
1250   const bool seek_compaction = (current_->file_to_compact_ != NULL);
1251   if (size_compaction) {
1252     level = current_->compaction_level_;
1253     assert(level >= 0);
1254     assert(level+1 < config::kNumLevels);
1255     c = new Compaction(level);
1256
1257     // Pick the first file that comes after compact_pointer_[level]
1258     for (size_t i = 0; i < current_->files_[level].size(); i++) {
1259       FileMetaData* f = current_->files_[level][i];
1260       if (compact_pointer_[level].empty() ||
1261           icmp_.Compare(f->largest.Encode(), compact_pointer_[level]) > 0) {
1262         c->inputs_[0].push_back(f);
1263         break;
1264       }
1265     }
1266     if (c->inputs_[0].empty()) {
1267       // Wrap-around to the beginning of the key space
1268       c->inputs_[0].push_back(current_->files_[level][0]);
1269     }
1270   } else if (seek_compaction) {
1271     level = current_->file_to_compact_level_;
1272     c = new Compaction(level);
1273     c->inputs_[0].push_back(current_->file_to_compact_);
1274   } else {
1275     return NULL;
1276   }
1277
1278   c->input_version_ = current_;
1279   c->input_version_->Ref();
1280
1281   // Files in level 0 may overlap each other, so pick up all overlapping ones
1282   if (level == 0) {
1283     InternalKey smallest, largest;
1284     GetRange(c->inputs_[0], &smallest, &largest);
1285     // Note that the next call will discard the file we placed in
1286     // c->inputs_[0] earlier and replace it with an overlapping set
1287     // which will include the picked file.
1288     current_->GetOverlappingInputs(0, &smallest, &largest, &c->inputs_[0]);
1289     assert(!c->inputs_[0].empty());
1290   }
1291
1292   SetupOtherInputs(c);
1293
1294   return c;
1295 }
1296
1297 void VersionSet::SetupOtherInputs(Compaction* c) {
1298   const int level = c->level();
1299   InternalKey smallest, largest;
1300   GetRange(c->inputs_[0], &smallest, &largest);
1301
1302   current_->GetOverlappingInputs(level+1, &smallest, &largest, &c->inputs_[1]);
1303
1304   // Get entire range covered by compaction
1305   InternalKey all_start, all_limit;
1306   GetRange2(c->inputs_[0], c->inputs_[1], &all_start, &all_limit);
1307
1308   // See if we can grow the number of inputs in "level" without
1309   // changing the number of "level+1" files we pick up.
1310   if (!c->inputs_[1].empty()) {
1311     std::vector<FileMetaData*> expanded0;
1312     current_->GetOverlappingInputs(level, &all_start, &all_limit, &expanded0);
1313     const int64_t inputs0_size = TotalFileSize(c->inputs_[0]);
1314     const int64_t inputs1_size = TotalFileSize(c->inputs_[1]);
1315     const int64_t expanded0_size = TotalFileSize(expanded0);
1316     if (expanded0.size() > c->inputs_[0].size() &&
1317         inputs1_size + expanded0_size < kExpandedCompactionByteSizeLimit) {
1318       InternalKey new_start, new_limit;
1319       GetRange(expanded0, &new_start, &new_limit);
1320       std::vector<FileMetaData*> expanded1;
1321       current_->GetOverlappingInputs(level+1, &new_start, &new_limit,
1322                                      &expanded1);
1323       if (expanded1.size() == c->inputs_[1].size()) {
1324         Log(options_->info_log,
1325             "Expanding@%d %d+%d (%ld+%ld bytes) to %d+%d (%ld+%ld bytes)\n",
1326             level,
1327             int(c->inputs_[0].size()),
1328             int(c->inputs_[1].size()),
1329             long(inputs0_size), long(inputs1_size),
1330             int(expanded0.size()),
1331             int(expanded1.size()),
1332             long(expanded0_size), long(inputs1_size));
1333         smallest = new_start;
1334         largest = new_limit;
1335         c->inputs_[0] = expanded0;
1336         c->inputs_[1] = expanded1;
1337         GetRange2(c->inputs_[0], c->inputs_[1], &all_start, &all_limit);
1338       }
1339     }
1340   }
1341
1342   // Compute the set of grandparent files that overlap this compaction
1343   // (parent == level+1; grandparent == level+2)
1344   if (level + 2 < config::kNumLevels) {
1345     current_->GetOverlappingInputs(level + 2, &all_start, &all_limit,
1346                                    &c->grandparents_);
1347   }
1348
1349   if (false) {
1350     Log(options_->info_log, "Compacting %d '%s' .. '%s'",
1351         level,
1352         smallest.DebugString().c_str(),
1353         largest.DebugString().c_str());
1354   }
1355
1356   // Update the place where we will do the next compaction for this level.
1357   // We update this immediately instead of waiting for the VersionEdit
1358   // to be applied so that if the compaction fails, we will try a different
1359   // key range next time.
1360   compact_pointer_[level] = largest.Encode().ToString();
1361   c->edit_.SetCompactPointer(level, largest);
1362 }
1363
1364 Compaction* VersionSet::CompactRange(
1365     int level,
1366     const InternalKey* begin,
1367     const InternalKey* end) {
1368   std::vector<FileMetaData*> inputs;
1369   current_->GetOverlappingInputs(level, begin, end, &inputs);
1370   if (inputs.empty()) {
1371     return NULL;
1372   }
1373
1374   // Avoid compacting too much in one shot in case the range is large.
1375   // But we cannot do this for level-0 since level-0 files can overlap
1376   // and we must not pick one file and drop another older file if the
1377   // two files overlap.
1378   if (level > 0) {
1379     const uint64_t limit = MaxFileSizeForLevel(level);
1380     uint64_t total = 0;
1381     for (size_t i = 0; i < inputs.size(); i++) {
1382       uint64_t s = inputs[i]->file_size;
1383       total += s;
1384       if (total >= limit) {
1385         inputs.resize(i + 1);
1386         break;
1387       }
1388     }
1389   }
1390
1391   Compaction* c = new Compaction(level);
1392   c->input_version_ = current_;
1393   c->input_version_->Ref();
1394   c->inputs_[0] = inputs;
1395   SetupOtherInputs(c);
1396   return c;
1397 }
1398
1399 Compaction::Compaction(int level)
1400     : level_(level),
1401       max_output_file_size_(MaxFileSizeForLevel(level)),
1402       input_version_(NULL),
1403       grandparent_index_(0),
1404       seen_key_(false),
1405       overlapped_bytes_(0) {
1406   for (int i = 0; i < config::kNumLevels; i++) {
1407     level_ptrs_[i] = 0;
1408   }
1409 }
1410
1411 Compaction::~Compaction() {
1412   if (input_version_ != NULL) {
1413     input_version_->Unref();
1414   }
1415 }
1416
1417 bool Compaction::IsTrivialMove() const {
1418   // Avoid a move if there is lots of overlapping grandparent data.
1419   // Otherwise, the move could create a parent file that will require
1420   // a very expensive merge later on.
1421   return (num_input_files(0) == 1 &&
1422           num_input_files(1) == 0 &&
1423           TotalFileSize(grandparents_) <= kMaxGrandParentOverlapBytes);
1424 }
1425
1426 void Compaction::AddInputDeletions(VersionEdit* edit) {
1427   for (int which = 0; which < 2; which++) {
1428     for (size_t i = 0; i < inputs_[which].size(); i++) {
1429       edit->DeleteFile(level_ + which, inputs_[which][i]->number);
1430     }
1431   }
1432 }
1433
1434 bool Compaction::IsBaseLevelForKey(const Slice& user_key) {
1435   // Maybe use binary search to find right entry instead of linear search?
1436   const Comparator* user_cmp = input_version_->vset_->icmp_.user_comparator();
1437   for (int lvl = level_ + 2; lvl < config::kNumLevels; lvl++) {
1438     const std::vector<FileMetaData*>& files = input_version_->files_[lvl];
1439     for (; level_ptrs_[lvl] < files.size(); ) {
1440       FileMetaData* f = files[level_ptrs_[lvl]];
1441       if (user_cmp->Compare(user_key, f->largest.user_key()) <= 0) {
1442         // We've advanced far enough
1443         if (user_cmp->Compare(user_key, f->smallest.user_key()) >= 0) {
1444           // Key falls in this file's range, so definitely not base level
1445           return false;
1446         }
1447         break;
1448       }
1449       level_ptrs_[lvl]++;
1450     }
1451   }
1452   return true;
1453 }
1454
1455 bool Compaction::ShouldStopBefore(const Slice& internal_key) {
1456   // Scan to find earliest grandparent file that contains key.
1457   const InternalKeyComparator* icmp = &input_version_->vset_->icmp_;
1458   while (grandparent_index_ < grandparents_.size() &&
1459       icmp->Compare(internal_key,
1460                     grandparents_[grandparent_index_]->largest.Encode()) > 0) {
1461     if (seen_key_) {
1462       overlapped_bytes_ += grandparents_[grandparent_index_]->file_size;
1463     }
1464     grandparent_index_++;
1465   }
1466   seen_key_ = true;
1467
1468   if (overlapped_bytes_ > kMaxGrandParentOverlapBytes) {
1469     // Too much overlap for current output; start new output
1470     overlapped_bytes_ = 0;
1471     return true;
1472   } else {
1473     return false;
1474   }
1475 }
1476
1477 void Compaction::ReleaseInputs() {
1478   if (input_version_ != NULL) {
1479     input_version_->Unref();
1480     input_version_ = NULL;
1481   }
1482 }
1483
1484 }  // namespace leveldb