Fix level size overflow for options_.level_compaction_dynamic_level_bytes=true

Summary: Int is used for level size targets when options_.level_compaction_dynamic_level_bytes=true, which will cause overflow when database grows big. Fix it.

Test Plan: Add a new unit test which fails without the fix.

Reviewers: rven, yhchiang, MarkCallaghan, igor

Reviewed By: igor

Subscribers: leveldb, dhruba, yoshinorim

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

db/version_set.cc

@@ -1604,12 +1604,12 @@ void VersionStorageInfo::CalculateBaseBytes(const ImmutableCFOptions& ioptions,
       }
 
       // Calculate base level and its size.
-      int base_level_size;
+      uint64_t base_level_size;
       if (cur_level_size <= base_bytes_min) {
         // Case 1. If we make target size of last level to be max_level_size,
         // target size of the first non-empty level would be smaller than
         // base_bytes_min. We set it be base_bytes_min.
-        base_level_size = static_cast<int>(base_bytes_min + 1);
+        base_level_size = base_bytes_min + 1U;
         base_level_ = first_non_empty_level;
         Warn(ioptions.info_log,
              "More existing levels in DB than needed. "
@@ -1625,16 +1625,17 @@ void VersionStorageInfo::CalculateBaseBytes(const ImmutableCFOptions& ioptions,
         if (cur_level_size > base_bytes_max) {
           // Even L1 will be too large
           assert(base_level_ == 1);
-          base_level_size = static_cast<int>(base_bytes_max);
+          base_level_size = base_bytes_max;
         } else {
-          base_level_size = static_cast<int>(cur_level_size);
+          base_level_size = cur_level_size;
         }
       }
 
-      int level_size = base_level_size;
+      uint64_t level_size = base_level_size;
       for (int i = base_level_; i < num_levels_; i++) {
         if (i > base_level_) {
-          level_size = level_size * options.max_bytes_for_level_multiplier;
+          level_size = MultiplyCheckOverflow(
+              level_size, options.max_bytes_for_level_multiplier);
         }
         level_max_bytes_[i] = level_size;
       }

db/version_set_test.cc

@@ -215,6 +215,25 @@ TEST_F(VersionStorageInfoTest, MaxBytesForLevelDynamicLotsOfData) {
   ASSERT_EQ(0, logger_->log_count);
 }
 
+TEST_F(VersionStorageInfoTest, MaxBytesForLevelDynamicLargeLevel) {
+  uint64_t kOneGB = 1000U * 1000U * 1000U;
+  ioptions_.level_compaction_dynamic_level_bytes = true;
+  mutable_cf_options_.max_bytes_for_level_base = 10U * kOneGB;
+  mutable_cf_options_.max_bytes_for_level_multiplier = 10;
+  Add(0, 1U, "1", "2", 50U);
+  Add(3, 4U, "1", "2", 32U * kOneGB);
+  Add(4, 5U, "1", "2", 500U * kOneGB);
+  Add(5, 6U, "1", "2", 3000U * kOneGB);
+
+  vstorage_.CalculateBaseBytes(ioptions_, mutable_cf_options_);
+  ASSERT_EQ(vstorage_.MaxBytesForLevel(5), 3000U * kOneGB);
+  ASSERT_EQ(vstorage_.MaxBytesForLevel(4), 300U * kOneGB);
+  ASSERT_EQ(vstorage_.MaxBytesForLevel(3), 30U * kOneGB);
+  ASSERT_EQ(vstorage_.MaxBytesForLevel(2), 3U * kOneGB);
+  ASSERT_EQ(vstorage_.base_level(), 2);
+  ASSERT_EQ(0, logger_->log_count);
+}
+
 class FindLevelFileTest : public testing::Test {
  public:
   LevelFilesBrief file_level_;