YugabyteDB (2.13.0.0-b42, bfc6a6643e7399ac8a0e81d06a3ee6d6571b33ab)

//  Copyright (c) 2011-present, Facebook, Inc.  All rights reserved.

//  This source code is licensed under the BSD-style license found in the

//  LICENSE file in the root directory of this source tree. An additional grant

//  of patent rights can be found in the PATENTS file in the same directory.

//

// The following only applies to changes made to this file as part of YugaByte development.

//

// Portions Copyright (c) YugaByte, Inc.

//

// Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except

// in compliance with the License.  You may obtain a copy of the License at

//

// http://www.apache.org/licenses/LICENSE-2.0

//

// Unless required by applicable law or agreed to in writing, software distributed under the License

// is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express

// or implied.  See the License for the specific language governing permissions and limitations

// under the License.

//

#include <limits>

#include <string>

#include <utility>

#include <gtest/gtest.h>

#include "yb/rocksdb/db/compaction_picker.h"

#include "yb/rocksdb/db/version_set.h"

#include "yb/rocksdb/env.h"

#include "yb/util/size_literals.h"

#include "yb/util/string_util.h"

#include "yb/rocksdb/util/testutil.h"

using namespace yb::size_literals;

namespace rocksdb {

class CountingLogger : public Logger {

 public:

  using Logger::Logv;

  void Logv(const char* format, va_list ap) override { log_count++; }

  size_t log_count;

};

class CompactionPickerTest : public RocksDBTest {

 public:

  const Comparator* ucmp_;

  InternalKeyComparatorPtr icmp_;

  Options options_;

  ImmutableCFOptions ioptions_;

  MutableCFOptions mutable_cf_options_;

  LevelCompactionPicker level_compaction_picker;

  std::string cf_name_;

  CountingLogger logger_;

  LogBuffer log_buffer_;

  uint32_t file_num_;

  CompactionOptionsFIFO fifo_options_;

  std::unique_ptr<VersionStorageInfo> vstorage_;

  std::vector<std::unique_ptr<FileMetaData>> files_;

  // does not own FileMetaData

  std::unordered_map<uint32_t, std::pair<FileMetaData*, int>> file_map_;

  // input files to compaction process.

  std::vector<CompactionInputFiles> input_files_;

  int compaction_level_start_;

  CompactionPickerTest()

      : ucmp_(BytewiseComparator()),

        icmp_(std::make_shared<InternalKeyComparator>(ucmp_)),

        ioptions_(options_),

        mutable_cf_options_(options_, ioptions_),

        level_compaction_picker(ioptions_, icmp_.get()),

        cf_name_("dummy"),

        log_buffer_(InfoLogLevel::INFO_LEVEL, &logger_),

        file_num_(1),

        vstorage_(nullptr) {

    fifo_options_.max_table_files_size = 1;

    mutable_cf_options_.RefreshDerivedOptions(ioptions_);

    ioptions_.db_paths.emplace_back("dummy",

                                    std::numeric_limits<uint64_t>::max());

  }

  ~CompactionPickerTest() {

    log_buffer_.FlushBufferToLog();

  }

  void NewVersionStorage(int num_levels, CompactionStyle style) {

    DeleteVersionStorage();

    options_.num_levels = num_levels;

    vstorage_.reset(new VersionStorageInfo(

        icmp_, ucmp_, options_.num_levels, style, nullptr));

    vstorage_->CalculateBaseBytes(ioptions_, mutable_cf_options_);

  }

  void DeleteVersionStorage() {

    vstorage_.reset();

    files_.clear();

    file_map_.clear();

    input_files_.clear();

  }

  void Add(int level, uint32_t file_number, const char* smallest,

           const char* largest, uint64_t file_size = 0, uint32_t path_id = 0,

           SequenceNumber smallest_seq = 100,

           SequenceNumber largest_seq = 100) {

    assert(level < vstorage_->num_levels());

    FileMetaData* f = new FileMetaData;

    // For large SST use 512 bytes as base file size, for files smaller than 512 bytes use half of

    // total size as a base file size.

    // Some compaction picker tests are using small files (<512 and even 1-3) bytes, so for these

    // files we split them into (file_size / 2) bytes base file and the rest goes to data file.

    const uint64_t base_file_size = file_size > 512 ? 512 : (file_size / 2);

    f->fd = FileDescriptor(file_number, path_id, file_size, base_file_size);

    f->smallest = MakeFileBoundaryValues(smallest, smallest_seq, kTypeValue);

    f->largest = MakeFileBoundaryValues(largest, largest_seq, kTypeValue);

    f->compensated_file_size = file_size;

    f->refs = 0;

    vstorage_->AddFile(level, f);

    files_.emplace_back(f);

    file_map_.insert({file_number, {f, level}});

  }

  void SetCompactionInputFilesLevels(int level_count, int start_level) {

    input_files_.resize(level_count);

    for (int i = 0; i < level_count; ++i) {

      input_files_[i].level = start_level + i;

    }

    compaction_level_start_ = start_level;

  }

  void AddToCompactionFiles(uint32_t file_number) {

    auto iter = file_map_.find(file_number);

    assert(iter != file_map_.end());

    int level = iter->second.second;

    assert(level < vstorage_->num_levels());

    input_files_[level - compaction_level_start_].files.emplace_back(

        iter->second.first);

  }

  void UpdateVersionStorageInfo() {

    vstorage_->CalculateBaseBytes(ioptions_, mutable_cf_options_);

    vstorage_->UpdateFilesByCompactionPri(mutable_cf_options_);

    vstorage_->UpdateNumNonEmptyLevels();

    vstorage_->GenerateFileIndexer();

    vstorage_->GenerateLevelFilesBrief();

    vstorage_->ComputeCompactionScore(mutable_cf_options_, fifo_options_);

    vstorage_->GenerateLevel0NonOverlapping();

    vstorage_->SetFinalized();

  }

};

TEST_F(CompactionPickerTest, Empty) {

  NewVersionStorage(6, kCompactionStyleLevel);

  UpdateVersionStorageInfo();

  std::unique_ptr<Compaction> compaction(level_compaction_picker.PickCompaction(

      cf_name_, mutable_cf_options_, vstorage_.get(), &log_buffer_));

  ASSERT_TRUE(compaction.get() == nullptr);

}

TEST_F(CompactionPickerTest, Single) {

  NewVersionStorage(6, kCompactionStyleLevel);

  mutable_cf_options_.level0_file_num_compaction_trigger = 2;

  Add(0, 1U, "p", "q");

  UpdateVersionStorageInfo();

  std::unique_ptr<Compaction> compaction(level_compaction_picker.PickCompaction(

      cf_name_, mutable_cf_options_, vstorage_.get(), &log_buffer_));

  ASSERT_TRUE(compaction.get() == nullptr);

}

TEST_F(CompactionPickerTest, Level0Trigger) {

  NewVersionStorage(6, kCompactionStyleLevel);

  mutable_cf_options_.level0_file_num_compaction_trigger = 2;

  Add(0, 1U, "150", "200");

  Add(0, 2U, "200", "250");

  UpdateVersionStorageInfo();

  std::unique_ptr<Compaction> compaction(level_compaction_picker.PickCompaction(

      cf_name_, mutable_cf_options_, vstorage_.get(), &log_buffer_));

  ASSERT_TRUE(compaction.get() != nullptr);

  ASSERT_EQ(2U, compaction->num_input_files(0));

  ASSERT_EQ(1U, compaction->input(0, 0)->fd.GetNumber());

  ASSERT_EQ(2U, compaction->input(0, 1)->fd.GetNumber());

}

TEST_F(CompactionPickerTest, Level1Trigger) {

  NewVersionStorage(6, kCompactionStyleLevel);

  Add(1, 66U, "150", "200", 1000000000U);

  UpdateVersionStorageInfo();

  std::unique_ptr<Compaction> compaction(level_compaction_picker.PickCompaction(

      cf_name_, mutable_cf_options_, vstorage_.get(), &log_buffer_));

  ASSERT_TRUE(compaction.get() != nullptr);

  ASSERT_EQ(1U, compaction->num_input_files(0));

  ASSERT_EQ(66U, compaction->input(0, 0)->fd.GetNumber());

}

TEST_F(CompactionPickerTest, Level1Trigger2) {

  NewVersionStorage(6, kCompactionStyleLevel);

  Add(1, 66U, "150", "200", 1000000001U);

  Add(1, 88U, "201", "300", 1000000000U);

  Add(2, 6U, "150", "179", 1000000000U);

  Add(2, 7U, "180", "220", 1000000000U);

  Add(2, 8U, "221", "300", 1000000000U);

  UpdateVersionStorageInfo();

  std::unique_ptr<Compaction> compaction(level_compaction_picker.PickCompaction(

      cf_name_, mutable_cf_options_, vstorage_.get(), &log_buffer_));

  ASSERT_TRUE(compaction.get() != nullptr);

  ASSERT_EQ(1U, compaction->num_input_files(0));

  ASSERT_EQ(2U, compaction->num_input_files(1));

  ASSERT_EQ(66U, compaction->input(0, 0)->fd.GetNumber());

  ASSERT_EQ(6U, compaction->input(1, 0)->fd.GetNumber());

  ASSERT_EQ(7U, compaction->input(1, 1)->fd.GetNumber());

}

TEST_F(CompactionPickerTest, LevelMaxScore) {

  NewVersionStorage(6, kCompactionStyleLevel);

  mutable_cf_options_.target_file_size_base = 10000000;

  mutable_cf_options_.target_file_size_multiplier = 10;

  Add(0, 1U, "150", "200", 1000000000U);

  // Level 1 score 1.2

  Add(1, 66U, "150", "200", 6000000U);

  Add(1, 88U, "201", "300", 6000000U);

  // Level 2 score 1.8. File 7 is the largest. Should be picked

  Add(2, 6U, "150", "179", 60000000U);

  Add(2, 7U, "180", "220", 60000001U);

  Add(2, 8U, "221", "300", 60000000U);

  // Level 3 score slightly larger than 1

  Add(3, 26U, "150", "170", 260000000U);

  Add(3, 27U, "171", "179", 260000000U);

  Add(3, 28U, "191", "220", 260000000U);

  Add(3, 29U, "221", "300", 260000000U);

  UpdateVersionStorageInfo();

  std::unique_ptr<Compaction> compaction(level_compaction_picker.PickCompaction(

      cf_name_, mutable_cf_options_, vstorage_.get(), &log_buffer_));

  ASSERT_TRUE(compaction.get() != nullptr);

  ASSERT_EQ(1U, compaction->num_input_files(0));

  ASSERT_EQ(7U, compaction->input(0, 0)->fd.GetNumber());

}

TEST_F(CompactionPickerTest, NeedsCompactionLevel) {

  const int kLevels = 6;

  const int kFileCount = 20;

  for (int level = 0; level < kLevels - 1; ++level) {

    NewVersionStorage(kLevels, kCompactionStyleLevel);

    uint64_t file_size = vstorage_->MaxBytesForLevel(level) * 2 / kFileCount;

    for (int file_count = 1; file_count <= kFileCount; ++file_count) {

      // start a brand new version in each test.

      NewVersionStorage(kLevels, kCompactionStyleLevel);

      for (int i = 0; i < file_count; ++i) {

        Add(level, i, ToString((i + 100) * 1000).c_str(),

            ToString((i + 100) * 1000 + 999).c_str(),

            file_size, 0, i * 100, i * 100 + 99);

      }

      UpdateVersionStorageInfo();

      ASSERT_EQ(vstorage_->CompactionScoreLevel(0), level);

      ASSERT_EQ(level_compaction_picker.NeedsCompaction(vstorage_.get()),

                vstorage_->CompactionScore(0) >= 1);

      // release the version storage

      DeleteVersionStorage();

    }

  }

}

TEST_F(CompactionPickerTest, Level0TriggerDynamic) {

  int num_levels = ioptions_.num_levels;

  ioptions_.level_compaction_dynamic_level_bytes = true;

  mutable_cf_options_.level0_file_num_compaction_trigger = 2;

  mutable_cf_options_.max_bytes_for_level_base = 200;

  mutable_cf_options_.max_bytes_for_level_multiplier = 10;

  NewVersionStorage(num_levels, kCompactionStyleLevel);

  Add(0, 1U, "150", "200");

  Add(0, 2U, "200", "250");

  UpdateVersionStorageInfo();

  std::unique_ptr<Compaction> compaction(level_compaction_picker.PickCompaction(

      cf_name_, mutable_cf_options_, vstorage_.get(), &log_buffer_));

  ASSERT_TRUE(compaction.get() != nullptr);

  ASSERT_EQ(2U, compaction->num_input_files(0));

  ASSERT_EQ(1U, compaction->input(0, 0)->fd.GetNumber());

  ASSERT_EQ(2U, compaction->input(0, 1)->fd.GetNumber());

  ASSERT_EQ(1, static_cast<int>(compaction->num_input_levels()));

  ASSERT_EQ(num_levels - 1, compaction->output_level());

}

TEST_F(CompactionPickerTest, Level0TriggerDynamic2) {

  int num_levels = ioptions_.num_levels;

  ioptions_.level_compaction_dynamic_level_bytes = true;

  mutable_cf_options_.level0_file_num_compaction_trigger = 2;

  mutable_cf_options_.max_bytes_for_level_base = 200;

  mutable_cf_options_.max_bytes_for_level_multiplier = 10;

  NewVersionStorage(num_levels, kCompactionStyleLevel);

  Add(0, 1U, "150", "200");

  Add(0, 2U, "200", "250");

  Add(num_levels - 1, 3U, "200", "250", 300U);

  UpdateVersionStorageInfo();

  ASSERT_EQ(vstorage_->base_level(), num_levels - 2);

  std::unique_ptr<Compaction> compaction(level_compaction_picker.PickCompaction(

      cf_name_, mutable_cf_options_, vstorage_.get(), &log_buffer_));

  ASSERT_TRUE(compaction.get() != nullptr);

  ASSERT_EQ(2U, compaction->num_input_files(0));

  ASSERT_EQ(1U, compaction->input(0, 0)->fd.GetNumber());

  ASSERT_EQ(2U, compaction->input(0, 1)->fd.GetNumber());

  ASSERT_EQ(1, static_cast<int>(compaction->num_input_levels()));

  ASSERT_EQ(num_levels - 2, compaction->output_level());

}

TEST_F(CompactionPickerTest, Level0TriggerDynamic3) {

  int num_levels = ioptions_.num_levels;

  ioptions_.level_compaction_dynamic_level_bytes = true;

  mutable_cf_options_.level0_file_num_compaction_trigger = 2;

  mutable_cf_options_.max_bytes_for_level_base = 200;

  mutable_cf_options_.max_bytes_for_level_multiplier = 10;

  NewVersionStorage(num_levels, kCompactionStyleLevel);

  Add(0, 1U, "150", "200");

  Add(0, 2U, "200", "250");

  Add(num_levels - 1, 3U, "200", "250", 300U);

  Add(num_levels - 1, 4U, "300", "350", 3000U);

  UpdateVersionStorageInfo();

  ASSERT_EQ(vstorage_->base_level(), num_levels - 3);

  std::unique_ptr<Compaction> compaction(level_compaction_picker.PickCompaction(

      cf_name_, mutable_cf_options_, vstorage_.get(), &log_buffer_));

  ASSERT_TRUE(compaction.get() != nullptr);

  ASSERT_EQ(2U, compaction->num_input_files(0));

  ASSERT_EQ(1U, compaction->input(0, 0)->fd.GetNumber());

  ASSERT_EQ(2U, compaction->input(0, 1)->fd.GetNumber());

  ASSERT_EQ(1, static_cast<int>(compaction->num_input_levels()));

  ASSERT_EQ(num_levels - 3, compaction->output_level());

}

TEST_F(CompactionPickerTest, Level0TriggerDynamic4) {

  int num_levels = ioptions_.num_levels;

  ioptions_.level_compaction_dynamic_level_bytes = true;

  mutable_cf_options_.level0_file_num_compaction_trigger = 2;

  mutable_cf_options_.max_bytes_for_level_base = 200;

  mutable_cf_options_.max_bytes_for_level_multiplier = 10;

  NewVersionStorage(num_levels, kCompactionStyleLevel);

  Add(0, 1U, "150", "200");

  Add(0, 2U, "200", "250");

  Add(num_levels - 1, 3U, "200", "250", 300U);

  Add(num_levels - 1, 4U, "300", "350", 3000U);

  Add(num_levels - 3, 5U, "150", "180", 3U);

  Add(num_levels - 3, 6U, "181", "300", 3U);

  Add(num_levels - 3, 7U, "400", "450", 3U);

  UpdateVersionStorageInfo();

  ASSERT_EQ(vstorage_->base_level(), num_levels - 3);

  std::unique_ptr<Compaction> compaction(level_compaction_picker.PickCompaction(

      cf_name_, mutable_cf_options_, vstorage_.get(), &log_buffer_));

  ASSERT_TRUE(compaction.get() != nullptr);

  ASSERT_EQ(2U, compaction->num_input_files(0));

  ASSERT_EQ(1U, compaction->input(0, 0)->fd.GetNumber());

  ASSERT_EQ(2U, compaction->input(0, 1)->fd.GetNumber());

  ASSERT_EQ(2U, compaction->num_input_files(1));

  ASSERT_EQ(num_levels - 3, compaction->level(1));

  ASSERT_EQ(5U, compaction->input(1, 0)->fd.GetNumber());

  ASSERT_EQ(6U, compaction->input(1, 1)->fd.GetNumber());

  ASSERT_EQ(2, static_cast<int>(compaction->num_input_levels()));

  ASSERT_EQ(num_levels - 3, compaction->output_level());

}

TEST_F(CompactionPickerTest, LevelTriggerDynamic4) {

  int num_levels = ioptions_.num_levels;

  ioptions_.level_compaction_dynamic_level_bytes = true;

  mutable_cf_options_.level0_file_num_compaction_trigger = 2;

  mutable_cf_options_.max_bytes_for_level_base = 200;

  mutable_cf_options_.max_bytes_for_level_multiplier = 10;

  NewVersionStorage(num_levels, kCompactionStyleLevel);

  Add(0, 1U, "150", "200");

  Add(num_levels - 1, 3U, "200", "250", 300U);

  Add(num_levels - 1, 4U, "300", "350", 3000U);

  Add(num_levels - 1, 4U, "400", "450", 3U);

  Add(num_levels - 2, 5U, "150", "180", 300U);

  Add(num_levels - 2, 6U, "181", "350", 500U);

  Add(num_levels - 2, 7U, "400", "450", 200U);

  UpdateVersionStorageInfo();

  std::unique_ptr<Compaction> compaction(level_compaction_picker.PickCompaction(

      cf_name_, mutable_cf_options_, vstorage_.get(), &log_buffer_));

  ASSERT_TRUE(compaction.get() != nullptr);

  ASSERT_EQ(1U, compaction->num_input_files(0));

  ASSERT_EQ(6U, compaction->input(0, 0)->fd.GetNumber());

  ASSERT_EQ(2U, compaction->num_input_files(1));

  ASSERT_EQ(3U, compaction->input(1, 0)->fd.GetNumber());

  ASSERT_EQ(4U, compaction->input(1, 1)->fd.GetNumber());

  ASSERT_EQ(2U, compaction->num_input_levels());

  ASSERT_EQ(num_levels - 1, compaction->output_level());

}

// Universal and FIFO Compactions are not supported in ROCKSDB_LITE

#ifndef ROCKSDB_LITE

TEST_F(CompactionPickerTest, NeedsCompactionUniversal) {

  NewVersionStorage(1, kCompactionStyleUniversal);

  UniversalCompactionPicker universal_compaction_picker(

      ioptions_, icmp_.get());

  // must return false when there's no files.

  ASSERT_EQ(universal_compaction_picker.NeedsCompaction(vstorage_.get()),

            false);

  UpdateVersionStorageInfo();

  // verify the trigger given different number of L0 files.

  for (int i = 1;

       i <= mutable_cf_options_.level0_file_num_compaction_trigger * 2; ++i) {

    NewVersionStorage(1, kCompactionStyleUniversal);

    Add(0, i, ToString((i + 100) * 1000).c_str(),

        ToString((i + 100) * 1000 + 999).c_str(), 1000000, 0, i * 100,

        i * 100 + 99);

    UpdateVersionStorageInfo();

    ASSERT_EQ(level_compaction_picker.NeedsCompaction(vstorage_.get()),

              vstorage_->CompactionScore(0) >= 1);

  }

}

// Tests if the correct compaction reason is used given which files are marked

// as being compacted. Specifically, the test ensures that if the earliest file

// available in a sequence of sorted runs is currently being compacted, it is

// not included in the compaction.

TEST_F(CompactionPickerTest, CompactionUniversalPickForFilesBeingCompacted) {

  NewVersionStorage(1, kCompactionStyleUniversal);

  ioptions_.compaction_style = kCompactionStyleUniversal;

  ioptions_.num_levels = 1;

  mutable_cf_options_.write_buffer_size = 100_KB;

  mutable_cf_options_.target_file_size_base = 32_KB;

  mutable_cf_options_.level0_file_num_compaction_trigger = 3;

  UniversalCompactionPicker universal_compaction_picker(ioptions_, icmp_.get());

  // must return false when there's no files.

  ASSERT_FALSE(universal_compaction_picker.NeedsCompaction(vstorage_.get()));

  UpdateVersionStorageInfo();

  NewVersionStorage(1, kCompactionStyleUniversal);

  // Create three files, the earliest of which is significantly smaller than the other

  // two. This should cause the compaction picker to include all three files,

  // choosing "size amplification" as the compaction reason.

  // Add third file

  Add(/* level = */ 0, /* file number = */ 3, /* smallest = */ "300000",

      /* largest = */ "300999", /* file_size = */ 1000000, /* path_id = */ 0,

      /* smallest seq num = */ 300, /* largest seq num = */ 399);

  // Add second file

  Add(/* level = */ 0, /* file number = */ 2, /* smallest = */ "200000",

      /* largest = */ "200999", /* file size = */ 1000000, /* path_id = */ 0,

      /* smallest seq num = */ 200, /* largest seq num = */ 299);

  // Add first file

  Add(/* level = */ 0, /* file number = */ 1, /* smallest = */ "100000",

      /* largest = */ "100999", /* file size (smaller) = */ 10,

      /* path_id = */ 0, /* smallest seq num = */ 100, /* largest seq num = */ 199);

  UpdateVersionStorageInfo();

  auto compaction = universal_compaction_picker.PickCompaction(

      cf_name_, mutable_cf_options_, vstorage_.get(), &log_buffer_);

  ASSERT_NE(compaction, nullptr);

  ASSERT_EQ(compaction->compaction_reason(), CompactionReason::kUniversalSizeAmplification);

  // Verify that all three files are now being compacted.

  for (int i = 1; i <= 3; i++) {

    ASSERT_TRUE(file_map_.at(i).first->being_compacted);

  }

  // Reset "being_compacted" on files 2 and 3 (simulates if earliest file is being compacted).

  // This scenario is particularly likely if file expiration based on TTL is enabled.

  file_map_.at(2).first->being_compacted = false;

  file_map_.at(3).first->being_compacted = false;

  // Files in sequence are [3 (available), 2 (available), 1 (being compacted)].

  // The compaction picker should still create a compaction, but it will only include files

  // 3 and 2, and will be due to file ratio/sorted run num.

  compaction = universal_compaction_picker.PickCompaction(

      cf_name_, mutable_cf_options_, vstorage_.get(), &log_buffer_);

  ASSERT_NE(compaction, nullptr);

  ASSERT_EQ(compaction->compaction_reason(), CompactionReason::kUniversalSortedRunNum);

  // Verify there are exactly 2 files being compacted.

  ASSERT_EQ(compaction->inputs(0)->size(), 2);

}

// Tests if the files can be trivially moved in multi level

// universal compaction when allow_trivial_move option is set

// In this test as the input files overlaps, they cannot

// be trivially moved.

TEST_F(CompactionPickerTest, CannotTrivialMoveUniversal) {

  const uint64_t kFileSize = 100000;

  ioptions_.compaction_options_universal.allow_trivial_move = true;

  NewVersionStorage(1, kCompactionStyleUniversal);

  UniversalCompactionPicker universal_compaction_picker(ioptions_, icmp_.get());

  // must return false when there's no files.

  ASSERT_EQ(universal_compaction_picker.NeedsCompaction(vstorage_.get()),

            false);

  NewVersionStorage(3, kCompactionStyleUniversal);

  Add(0, 1U, "150", "200", kFileSize, 0, 500, 550);

  Add(0, 2U, "201", "250", kFileSize, 0, 401, 450);

  Add(0, 4U, "260", "300", kFileSize, 0, 260, 300);

  Add(1, 5U, "100", "151", kFileSize, 0, 200, 251);

  Add(1, 3U, "301", "350", kFileSize, 0, 101, 150);

  Add(2, 6U, "120", "200", kFileSize, 0, 20, 100);

  UpdateVersionStorageInfo();

  std::unique_ptr<Compaction> compaction(

      universal_compaction_picker.PickCompaction(

          cf_name_, mutable_cf_options_, vstorage_.get(), &log_buffer_));

  ASSERT_TRUE(!compaction->is_trivial_move());

}

// Tests if the files can be trivially moved in multi level

// universal compaction when allow_trivial_move option is set

// In this test as the input files doesn't overlaps, they should

// be trivially moved.

TEST_F(CompactionPickerTest, AllowsTrivialMoveUniversal) {

  const uint64_t kFileSize = 100000;

  ioptions_.compaction_options_universal.allow_trivial_move = true;

  UniversalCompactionPicker universal_compaction_picker(ioptions_, icmp_.get());

  NewVersionStorage(3, kCompactionStyleUniversal);

  Add(0, 1U, "150", "200", kFileSize, 0, 500, 550);

  Add(0, 2U, "201", "250", kFileSize, 0, 401, 450);

  Add(0, 4U, "260", "300", kFileSize, 0, 260, 300);

  Add(1, 5U, "010", "080", kFileSize, 0, 200, 251);

  Add(2, 3U, "301", "350", kFileSize, 0, 101, 150);

  UpdateVersionStorageInfo();

  std::unique_ptr<Compaction> compaction(

      universal_compaction_picker.PickCompaction(

          cf_name_, mutable_cf_options_, vstorage_.get(), &log_buffer_));

  ASSERT_TRUE(compaction->is_trivial_move());

}

TEST_F(CompactionPickerTest, NeedsCompactionFIFO) {

  NewVersionStorage(1, kCompactionStyleFIFO);

  const int kFileCount =

      mutable_cf_options_.level0_file_num_compaction_trigger * 3;

  const uint64_t kFileSize = 100000;

  const uint64_t kMaxSize = kFileSize * kFileCount / 2;

  fifo_options_.max_table_files_size = kMaxSize;

  ioptions_.compaction_options_fifo = fifo_options_;

  FIFOCompactionPicker fifo_compaction_picker(ioptions_, icmp_.get());

  UpdateVersionStorageInfo();

  // must return false when there's no files.

  ASSERT_EQ(fifo_compaction_picker.NeedsCompaction(vstorage_.get()), false);

  // verify whether compaction is needed based on the current

  // size of L0 files.

  uint64_t current_size = 0;

  for (int i = 1; i <= kFileCount; ++i) {

    NewVersionStorage(1, kCompactionStyleFIFO);

    Add(0, i, ToString((i + 100) * 1000).c_str(),

        ToString((i + 100) * 1000 + 999).c_str(),

        kFileSize, 0, i * 100, i * 100 + 99);

    current_size += kFileSize;

    UpdateVersionStorageInfo();

    ASSERT_EQ(level_compaction_picker.NeedsCompaction(vstorage_.get()),

              vstorage_->CompactionScore(0) >= 1);

  }

}

#endif  // ROCKSDB_LITE

TEST_F(CompactionPickerTest, CompactionPriMinOverlapping1) {

  NewVersionStorage(6, kCompactionStyleLevel);

  mutable_cf_options_.target_file_size_base = 10000000;

  mutable_cf_options_.target_file_size_multiplier = 10;

  mutable_cf_options_.compaction_pri = kMinOverlappingRatio;

  Add(2, 6U, "150", "179", 50000000U);

  Add(2, 7U, "180", "220", 50000000U);

  Add(2, 8U, "321", "400", 50000000U);  // File not overlapping

  Add(2, 9U, "721", "800", 50000000U);

  Add(3, 26U, "150", "170", 260000000U);

  Add(3, 27U, "171", "179", 260000000U);

  Add(3, 28U, "191", "220", 260000000U);

  Add(3, 29U, "221", "300", 260000000U);

  Add(3, 30U, "750", "900", 260000000U);

  UpdateVersionStorageInfo();

  std::unique_ptr<Compaction> compaction(level_compaction_picker.PickCompaction(

      cf_name_, mutable_cf_options_, vstorage_.get(), &log_buffer_));

  ASSERT_TRUE(compaction.get() != nullptr);

  ASSERT_EQ(1U, compaction->num_input_files(0));

  // Pick file 8 because it overlaps with 0 files on level 3.

  ASSERT_EQ(8U, compaction->input(0, 0)->fd.GetNumber());

}

TEST_F(CompactionPickerTest, CompactionPriMinOverlapping2) {

  NewVersionStorage(6, kCompactionStyleLevel);

  mutable_cf_options_.target_file_size_base = 10000000;

  mutable_cf_options_.target_file_size_multiplier = 10;

  mutable_cf_options_.compaction_pri = kMinOverlappingRatio;

  Add(2, 6U, "150", "175",

      60000000U);  // Overlaps with file 26, 27, total size 521M

  Add(2, 7U, "176", "200", 60000000U);  // Overlaps with file 27, 28, total size

                                        // 520M, the smalelst overlapping

  Add(2, 8U, "201", "300",

      60000000U);  // Overlaps with file 28, 29, total size 521M

  Add(3, 26U, "100", "110", 261000000U);

  Add(3, 26U, "150", "170", 261000000U);

  Add(3, 27U, "171", "179", 260000000U);

  Add(3, 28U, "191", "220", 260000000U);

  Add(3, 29U, "221", "300", 261000000U);

  Add(3, 30U, "321", "400", 261000000U);

  UpdateVersionStorageInfo();

  std::unique_ptr<Compaction> compaction(level_compaction_picker.PickCompaction(

      cf_name_, mutable_cf_options_, vstorage_.get(), &log_buffer_));

  ASSERT_TRUE(compaction.get() != nullptr);

  ASSERT_EQ(1U, compaction->num_input_files(0));

  // Picking file 7 because overlapping ratio is the biggest.

  ASSERT_EQ(7U, compaction->input(0, 0)->fd.GetNumber());

}

TEST_F(CompactionPickerTest, CompactionPriMinOverlapping3) {

  NewVersionStorage(6, kCompactionStyleLevel);

  mutable_cf_options_.target_file_size_base = 10000000;

  mutable_cf_options_.target_file_size_multiplier = 10;

  mutable_cf_options_.compaction_pri = kMinOverlappingRatio;

  // file 7 and 8 over lap with the same file, but file 8 is smaller so

  // it will be picked.

  Add(2, 6U, "150", "175", 60000000U);  // Overlaps with file 26, 27

  Add(2, 7U, "176", "200", 60000000U);  // Overlaps with file 27

  Add(2, 8U, "201", "300", 61000000U);  // Overlaps with file 27

  Add(3, 26U, "160", "165", 260000000U);

  Add(3, 26U, "166", "170", 260000000U);

  Add(3, 27U, "180", "400", 260000000U);

  Add(3, 28U, "401", "500", 260000000U);

  UpdateVersionStorageInfo();

  std::unique_ptr<Compaction> compaction(level_compaction_picker.PickCompaction(

      cf_name_, mutable_cf_options_, vstorage_.get(), &log_buffer_));

  ASSERT_TRUE(compaction.get() != nullptr);

  ASSERT_EQ(1U, compaction->num_input_files(0));

  // Picking file 8 because overlapping ratio is the biggest.

  ASSERT_EQ(8U, compaction->input(0, 0)->fd.GetNumber());

}

// This test exhibits the bug where we don't properly reset parent_index in

// PickCompaction()

TEST_F(CompactionPickerTest, ParentIndexResetBug) {

  int num_levels = ioptions_.num_levels;

  mutable_cf_options_.level0_file_num_compaction_trigger = 2;

  mutable_cf_options_.max_bytes_for_level_base = 200;

  NewVersionStorage(num_levels, kCompactionStyleLevel);

  Add(0, 1U, "150", "200");       // <- marked for compaction

  Add(1, 3U, "400", "500", 600);  // <- this one needs compacting

  Add(2, 4U, "150", "200");

  Add(2, 5U, "201", "210");

  Add(2, 6U, "300", "310");

  Add(2, 7U, "400", "500");  // <- being compacted

  vstorage_->LevelFiles(2)[3]->being_compacted = true;

  vstorage_->LevelFiles(0)[0]->marked_for_compaction = true;

  UpdateVersionStorageInfo();

  std::unique_ptr<Compaction> compaction(level_compaction_picker.PickCompaction(

      cf_name_, mutable_cf_options_, vstorage_.get(), &log_buffer_));

}

// This test checks ExpandWhileOverlapping() by having overlapping user keys

// ranges (with different sequence numbers) in the input files.

TEST_F(CompactionPickerTest, OverlappingUserKeys) {

  NewVersionStorage(6, kCompactionStyleLevel);

  Add(1, 1U, "100", "150", 1U);

  // Overlapping user keys

  Add(1, 2U, "200", "400", 1U);

  Add(1, 3U, "400", "500", 1000000000U, 0, 0);

  Add(2, 4U, "600", "700", 1U);

  UpdateVersionStorageInfo();

  std::unique_ptr<Compaction> compaction(level_compaction_picker.PickCompaction(

              cf_name_, mutable_cf_options_, vstorage_.get(), &log_buffer_));

  ASSERT_TRUE(compaction.get() != nullptr);

  ASSERT_EQ(1U, compaction->num_input_levels());

  ASSERT_EQ(2U, compaction->num_input_files(0));

  ASSERT_EQ(2U, compaction->input(0, 0)->fd.GetNumber());

  ASSERT_EQ(3U, compaction->input(0, 1)->fd.GetNumber());

}

TEST_F(CompactionPickerTest, OverlappingUserKeys2) {

  NewVersionStorage(6, kCompactionStyleLevel);

  // Overlapping user keys on same level and output level

  Add(1, 1U, "200", "400", 1000000000U);

  Add(1, 2U, "400", "500", 1U, 0, 0);

  Add(2, 3U, "400", "600", 1U);

  // The following file is not in the compaction despite overlapping user keys

  Add(2, 4U, "600", "700", 1U, 0, 0);

  UpdateVersionStorageInfo();

  std::unique_ptr<Compaction> compaction(level_compaction_picker.PickCompaction(

              cf_name_, mutable_cf_options_, vstorage_.get(), &log_buffer_));

  ASSERT_TRUE(compaction.get() != nullptr);

  ASSERT_EQ(2U, compaction->num_input_levels());

  ASSERT_EQ(2U, compaction->num_input_files(0));

  ASSERT_EQ(1U, compaction->num_input_files(1));

  ASSERT_EQ(1U, compaction->input(0, 0)->fd.GetNumber());

  ASSERT_EQ(2U, compaction->input(0, 1)->fd.GetNumber());

  ASSERT_EQ(3U, compaction->input(1, 0)->fd.GetNumber());

}

TEST_F(CompactionPickerTest, OverlappingUserKeys3) {

  NewVersionStorage(6, kCompactionStyleLevel);

  // Chain of overlapping user key ranges (forces ExpandWhileOverlapping() to

  // expand multiple times)

  Add(1, 1U, "100", "150", 1U);

  Add(1, 2U, "150", "200", 1U, 0, 0);

  Add(1, 3U, "200", "250", 1000000000U, 0, 0);

  Add(1, 4U, "250", "300", 1U, 0, 0);

  Add(1, 5U, "300", "350", 1U, 0, 0);

  // Output level overlaps with the beginning and the end of the chain

  Add(2, 6U, "050", "100", 1U);

  Add(2, 7U, "350", "400", 1U);

  UpdateVersionStorageInfo();

  std::unique_ptr<Compaction> compaction(level_compaction_picker.PickCompaction(

              cf_name_, mutable_cf_options_, vstorage_.get(), &log_buffer_));

  ASSERT_TRUE(compaction.get() != nullptr);

  ASSERT_EQ(2U, compaction->num_input_levels());

  ASSERT_EQ(5U, compaction->num_input_files(0));

  ASSERT_EQ(2U, compaction->num_input_files(1));

  ASSERT_EQ(1U, compaction->input(0, 0)->fd.GetNumber());

  ASSERT_EQ(2U, compaction->input(0, 1)->fd.GetNumber());

  ASSERT_EQ(3U, compaction->input(0, 2)->fd.GetNumber());

  ASSERT_EQ(4U, compaction->input(0, 3)->fd.GetNumber());

  ASSERT_EQ(5U, compaction->input(0, 4)->fd.GetNumber());

  ASSERT_EQ(6U, compaction->input(1, 0)->fd.GetNumber());

  ASSERT_EQ(7U, compaction->input(1, 1)->fd.GetNumber());

}

TEST_F(CompactionPickerTest, NotScheduleL1IfL0WithHigherPri1) {

  NewVersionStorage(6, kCompactionStyleLevel);

  mutable_cf_options_.level0_file_num_compaction_trigger = 2;

  mutable_cf_options_.max_bytes_for_level_base = 900000000U;

  // 6 L0 files, score 3.

  Add(0, 1U, "000", "400", 1U);

  Add(0, 2U, "001", "400", 1U, 0, 0);

  Add(0, 3U, "001", "400", 1000000000U, 0, 0);

  Add(0, 31U, "001", "400", 1000000000U, 0, 0);

  Add(0, 32U, "001", "400", 1000000000U, 0, 0);

  Add(0, 33U, "001", "400", 1000000000U, 0, 0);

  // L1 total size 2GB, score 2.2. If one file being comapcted, score 1.1.

  Add(1, 4U, "050", "300", 1000000000U, 0, 0);

  file_map_[4u].first->being_compacted = true;

  Add(1, 5U, "301", "350", 1000000000U, 0, 0);

  // Output level overlaps with the beginning and the end of the chain

  Add(2, 6U, "050", "100", 1U);

  Add(2, 7U, "300", "400", 1U);

  // No compaction should be scheduled, if L0 has higher priority than L1

  // but L0->L1 compaction is blocked by a file in L1 being compacted.

  UpdateVersionStorageInfo();

  ASSERT_EQ(0, vstorage_->CompactionScoreLevel(0));

  ASSERT_EQ(1, vstorage_->CompactionScoreLevel(1));

  std::unique_ptr<Compaction> compaction(level_compaction_picker.PickCompaction(

      cf_name_, mutable_cf_options_, vstorage_.get(), &log_buffer_));

  ASSERT_TRUE(compaction.get() == nullptr);

}

TEST_F(CompactionPickerTest, NotScheduleL1IfL0WithHigherPri2) {

  NewVersionStorage(6, kCompactionStyleLevel);

  mutable_cf_options_.level0_file_num_compaction_trigger = 2;

  mutable_cf_options_.max_bytes_for_level_base = 900000000U;

  // 6 L0 files, score 3.

  Add(0, 1U, "000", "400", 1U);

  Add(0, 2U, "001", "400", 1U, 0, 0);

  Add(0, 3U, "001", "400", 1000000000U, 0, 0);

  Add(0, 31U, "001", "400", 1000000000U, 0, 0);

  Add(0, 32U, "001", "400", 1000000000U, 0, 0);

  Add(0, 33U, "001", "400", 1000000000U, 0, 0);

  // L1 total size 2GB, score 2.2. If one file being comapcted, score 1.1.

  Add(1, 4U, "050", "300", 1000000000U, 0, 0);

  Add(1, 5U, "301", "350", 1000000000U, 0, 0);

  // Output level overlaps with the beginning and the end of the chain

  Add(2, 6U, "050", "100", 1U);

  Add(2, 7U, "300", "400", 1U);

  // If no file in L1 being compacted, L0->L1 compaction will be scheduled.

  UpdateVersionStorageInfo();  // being_compacted flag is cleared here.

  ASSERT_EQ(0, vstorage_->CompactionScoreLevel(0));

  ASSERT_EQ(1, vstorage_->CompactionScoreLevel(1));

  std::unique_ptr<Compaction> compaction(level_compaction_picker.PickCompaction(

      cf_name_, mutable_cf_options_, vstorage_.get(), &log_buffer_));

  ASSERT_TRUE(compaction.get() != nullptr);

}

TEST_F(CompactionPickerTest, NotScheduleL1IfL0WithHigherPri3) {

  NewVersionStorage(6, kCompactionStyleLevel);

  mutable_cf_options_.level0_file_num_compaction_trigger = 2;

  mutable_cf_options_.max_bytes_for_level_base = 900000000U;

  // 6 L0 files, score 3.

  Add(0, 1U, "000", "400", 1U);

  Add(0, 2U, "001", "400", 1U, 0, 0);

  Add(0, 3U, "001", "400", 1000000000U, 0, 0);

  Add(0, 31U, "001", "400", 1000000000U, 0, 0);

  Add(0, 32U, "001", "400", 1000000000U, 0, 0);

  Add(0, 33U, "001", "400", 1000000000U, 0, 0);

  // L1 score more than 6.

  Add(1, 4U, "050", "300", 1000000000U, 0, 0);

  file_map_[4u].first->being_compacted = true;

  Add(1, 5U, "301", "350", 1000000000U, 0, 0);

  Add(1, 51U, "351", "400", 6000000000U, 0, 0);

  // Output level overlaps with the beginning and the end of the chain

  Add(2, 6U, "050", "100", 1U);

  Add(2, 7U, "300", "400", 1U);

  // If score in L1 is larger than L0, L1 compaction goes through despite

  // there is pending L0 compaction.

  UpdateVersionStorageInfo();

  ASSERT_EQ(1, vstorage_->CompactionScoreLevel(0));

  ASSERT_EQ(0, vstorage_->CompactionScoreLevel(1));

  std::unique_ptr<Compaction> compaction(level_compaction_picker.PickCompaction(

      cf_name_, mutable_cf_options_, vstorage_.get(), &log_buffer_));

  ASSERT_TRUE(compaction.get() != nullptr);

}

TEST_F(CompactionPickerTest, EstimateCompactionBytesNeeded1) {

  int num_levels = ioptions_.num_levels;

  ioptions_.level_compaction_dynamic_level_bytes = false;

  mutable_cf_options_.level0_file_num_compaction_trigger = 3;

  mutable_cf_options_.max_bytes_for_level_base = 1000;

  mutable_cf_options_.max_bytes_for_level_multiplier = 10;

  NewVersionStorage(num_levels, kCompactionStyleLevel);

  Add(0, 1U, "150", "200", 200);

  Add(0, 2U, "150", "200", 200);

  Add(0, 3U, "150", "200", 200);

  // Level 1 is over target by 200

  Add(1, 4U, "400", "500", 600);

  Add(1, 5U, "600", "700", 600);

  // Level 2 is less than target 10000 even added size of level 1

  Add(2, 6U, "150", "200", 2500);

  Add(2, 7U, "201", "210", 2000);

  Add(2, 8U, "300", "310", 2500);

  Add(2, 9U, "400", "500", 2500);

  // Level 3 exceeds target 100,000 of 1000

  Add(3, 10U, "400", "500", 101000);

  // Level 4 exceeds target 1,000,000 of 500 after adding size from level 3

  Add(4, 11U, "400", "500", 999500);

  Add(5, 11U, "400", "500", 8000000);

  UpdateVersionStorageInfo();

  ASSERT_EQ(2200u + 11000u + 5500u,

            vstorage_->estimated_compaction_needed_bytes());

}

TEST_F(CompactionPickerTest, EstimateCompactionBytesNeeded2) {

  int num_levels = ioptions_.num_levels;

  ioptions_.level_compaction_dynamic_level_bytes = false;

  mutable_cf_options_.level0_file_num_compaction_trigger = 3;

  mutable_cf_options_.max_bytes_for_level_base = 1000;

  mutable_cf_options_.max_bytes_for_level_multiplier = 10;

  NewVersionStorage(num_levels, kCompactionStyleLevel);

  Add(0, 1U, "150", "200", 200);

  Add(0, 2U, "150", "200", 200);

  Add(0, 4U, "150", "200", 200);

  Add(0, 5U, "150", "200", 200);

  Add(0, 6U, "150", "200", 200);

  // Level 1 is over target by

  Add(1, 7U, "400", "500", 200);

  Add(1, 8U, "600", "700", 200);

  // Level 2 is less than target 10000 even added size of level 1

  Add(2, 9U, "150", "200", 9500);

  Add(3, 10U, "400", "500", 101000);

  UpdateVersionStorageInfo();

  ASSERT_EQ(1400u + 4400u + 11000u,

            vstorage_->estimated_compaction_needed_bytes());

}

TEST_F(CompactionPickerTest, EstimateCompactionBytesNeededDynamicLevel) {

  int num_levels = ioptions_.num_levels;

  ioptions_.level_compaction_dynamic_level_bytes = true;

  mutable_cf_options_.level0_file_num_compaction_trigger = 3;

  mutable_cf_options_.max_bytes_for_level_base = 1000;

  mutable_cf_options_.max_bytes_for_level_multiplier = 10;

  NewVersionStorage(num_levels, kCompactionStyleLevel);

  // Set Last level size 50000

  // num_levels - 1 target 5000

  // num_levels - 2 is base level with taret 500

  Add(num_levels - 1, 10U, "400", "500", 50000);

  Add(0, 1U, "150", "200", 200);

  Add(0, 2U, "150", "200", 200);

  Add(0, 4U, "150", "200", 200);

  Add(0, 5U, "150", "200", 200);

  Add(0, 6U, "150", "200", 200);

  // num_levels - 3 is over target by 100 + 1000

  Add(num_levels - 3, 7U, "400", "500", 300);

  Add(num_levels - 3, 8U, "600", "700", 300);

  // Level 2 is over target by 1100 + 100

  Add(num_levels - 2, 9U, "150", "200", 5100);

  UpdateVersionStorageInfo();

  ASSERT_EQ(1600u + 12100u + 13200u,

            vstorage_->estimated_compaction_needed_bytes());

}

TEST_F(CompactionPickerTest, IsBottommostLevelTest) {

  // case 1: Higher levels are empty

  NewVersionStorage(6, kCompactionStyleLevel);

  Add(0, 1U, "a", "m");

  Add(0, 2U, "c", "z");

  Add(1, 3U, "d", "e");

  Add(1, 4U, "l", "p");

  Add(2, 5U, "g", "i");

  Add(2, 6U, "x", "z");

  UpdateVersionStorageInfo();

  SetCompactionInputFilesLevels(2, 1);

  AddToCompactionFiles(3U);

  AddToCompactionFiles(5U);

  bool result =

      Compaction::TEST_IsBottommostLevel(2, vstorage_.get(), input_files_);

  ASSERT_TRUE(result);

  // case 2: Higher levels have no overlap

  NewVersionStorage(6, kCompactionStyleLevel);

  Add(0, 1U, "a", "m");

  Add(0, 2U, "c", "z");

  Add(1, 3U, "d", "e");

  Add(1, 4U, "l", "p");

  Add(2, 5U, "g", "i");

  Add(2, 6U, "x", "z");

  Add(3, 7U, "k", "p");

  Add(3, 8U, "t", "w");

  Add(4, 9U, "a", "b");

  Add(5, 10U, "c", "cc");

  UpdateVersionStorageInfo();

  SetCompactionInputFilesLevels(2, 1);

  AddToCompactionFiles(3U);

  AddToCompactionFiles(5U);

  result = Compaction::TEST_IsBottommostLevel(2, vstorage_.get(), input_files_);

  ASSERT_TRUE(result);

  // case 3.1: Higher levels (level 3) have overlap

  NewVersionStorage(6, kCompactionStyleLevel);

  Add(0, 1U, "a", "m");

  Add(0, 2U, "c", "z");

  Add(1, 3U, "d", "e");

  Add(1, 4U, "l", "p");

  Add(2, 5U, "g", "i");

  Add(2, 6U, "x", "z");

  Add(3, 7U, "e", "g");

  Add(3, 8U, "h", "k");

  Add(4, 9U, "a", "b");

  Add(5, 10U, "c", "cc");

  UpdateVersionStorageInfo();

  SetCompactionInputFilesLevels(2, 1);

  AddToCompactionFiles(3U);

  AddToCompactionFiles(5U);

  result = Compaction::TEST_IsBottommostLevel(2, vstorage_.get(), input_files_);

  ASSERT_FALSE(result);

  // case 3.2: Higher levels (level 5) have overlap

  DeleteVersionStorage();

  NewVersionStorage(6, kCompactionStyleLevel);

  Add(0, 1U, "a", "m");

  Add(0, 2U, "c", "z");

  Add(1, 3U, "d", "e");

  Add(1, 4U, "l", "p");

  Add(2, 5U, "g", "i");

  Add(2, 6U, "x", "z");

  Add(3, 7U, "j", "k");

  Add(3, 8U, "l", "m");

  Add(4, 9U, "a", "b");

  Add(5, 10U, "c", "cc");

  Add(5, 11U, "h", "k");

  Add(5, 12U, "y", "yy");

  Add(5, 13U, "z", "zz");

  UpdateVersionStorageInfo();

  SetCompactionInputFilesLevels(2, 1);

  AddToCompactionFiles(3U);

  AddToCompactionFiles(5U);

  result = Compaction::TEST_IsBottommostLevel(2, vstorage_.get(), input_files_);

  ASSERT_FALSE(result);

  // case 3.3: Higher levels (level 5) have overlap, but it's only overlapping

  // one key ("d")

  NewVersionStorage(6, kCompactionStyleLevel);

  Add(0, 1U, "a", "m");

  Add(0, 2U, "c", "z");

  Add(1, 3U, "d", "e");

  Add(1, 4U, "l", "p");

  Add(2, 5U, "g", "i");

  Add(2, 6U, "x", "z");

  Add(3, 7U, "j", "k");

  Add(3, 8U, "l", "m");

  Add(4, 9U, "a", "b");

  Add(5, 10U, "c", "cc");

  Add(5, 11U, "ccc", "d");

  Add(5, 12U, "y", "yy");

  Add(5, 13U, "z", "zz");

  UpdateVersionStorageInfo();

  SetCompactionInputFilesLevels(2, 1);

  AddToCompactionFiles(3U);

  AddToCompactionFiles(5U);

  result = Compaction::TEST_IsBottommostLevel(2, vstorage_.get(), input_files_);

  ASSERT_FALSE(result);

  // Level 0 files overlap

  NewVersionStorage(6, kCompactionStyleLevel);

  Add(0, 1U, "s", "t");

  Add(0, 2U, "a", "m");