Some fixes on size compensation logic for deletion entry in compaction

Summary:
This patch include two fixes:
1. newly created Version will now takes the aggregated stats for average-value-size from the latest Version.
2. compensated size of a file is now computed only for newly created / loaded file, this addresses the issue where files are already sorted by their compensated file size but might sometimes observe some out-of-order due to later update on compensated file size.

Test Plan:
export ROCKSDB_TESTS=CompactionDele
./db_test

Reviewers: ljin, igor, sdong

Reviewed By: sdong

Subscribers: leveldb

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

db/compaction_picker.cc

@@ -508,7 +508,7 @@ Compaction* LevelCompactionPicker::PickCompactionBySize(Version* version,
     int index = file_size[i];
     FileMetaData* f = c->input_version_->files_[level][index];
 
-    // check to verify files are arranged in descending size
+    // Check to verify files are arranged in descending compensated size.
     assert((i == file_size.size() - 1) ||
            (i >= Version::number_of_files_to_sort_ - 1) ||
            (f->compensated_file_size >=

db/version_edit.h

@@ -69,8 +69,12 @@ struct FileMetaData {
   // Needs to be disposed when refs becomes 0.
   Cache::Handle* table_reader_handle;
 
-  // stats for compensating deletion entries during compaction
+  // Stats for compensating deletion entries during compaction
-  uint64_t compensated_file_size;  // File size compensated by deletion entry.
+
+  // File size compensated by deletion entry.
+  // This is updated in Version::UpdateTemporaryStats() first time when the
+  // file is created or loaded.  After it is updated, it is immutable.
+  uint64_t compensated_file_size;
   uint64_t num_entries;            // the number of entries.
   uint64_t num_deletions;          // the number of deletion entries.
   uint64_t raw_key_size;           // total uncompressed key size.

db/version_set.cc

@@ -526,6 +526,12 @@ Version::Version(ColumnFamilyData* cfd, VersionSet* vset,
       total_raw_key_size_(0),
       total_raw_value_size_(0),
       num_non_deletions_(0) {
+  if (cfd != nullptr && cfd->current() != nullptr) {
+      total_file_size_ = cfd->current()->total_file_size_;
+      total_raw_key_size_ = cfd->current()->total_raw_key_size_;
+      total_raw_value_size_ = cfd->current()->total_raw_value_size_;
+      num_non_deletions_ = cfd->current()->num_non_deletions_;
+  }
 }
 
 void Version::Get(const ReadOptions& options,
@@ -727,6 +733,7 @@ void Version::Get(const ReadOptions& options,
 }
 
 void Version::PrepareApply(std::vector<uint64_t>& size_being_compacted) {
+  UpdateTemporaryStats();
   ComputeCompactionScore(size_being_compacted);
   UpdateFilesBySize();
   UpdateNumNonEmptyLevels();
@@ -750,7 +757,7 @@ bool Version::MaybeInitializeFileMetaData(FileMetaData* file_meta) {
   return true;
 }
 
-void Version::UpdateTemporaryStats(const VersionEdit* edit) {
+void Version::UpdateTemporaryStats() {
   static const int kDeletionWeightOnCompaction = 2;
 
   // incrementally update the average value size by
@@ -777,12 +784,16 @@ void Version::UpdateTemporaryStats(const VersionEdit* edit) {
   // compute the compensated size
   for (int level = 0; level < num_levels_; level++) {
     for (auto* file_meta : files_[level]) {
+      // Here we only compute compensated_file_size for those file_meta
+      // which compensated_file_size is uninitialized (== 0).
+      if (file_meta->compensated_file_size == 0) {
         file_meta->compensated_file_size = file_meta->fd.GetFileSize() +
             file_meta->num_deletions * average_value_size *
             kDeletionWeightOnCompaction;
       }
     }
   }
+}
 
 void Version::ComputeCompactionScore(
     std::vector<uint64_t>& size_being_compacted) {
@@ -1740,7 +1751,6 @@ Status VersionSet::LogAndApply(ColumnFamilyData* column_family_data,
 
     if (!edit->IsColumnFamilyManipulation()) {
       // This is cpu-heavy operations, which should be called outside mutex.
-      v->UpdateTemporaryStats(edit);
       v->PrepareApply(size_being_compacted);
     }
 

db/version_set.h

@@ -254,7 +254,7 @@ class Version {
 
   // Update the temporary stats associated with the current version.
   // This temporary stats will be used in compaction.
-  void UpdateTemporaryStats(const VersionEdit* edit);
+  void UpdateTemporaryStats();
 
   // Sort all files for this version based on their file size and
   // record results in files_by_size_. The largest files are listed first.