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

//  This source code is licensed under both the GPLv2 (found in the

//  COPYING file in the root directory) and Apache 2.0 License

//  (found in the LICENSE.Apache file in the root directory).

//

// Copyright (c) 2011 The LevelDB Authors. All rights reserved.

// Use of this source code is governed by a BSD-style license that can be

// found in the LICENSE file. See the AUTHORS file for names of contributors.

#include "db/compaction/compaction_job.h"

#include <algorithm>

#include <cinttypes>

#include <memory>

#include <optional>

#include <set>

#include <utility>

#include <vector>

#include "db/blob/blob_counting_iterator.h"

#include "db/blob/blob_file_addition.h"

#include "db/blob/blob_file_builder.h"

#include "db/builder.h"

#include "db/compaction/clipping_iterator.h"

#include "db/compaction/compaction_state.h"

#include "db/db_impl/db_impl.h"

#include "db/dbformat.h"

#include "db/error_handler.h"

#include "db/event_helpers.h"

#include "db/history_trimming_iterator.h"

#include "db/log_writer.h"

#include "db/merge_helper.h"

#include "db/range_del_aggregator.h"

#include "db/version_edit.h"

#include "db/version_set.h"

#include "file/filename.h"

#include "file/read_write_util.h"

#include "file/sst_file_manager_impl.h"

#include "file/writable_file_writer.h"

#include "logging/log_buffer.h"

#include "logging/logging.h"

#include "monitoring/iostats_context_imp.h"

#include "monitoring/thread_status_util.h"

#include "options/configurable_helper.h"

#include "options/options_helper.h"

#include "port/port.h"

#include "rocksdb/db.h"

#include "rocksdb/env.h"

#include "rocksdb/options.h"

#include "rocksdb/statistics.h"

#include "rocksdb/status.h"

#include "rocksdb/table.h"

#include "rocksdb/utilities/options_type.h"

#include "table/merging_iterator.h"

#include "table/table_builder.h"

#include "table/unique_id_impl.h"

#include "test_util/sync_point.h"

#include "util/stop_watch.h"

namespace ROCKSDB_NAMESPACE {

const char* GetCompactionReasonString(CompactionReason compaction_reason) {

  switch (compaction_reason) {

    case CompactionReason::kUnknown:

      return "Unknown";

    case CompactionReason::kLevelL0FilesNum:

      return "LevelL0FilesNum";

    case CompactionReason::kLevelMaxLevelSize:

      return "LevelMaxLevelSize";

    case CompactionReason::kUniversalSizeAmplification:

      return "UniversalSizeAmplification";

    case CompactionReason::kUniversalSizeRatio:

      return "UniversalSizeRatio";

    case CompactionReason::kUniversalSortedRunNum:

      return "UniversalSortedRunNum";

    case CompactionReason::kFIFOMaxSize:

      return "FIFOMaxSize";

    case CompactionReason::kFIFOReduceNumFiles:

      return "FIFOReduceNumFiles";

    case CompactionReason::kFIFOTtl:

      return "FIFOTtl";

    case CompactionReason::kManualCompaction:

      return "ManualCompaction";

    case CompactionReason::kFilesMarkedForCompaction:

      return "FilesMarkedForCompaction";

    case CompactionReason::kBottommostFiles:

      return "BottommostFiles";

    case CompactionReason::kTtl:

      return "Ttl";

    case CompactionReason::kFlush:

      return "Flush";

    case CompactionReason::kExternalSstIngestion:

      return "ExternalSstIngestion";

    case CompactionReason::kPeriodicCompaction:

      return "PeriodicCompaction";

    case CompactionReason::kChangeTemperature:

      return "ChangeTemperature";

    case CompactionReason::kForcedBlobGC:

      return "ForcedBlobGC";

    case CompactionReason::kRoundRobinTtl:

      return "RoundRobinTtl";

    case CompactionReason::kRefitLevel:

      return "RefitLevel";

    case CompactionReason::kNumOfReasons:

      // fall through

    default:

      assert(false);

      return "Invalid";

  }

}

const char* GetCompactionProximalOutputRangeTypeString(

    Compaction::ProximalOutputRangeType range_type) {

  switch (range_type) {

    case Compaction::ProximalOutputRangeType::kNotSupported:

      return "NotSupported";

    case Compaction::ProximalOutputRangeType::kFullRange:

      return "FullRange";

    case Compaction::ProximalOutputRangeType::kNonLastRange:

      return "NonLastRange";

    case Compaction::ProximalOutputRangeType::kDisabled:

      return "Disabled";

    default:

      assert(false);

      return "Invalid";

  }

}

CompactionJob::CompactionJob(

    int job_id, Compaction* compaction, const ImmutableDBOptions& db_options,

    const MutableDBOptions& mutable_db_options, const FileOptions& file_options,

    VersionSet* versions, const std::atomic<bool>* shutting_down,

    LogBuffer* log_buffer, FSDirectory* db_directory,

    FSDirectory* output_directory, FSDirectory* blob_output_directory,

    Statistics* stats, InstrumentedMutex* db_mutex,

    ErrorHandler* db_error_handler, JobContext* job_context,

    std::shared_ptr<Cache> table_cache, EventLogger* event_logger,

    bool paranoid_file_checks, bool measure_io_stats, const std::string& dbname,

    CompactionJobStats* compaction_job_stats, Env::Priority thread_pri,

    const std::shared_ptr<IOTracer>& io_tracer,

    const std::atomic<bool>& manual_compaction_canceled,

    const std::string& db_id, const std::string& db_session_id,

    std::string full_history_ts_low, std::string trim_ts,

    BlobFileCompletionCallback* blob_callback, int* bg_compaction_scheduled,

    int* bg_bottom_compaction_scheduled)

    : compact_(new CompactionState(compaction)),

      internal_stats_(compaction->compaction_reason(), 1),

      db_options_(db_options),

      mutable_db_options_copy_(mutable_db_options),

      log_buffer_(log_buffer),

      output_directory_(output_directory),

      stats_(stats),

      bottommost_level_(false),

      write_hint_(Env::WLTH_NOT_SET),

      job_stats_(compaction_job_stats),

      job_id_(job_id),

      dbname_(dbname),

      db_id_(db_id),

      db_session_id_(db_session_id),

      file_options_(file_options),

      env_(db_options.env),

      io_tracer_(io_tracer),

      fs_(db_options.fs, io_tracer),

      file_options_for_read_(

          fs_->OptimizeForCompactionTableRead(file_options, db_options_)),

      versions_(versions),

      shutting_down_(shutting_down),

      manual_compaction_canceled_(manual_compaction_canceled),

      db_directory_(db_directory),

      blob_output_directory_(blob_output_directory),

      db_mutex_(db_mutex),

      db_error_handler_(db_error_handler),

      earliest_snapshot_(job_context->GetEarliestSnapshotSequence()),

      job_context_(job_context),

      table_cache_(std::move(table_cache)),

      event_logger_(event_logger),

      paranoid_file_checks_(paranoid_file_checks),

      measure_io_stats_(measure_io_stats),

      thread_pri_(thread_pri),

      full_history_ts_low_(std::move(full_history_ts_low)),

      trim_ts_(std::move(trim_ts)),

      blob_callback_(blob_callback),

      extra_num_subcompaction_threads_reserved_(0),

      bg_compaction_scheduled_(bg_compaction_scheduled),

      bg_bottom_compaction_scheduled_(bg_bottom_compaction_scheduled) {

  assert(job_stats_ != nullptr);

  assert(log_buffer_ != nullptr);

  assert(job_context->snapshot_context_initialized);

  const auto* cfd = compact_->compaction->column_family_data();

  ThreadStatusUtil::SetEnableTracking(db_options_.enable_thread_tracking);

  ThreadStatusUtil::SetColumnFamily(cfd);

  ThreadStatusUtil::SetThreadOperation(ThreadStatus::OP_COMPACTION);

  ReportStartedCompaction(compaction);

}

CompactionJob::~CompactionJob() {

  assert(compact_ == nullptr);

  ThreadStatusUtil::ResetThreadStatus();

}

void CompactionJob::ReportStartedCompaction(Compaction* compaction) {

  ThreadStatusUtil::SetThreadOperationProperty(ThreadStatus::COMPACTION_JOB_ID,

                                               job_id_);

  ThreadStatusUtil::SetThreadOperationProperty(

      ThreadStatus::COMPACTION_INPUT_OUTPUT_LEVEL,

      (static_cast<uint64_t>(compact_->compaction->start_level()) << 32) +

          compact_->compaction->output_level());

  // In the current design, a CompactionJob is always created

  // for non-trivial compaction.

  assert(compaction->IsTrivialMove() == false ||

         compaction->is_manual_compaction() == true);

  ThreadStatusUtil::SetThreadOperationProperty(

      ThreadStatus::COMPACTION_PROP_FLAGS,

      compaction->is_manual_compaction() +

          (compaction->deletion_compaction() << 1));

  auto total_input_bytes = compaction->CalculateTotalInputSize();

  ThreadStatusUtil::SetThreadOperationProperty(

      ThreadStatus::COMPACTION_TOTAL_INPUT_BYTES, total_input_bytes);

  IOSTATS_RESET(bytes_written);

  IOSTATS_RESET(bytes_read);

  ThreadStatusUtil::SetThreadOperationProperty(

      ThreadStatus::COMPACTION_BYTES_WRITTEN, 0);

  ThreadStatusUtil::SetThreadOperationProperty(

      ThreadStatus::COMPACTION_BYTES_READ, 0);

  // Set the thread operation after operation properties

  // to ensure GetThreadList() can always show them all together.

  ThreadStatusUtil::SetThreadOperation(ThreadStatus::OP_COMPACTION);

  job_stats_->is_manual_compaction = compaction->is_manual_compaction();

  job_stats_->is_full_compaction = compaction->is_full_compaction();

  // populate compaction stats num_input_files and total_num_of_bytes

  size_t num_input_files = 0;

  for (int input_level = 0;

       input_level < static_cast<int>(compaction->num_input_levels());

       ++input_level) {

    const LevelFilesBrief* flevel = compaction->input_levels(input_level);

    num_input_files += flevel->num_files;

  }

  job_stats_->CompactionJobStats::num_input_files = num_input_files;

  job_stats_->total_input_bytes = total_input_bytes;

}

void CompactionJob::Prepare(

    std::optional<std::pair<std::optional<Slice>, std::optional<Slice>>>

        known_single_subcompact) {

  db_mutex_->AssertHeld();

  AutoThreadOperationStageUpdater stage_updater(

      ThreadStatus::STAGE_COMPACTION_PREPARE);

  // Generate file_levels_ for compaction before making Iterator

  auto* c = compact_->compaction;

  [[maybe_unused]] ColumnFamilyData* cfd = c->column_family_data();

  assert(cfd != nullptr);

  const VersionStorageInfo* storage_info = c->input_version()->storage_info();

  assert(storage_info);

  assert(storage_info->NumLevelFiles(compact_->compaction->level()) > 0);

  write_hint_ = storage_info->CalculateSSTWriteHint(

      c->output_level(), db_options_.calculate_sst_write_lifetime_hint_set);

  bottommost_level_ = c->bottommost_level();

  if (!known_single_subcompact.has_value() && c->ShouldFormSubcompactions()) {

    StopWatch sw(db_options_.clock, stats_, SUBCOMPACTION_SETUP_TIME);

    GenSubcompactionBoundaries();

  }

  if (boundaries_.size() >= 1) {

    assert(!known_single_subcompact.has_value());

    for (size_t i = 0; i <= boundaries_.size(); i++) {

      compact_->sub_compact_states.emplace_back(

          c, (i != 0) ? std::optional<Slice>(boundaries_[i - 1]) : std::nullopt,

          (i != boundaries_.size()) ? std::optional<Slice>(boundaries_[i])

                                    : std::nullopt,

          static_cast<uint32_t>(i));

      // assert to validate that boundaries don't have same user keys (without

      // timestamp part).

      assert(i == 0 || i == boundaries_.size() ||

             cfd->user_comparator()->CompareWithoutTimestamp(

                 boundaries_[i - 1], boundaries_[i]) < 0);

    }

    RecordInHistogram(stats_, NUM_SUBCOMPACTIONS_SCHEDULED,

                      compact_->sub_compact_states.size());

  } else {

    std::optional<Slice> start_key;

    std::optional<Slice> end_key;

    if (known_single_subcompact.has_value()) {

      start_key = known_single_subcompact.value().first;

      end_key = known_single_subcompact.value().second;

    } else {

      assert(!start_key.has_value() && !end_key.has_value());

    }

    compact_->sub_compact_states.emplace_back(c, start_key, end_key,

                                              /*sub_job_id*/ 0);

  }

  // collect all seqno->time information from the input files which will be used

  // to encode seqno->time to the output files.

  SequenceNumber preserve_time_min_seqno = kMaxSequenceNumber;

  SequenceNumber preclude_last_level_min_seqno = kMaxSequenceNumber;

  uint64_t preserve_time_duration =

      MinAndMaxPreserveSeconds(c->mutable_cf_options()).max_preserve_seconds;

  if (preserve_time_duration > 0) {

    const ReadOptions read_options(Env::IOActivity::kCompaction);

    // Setup seqno_to_time_mapping_ with relevant time range.

    seqno_to_time_mapping_.SetMaxTimeSpan(preserve_time_duration);

    for (const auto& each_level : *c->inputs()) {

      for (const auto& fmd : each_level.files) {

        std::shared_ptr<const TableProperties> tp;

        Status s = c->input_version()->GetTableProperties(read_options, &tp,

                                                          fmd, nullptr);

        if (s.ok()) {

          s = seqno_to_time_mapping_.DecodeFrom(tp->seqno_to_time_mapping);

        }

        if (!s.ok()) {

          ROCKS_LOG_WARN(

              db_options_.info_log,

              "Problem reading or processing seqno-to-time mapping: %s",

              s.ToString().c_str());

        }

      }

    }

    int64_t _current_time = 0;

    Status s = db_options_.clock->GetCurrentTime(&_current_time);

    if (!s.ok()) {

      ROCKS_LOG_WARN(db_options_.info_log,

                     "Failed to get current time in compaction: Status: %s",

                     s.ToString().c_str());

      // preserve all time information

      preserve_time_min_seqno = 0;

      preclude_last_level_min_seqno = 0;

      seqno_to_time_mapping_.Enforce();

    } else {

      seqno_to_time_mapping_.Enforce(_current_time);

      seqno_to_time_mapping_.GetCurrentTieringCutoffSeqnos(

          static_cast<uint64_t>(_current_time),

          c->mutable_cf_options().preserve_internal_time_seconds,

          c->mutable_cf_options().preclude_last_level_data_seconds,

          &preserve_time_min_seqno, &preclude_last_level_min_seqno);

    }

    // For accuracy of the GetProximalSeqnoBeforeTime queries above, we only

    // limit the capacity after them.

    // Here If we set capacity to the per-SST limit, we could be throwing away

    // fidelity when a compaction output file has a narrower seqno range than

    // all the inputs. If we only limit capacity for each compaction output, we

    // could be doing a lot of unnecessary recomputation in a large compaction

    // (up to quadratic in number of files). Thus, we do soemthing in the

    // middle: enforce a resonably large constant size limit substantially

    // larger than kMaxSeqnoTimePairsPerSST.

    seqno_to_time_mapping_.SetCapacity(kMaxSeqnoToTimeEntries);

  }

#ifndef NDEBUG

  assert(preserve_time_min_seqno <= preclude_last_level_min_seqno);

  TEST_SYNC_POINT_CALLBACK(

      "CompactionJob::PrepareTimes():preclude_last_level_min_seqno",

      static_cast<void*>(&preclude_last_level_min_seqno));

  // Restore the invariant asserted above, in case it was broken under the

  // callback

  preserve_time_min_seqno =

      std::min(preclude_last_level_min_seqno, preserve_time_min_seqno);

#endif

  // Preserve sequence numbers for preserved write times and snapshots, though

  // the specific sequence number of the earliest snapshot can be zeroed.

  preserve_seqno_after_ =

      std::max(preserve_time_min_seqno, SequenceNumber{1}) - 1;

  preserve_seqno_after_ = std::min(preserve_seqno_after_, earliest_snapshot_);

  // If using preclude feature, also preclude snapshots from last level, just

  // because they are heuristically more likely to be accessed than non-snapshot

  // data.

  if (preclude_last_level_min_seqno < kMaxSequenceNumber &&

      earliest_snapshot_ < preclude_last_level_min_seqno) {

    preclude_last_level_min_seqno = earliest_snapshot_;

  }

  // Now combine what we would like to preclude from last level with what we

  // can safely support without dangerously moving data back up the LSM tree,

  // to get the final seqno threshold for proximal vs. last. In particular,

  // when the reserved output key range for the proximal level does not

  // include the entire last level input key range, we need to keep entries

  // already in the last level there. (Even allowing within-range entries to

  // move back up could cause problems with range tombstones. Perhaps it

  // would be better in some rare cases to keep entries in the last level

  // one-by-one rather than based on sequence number, but that would add extra

  // tracking and complexity to CompactionIterator that is probably not

  // worthwhile overall. Correctness is also more clear when splitting by

  // seqno threshold.)

  proximal_after_seqno_ = std::max(preclude_last_level_min_seqno,

                                   c->GetKeepInLastLevelThroughSeqno());

  options_file_number_ = versions_->options_file_number();

}

uint64_t CompactionJob::GetSubcompactionsLimit() {

  return extra_num_subcompaction_threads_reserved_ +

         std::max(

             std::uint64_t(1),

             static_cast<uint64_t>(compact_->compaction->max_subcompactions()));

}

void CompactionJob::AcquireSubcompactionResources(

    int num_extra_required_subcompactions) {

  TEST_SYNC_POINT("CompactionJob::AcquireSubcompactionResources:0");

  TEST_SYNC_POINT("CompactionJob::AcquireSubcompactionResources:1");

  int max_db_compactions =

      DBImpl::GetBGJobLimits(

          mutable_db_options_copy_.max_background_flushes,

          mutable_db_options_copy_.max_background_compactions,

          mutable_db_options_copy_.max_background_jobs,

          versions_->GetColumnFamilySet()

              ->write_controller()

              ->NeedSpeedupCompaction())

          .max_compactions;

  InstrumentedMutexLock l(db_mutex_);

  // Apply min function first since We need to compute the extra subcompaction

  // against compaction limits. And then try to reserve threads for extra

  // subcompactions. The actual number of reserved threads could be less than

  // the desired number.

  int available_bg_compactions_against_db_limit =

      std::max(max_db_compactions - *bg_compaction_scheduled_ -

                   *bg_bottom_compaction_scheduled_,

               0);

  // Reservation only supports backgrdoun threads of which the priority is

  // between BOTTOM and HIGH. Need to degrade the priority to HIGH if the

  // origin thread_pri_ is higher than that. Similar to ReleaseThreads().

  extra_num_subcompaction_threads_reserved_ =

      env_->ReserveThreads(std::min(num_extra_required_subcompactions,

                                    available_bg_compactions_against_db_limit),

                           std::min(thread_pri_, Env::Priority::HIGH));

  // Update bg_compaction_scheduled_ or bg_bottom_compaction_scheduled_

  // depending on if this compaction has the bottommost priority

  if (thread_pri_ == Env::Priority::BOTTOM) {

    *bg_bottom_compaction_scheduled_ +=

        extra_num_subcompaction_threads_reserved_;

  } else {

    *bg_compaction_scheduled_ += extra_num_subcompaction_threads_reserved_;

  }

}

void CompactionJob::ShrinkSubcompactionResources(uint64_t num_extra_resources) {

  // Do nothing when we have zero resources to shrink

  if (num_extra_resources == 0) {

    return;

  }

  db_mutex_->Lock();

  // We cannot release threads more than what we reserved before

  int extra_num_subcompaction_threads_released = env_->ReleaseThreads(

      (int)num_extra_resources, std::min(thread_pri_, Env::Priority::HIGH));

  // Update the number of reserved threads and the number of background

  // scheduled compactions for this compaction job

  extra_num_subcompaction_threads_reserved_ -=

      extra_num_subcompaction_threads_released;

  // TODO (zichen): design a test case with new subcompaction partitioning

  // when the number of actual partitions is less than the number of planned

  // partitions

  assert(extra_num_subcompaction_threads_released == (int)num_extra_resources);

  // Update bg_compaction_scheduled_ or bg_bottom_compaction_scheduled_

  // depending on if this compaction has the bottommost priority

  if (thread_pri_ == Env::Priority::BOTTOM) {

    *bg_bottom_compaction_scheduled_ -=

        extra_num_subcompaction_threads_released;

  } else {

    *bg_compaction_scheduled_ -= extra_num_subcompaction_threads_released;

  }

  db_mutex_->Unlock();

  TEST_SYNC_POINT("CompactionJob::ShrinkSubcompactionResources:0");

}

void CompactionJob::ReleaseSubcompactionResources() {

  if (extra_num_subcompaction_threads_reserved_ == 0) {

    return;

  }

  {

    InstrumentedMutexLock l(db_mutex_);

    // The number of reserved threads becomes larger than 0 only if the

    // compaction prioity is round robin and there is no sufficient

    // sub-compactions available

    // The scheduled compaction must be no less than 1 + extra number

    // subcompactions using acquired resources since this compaction job has not

    // finished yet

    assert(*bg_bottom_compaction_scheduled_ >=

               1 + extra_num_subcompaction_threads_reserved_ ||

           *bg_compaction_scheduled_ >=

               1 + extra_num_subcompaction_threads_reserved_);

  }

  ShrinkSubcompactionResources(extra_num_subcompaction_threads_reserved_);

}

void CompactionJob::GenSubcompactionBoundaries() {

  // The goal is to find some boundary keys so that we can evenly partition

  // the compaction input data into max_subcompactions ranges.

  // For every input file, we ask TableReader to estimate 128 anchor points

  // that evenly partition the input file into 128 ranges and the range

  // sizes. This can be calculated by scanning index blocks of the file.

  // Once we have the anchor points for all the input files, we merge them

  // together and try to find keys dividing ranges evenly.

  // For example, if we have two input files, and each returns following

  // ranges:

  //   File1: (a1, 1000), (b1, 1200), (c1, 1100)

  //   File2: (a2, 1100), (b2, 1000), (c2, 1000)

  // We total sort the keys to following:

  //  (a1, 1000), (a2, 1100), (b1, 1200), (b2, 1000), (c1, 1100), (c2, 1000)

  // We calculate the total size by adding up all ranges' size, which is 6400.

  // If we would like to partition into 2 subcompactions, the target of the

  // range size is 3200. Based on the size, we take "b1" as the partition key

  // since the first three ranges would hit 3200.

  //

  // Note that the ranges are actually overlapping. For example, in the example

  // above, the range ending with "b1" is overlapping with the range ending with

  // "b2". So the size 1000+1100+1200 is an underestimation of data size up to

  // "b1". In extreme cases where we only compact N L0 files, a range can

  // overlap with N-1 other ranges. Since we requested a relatively large number

  // (128) of ranges from each input files, even N range overlapping would

  // cause relatively small inaccuracy.

  ReadOptions read_options(Env::IOActivity::kCompaction);

  read_options.rate_limiter_priority = GetRateLimiterPriority();

  auto* c = compact_->compaction;

  if (c->mutable_cf_options().table_factory->Name() ==

      TableFactory::kPlainTableName()) {

    return;

  }

  if (c->max_subcompactions() <= 1 &&

      !(c->immutable_options().compaction_pri == kRoundRobin &&

        c->immutable_options().compaction_style == kCompactionStyleLevel)) {

    return;

  }

  auto* cfd = c->column_family_data();

  const Comparator* cfd_comparator = cfd->user_comparator();

  const InternalKeyComparator& icomp = cfd->internal_comparator();

  auto* v = compact_->compaction->input_version();

  int base_level = v->storage_info()->base_level();

  InstrumentedMutexUnlock unlock_guard(db_mutex_);

  uint64_t total_size = 0;

  std::vector<TableReader::Anchor> all_anchors;

  int start_lvl = c->start_level();

  int out_lvl = c->output_level();

  for (size_t lvl_idx = 0; lvl_idx < c->num_input_levels(); lvl_idx++) {

    int lvl = c->level(lvl_idx);

    if (lvl >= start_lvl && lvl <= out_lvl) {

      const LevelFilesBrief* flevel = c->input_levels(lvl_idx);

      size_t num_files = flevel->num_files;

      if (num_files == 0) {

        continue;

      }

      for (size_t i = 0; i < num_files; i++) {

        FileMetaData* f = flevel->files[i].file_metadata;

        std::vector<TableReader::Anchor> my_anchors;

        Status s = cfd->table_cache()->ApproximateKeyAnchors(

            read_options, icomp, *f, c->mutable_cf_options(), my_anchors);

        if (!s.ok() || my_anchors.empty()) {

          my_anchors.emplace_back(f->largest.user_key(), f->fd.GetFileSize());

        }

        for (auto& ac : my_anchors) {

          // Can be optimize to avoid this loop.

          total_size += ac.range_size;

        }

        all_anchors.insert(all_anchors.end(), my_anchors.begin(),

                           my_anchors.end());

      }

    }

  }

  // Here we total sort all the anchor points across all files and go through

  // them in the sorted order to find partitioning boundaries.

  // Not the most efficient implementation. A much more efficient algorithm

  // probably exists. But they are more complex. If performance turns out to

  // be a problem, we can optimize.

  std::sort(

      all_anchors.begin(), all_anchors.end(),

      [cfd_comparator](TableReader::Anchor& a, TableReader::Anchor& b) -> bool {

        return cfd_comparator->CompareWithoutTimestamp(a.user_key, b.user_key) <

               0;

      });

  // Remove duplicated entries from boundaries.

  all_anchors.erase(

      std::unique(all_anchors.begin(), all_anchors.end(),

                  [cfd_comparator](TableReader::Anchor& a,

                                   TableReader::Anchor& b) -> bool {

                    return cfd_comparator->CompareWithoutTimestamp(

                               a.user_key, b.user_key) == 0;

                  }),

      all_anchors.end());

  // Get the number of planned subcompactions, may update reserve threads

  // and update extra_num_subcompaction_threads_reserved_ for round-robin

  uint64_t num_planned_subcompactions;

  if (c->immutable_options().compaction_pri == kRoundRobin &&

      c->immutable_options().compaction_style == kCompactionStyleLevel) {

    // For round-robin compaction prioity, we need to employ more

    // subcompactions (may exceed the max_subcompaction limit). The extra

    // subcompactions will be executed using reserved threads and taken into

    // account bg_compaction_scheduled or bg_bottom_compaction_scheduled.

    // Initialized by the number of input files

    num_planned_subcompactions = static_cast<uint64_t>(c->num_input_files(0));

    uint64_t max_subcompactions_limit = GetSubcompactionsLimit();

    if (max_subcompactions_limit < num_planned_subcompactions) {

      // Assert two pointers are not empty so that we can use extra

      // subcompactions against db compaction limits

      assert(bg_bottom_compaction_scheduled_ != nullptr);

      assert(bg_compaction_scheduled_ != nullptr);

      // Reserve resources when max_subcompaction is not sufficient

      AcquireSubcompactionResources(

          (int)(num_planned_subcompactions - max_subcompactions_limit));

      // Subcompactions limit changes after acquiring additional resources.

      // Need to call GetSubcompactionsLimit() again to update the number

      // of planned subcompactions

      num_planned_subcompactions =

          std::min(num_planned_subcompactions, GetSubcompactionsLimit());

    } else {

      num_planned_subcompactions = max_subcompactions_limit;

    }

  } else {

    num_planned_subcompactions = GetSubcompactionsLimit();

  }

  TEST_SYNC_POINT_CALLBACK("CompactionJob::GenSubcompactionBoundaries:0",

                           &num_planned_subcompactions);

  if (num_planned_subcompactions == 1) {

    return;

  }

  // Group the ranges into subcompactions

  uint64_t target_range_size = std::max(

      total_size / num_planned_subcompactions,

      MaxFileSizeForLevel(

          c->mutable_cf_options(), out_lvl,

          c->immutable_options().compaction_style, base_level,

          c->immutable_options().level_compaction_dynamic_level_bytes));

  if (target_range_size >= total_size) {

    return;

  }

  uint64_t next_threshold = target_range_size;

  uint64_t cumulative_size = 0;

  uint64_t num_actual_subcompactions = 1U;

  for (TableReader::Anchor& anchor : all_anchors) {

    cumulative_size += anchor.range_size;

    if (cumulative_size > next_threshold) {

      next_threshold += target_range_size;

      num_actual_subcompactions++;

      boundaries_.push_back(anchor.user_key);

    }

    if (num_actual_subcompactions == num_planned_subcompactions) {

      break;

    }

  }

  TEST_SYNC_POINT_CALLBACK("CompactionJob::GenSubcompactionBoundaries:1",

                           &num_actual_subcompactions);

  // Shrink extra subcompactions resources when extra resrouces are acquired

  ShrinkSubcompactionResources(

      std::min((int)(num_planned_subcompactions - num_actual_subcompactions),

               extra_num_subcompaction_threads_reserved_));

}

Status CompactionJob::Run() {

  AutoThreadOperationStageUpdater stage_updater(

      ThreadStatus::STAGE_COMPACTION_RUN);

  TEST_SYNC_POINT("CompactionJob::Run():Start");

  log_buffer_->FlushBufferToLog();

  LogCompaction();

  const size_t num_threads = compact_->sub_compact_states.size();

  assert(num_threads > 0);

  const uint64_t start_micros = db_options_.clock->NowMicros();

  compact_->compaction->GetOrInitInputTableProperties();

  // Launch a thread for each of subcompactions 1...num_threads-1

  std::vector<port::Thread> thread_pool;

  thread_pool.reserve(num_threads - 1);

  for (size_t i = 1; i < compact_->sub_compact_states.size(); i++) {

    thread_pool.emplace_back(&CompactionJob::ProcessKeyValueCompaction, this,

                             &compact_->sub_compact_states[i]);

  }

  // Always schedule the first subcompaction (whether or not there are also

  // others) in the current thread to be efficient with resources

  ProcessKeyValueCompaction(compact_->sub_compact_states.data());

  // Wait for all other threads (if there are any) to finish execution

  for (auto& thread : thread_pool) {

    thread.join();

  }

  internal_stats_.SetMicros(db_options_.clock->NowMicros() - start_micros);

  for (auto& state : compact_->sub_compact_states) {

    internal_stats_.AddCpuMicros(state.compaction_job_stats.cpu_micros);

    state.RemoveLastEmptyOutput();

  }

  RecordTimeToHistogram(stats_, COMPACTION_TIME,

                        internal_stats_.output_level_stats.micros);

  RecordTimeToHistogram(stats_, COMPACTION_CPU_TIME,

                        internal_stats_.output_level_stats.cpu_micros);

  TEST_SYNC_POINT("CompactionJob::Run:BeforeVerify");

  // Check if any thread encountered an error during execution

  Status status;

  IOStatus io_s;

  bool wrote_new_blob_files = false;

  for (const auto& state : compact_->sub_compact_states) {

    if (!state.status.ok()) {

      status = state.status;

      io_s = state.io_status;

      break;

    }

    if (state.Current().HasBlobFileAdditions()) {

      wrote_new_blob_files = true;

    }

  }

  if (io_status_.ok()) {

    io_status_ = io_s;

  }

  if (status.ok()) {

    constexpr IODebugContext* dbg = nullptr;

    if (output_directory_) {

      io_s = output_directory_->FsyncWithDirOptions(

          IOOptions(), dbg,

          DirFsyncOptions(DirFsyncOptions::FsyncReason::kNewFileSynced));

    }

    if (io_s.ok() && wrote_new_blob_files && blob_output_directory_ &&

        blob_output_directory_ != output_directory_) {

      io_s = blob_output_directory_->FsyncWithDirOptions(

          IOOptions(), dbg,

          DirFsyncOptions(DirFsyncOptions::FsyncReason::kNewFileSynced));

    }

  }

  if (io_status_.ok()) {

    io_status_ = io_s;

  }

  if (status.ok()) {

    status = io_s;

  }

  if (status.ok()) {

    thread_pool.clear();

    std::vector<const CompactionOutputs::Output*> files_output;

    for (const auto& state : compact_->sub_compact_states) {

      for (const auto& output : state.GetOutputs()) {

        files_output.emplace_back(&output);

      }

    }

    ColumnFamilyData* cfd = compact_->compaction->column_family_data();

    std::atomic<size_t> next_file_idx(0);

    auto verify_table = [&](Status& output_status) {

      while (true) {

        size_t file_idx = next_file_idx.fetch_add(1);

        if (file_idx >= files_output.size()) {

          break;

        }

        // Verify that the table is usable

        // We set for_compaction to false and don't

        // OptimizeForCompactionTableRead here because this is a special case

        // after we finish the table building No matter whether

        // use_direct_io_for_flush_and_compaction is true, we will regard this

        // verification as user reads since the goal is to cache it here for

        // further user reads

        ReadOptions verify_table_read_options(Env::IOActivity::kCompaction);

        verify_table_read_options.rate_limiter_priority =

            GetRateLimiterPriority();

        InternalIterator* iter = cfd->table_cache()->NewIterator(

            verify_table_read_options, file_options_,

            cfd->internal_comparator(), files_output[file_idx]->meta,

            /*range_del_agg=*/nullptr,

            compact_->compaction->mutable_cf_options(),

            /*table_reader_ptr=*/nullptr,

            cfd->internal_stats()->GetFileReadHist(

                compact_->compaction->output_level()),

            TableReaderCaller::kCompactionRefill, /*arena=*/nullptr,

            /*skip_filters=*/false, compact_->compaction->output_level(),

            MaxFileSizeForL0MetaPin(compact_->compaction->mutable_cf_options()),

            /*smallest_compaction_key=*/nullptr,

            /*largest_compaction_key=*/nullptr,

            /*allow_unprepared_value=*/false);

        auto s = iter->status();

        if (s.ok() && paranoid_file_checks_) {

          OutputValidator validator(cfd->internal_comparator(),

                                    /*_enable_hash=*/true);

          for (iter->SeekToFirst(); iter->Valid(); iter->Next()) {

            s = validator.Add(iter->key(), iter->value());

            if (!s.ok()) {

              break;

            }

          }

          if (s.ok()) {

            s = iter->status();

          }

          if (s.ok() &&

              !validator.CompareValidator(files_output[file_idx]->validator)) {

            s = Status::Corruption("Paranoid checksums do not match");

          }

        }

        delete iter;

        if (!s.ok()) {

          output_status = s;

          break;

        }

      }

    };

    for (size_t i = 1; i < compact_->sub_compact_states.size(); i++) {

      thread_pool.emplace_back(

          verify_table, std::ref(compact_->sub_compact_states[i].status));

    }

    verify_table(compact_->sub_compact_states[0].status);

    for (auto& thread : thread_pool) {

      thread.join();

    }

    for (const auto& state : compact_->sub_compact_states) {

      if (!state.status.ok()) {

        status = state.status;

        break;

      }

    }

  }

  ReleaseSubcompactionResources();

  TEST_SYNC_POINT("CompactionJob::ReleaseSubcompactionResources");

  for (const auto& state : compact_->sub_compact_states) {

    for (const auto& output : state.GetOutputs()) {

      auto fn =

          TableFileName(state.compaction->immutable_options().cf_paths,

                        output.meta.fd.GetNumber(), output.meta.fd.GetPathId());

      compact_->compaction->SetOutputTableProperties(fn,

                                                     output.table_properties);

    }

  }

  // Before the compaction starts, is_remote_compaction was set to true if

  // compaction_service is set. We now know whether each sub_compaction was

  // done remotely or not. Reset is_remote_compaction back to false and allow

  // AggregateCompactionStats() to set the right value.

  job_stats_->is_remote_compaction = false;

  // Finish up all bookkeeping to unify the subcompaction results.

  compact_->AggregateCompactionStats(internal_stats_, *job_stats_);

  uint64_t num_input_range_del = 0;

  bool ok = BuildStatsFromInputTableProperties(&num_input_range_del);

  // (Sub)compactions returned ok, do sanity check on the number of input

  // keys.

  if (status.ok() && ok) {

    if (job_stats_->has_num_input_records) {

      status = VerifyInputRecordCount(num_input_range_del);

      if (!status.ok()) {

        ROCKS_LOG_WARN(

            db_options_.info_log, "[%s] [JOB %d] Compaction with status: %s",

            compact_->compaction->column_family_data()->GetName().c_str(),

            job_context_->job_id, status.ToString().c_str());

      }

    }

    UpdateCompactionJobInputStats(internal_stats_, num_input_range_del);

  }

  UpdateCompactionJobOutputStats(internal_stats_);

  // Verify number of output records

  // Only verify on table with format collects table properties

  const auto& mutable_cf_options = compact_->compaction->mutable_cf_options();

  if (status.ok() &&

      (mutable_cf_options.table_factory->IsInstanceOf(

           TableFactory::kBlockBasedTableName()) ||

       mutable_cf_options.table_factory->IsInstanceOf(

           TableFactory::kPlainTableName())) &&

      db_options_.compaction_verify_record_count) {

    uint64_t total_output_num = 0;

    for (const auto& state : compact_->sub_compact_states) {

      for (const auto& output : state.GetOutputs()) {

        total_output_num += output.table_properties->num_entries -

                            output.table_properties->num_range_deletions;

      }

    }

    uint64_t expected = internal_stats_.output_level_stats.num_output_records;

    if (internal_stats_.has_proximal_level_output) {

      expected += internal_stats_.proximal_level_stats.num_output_records;

    }

    if (expected != total_output_num) {

      char scratch[2345];

      compact_->compaction->Summary(scratch, sizeof(scratch));

      std::string msg =

          "Number of keys in compaction output SST files does not match "

          "number of keys added. Expected " +

          std::to_string(expected) + " but there are " +

          std::to_string(total_output_num) +

          " in output SST files. Compaction summary: " + scratch;

      ROCKS_LOG_WARN(

          db_options_.info_log, "[%s] [JOB %d] Compaction with status: %s",

          compact_->compaction->column_family_data()->GetName().c_str(),

          job_context_->job_id, msg.c_str());

      status = Status::Corruption(msg);

    }

  }

  RecordCompactionIOStats();

  LogFlush(db_options_.info_log);

  TEST_SYNC_POINT("CompactionJob::Run():End");

  compact_->status = status;

  TEST_SYNC_POINT_CALLBACK("CompactionJob::Run():EndStatusSet", &status);

  return status;

}

Status CompactionJob::Install(bool* compaction_released) {

  assert(compact_);

  AutoThreadOperationStageUpdater stage_updater(

      ThreadStatus::STAGE_COMPACTION_INSTALL);

  db_mutex_->AssertHeld();

  Status status = compact_->status;

  ColumnFamilyData* cfd = compact_->compaction->column_family_data();

  assert(cfd);

  int output_level = compact_->compaction->output_level();

  cfd->internal_stats()->AddCompactionStats(output_level, thread_pri_,

                                            internal_stats_);

  if (status.ok()) {

    status = InstallCompactionResults(compaction_released);

  }

  if (!versions_->io_status().ok()) {

    io_status_ = versions_->io_status();

  }

  VersionStorageInfo::LevelSummaryStorage tmp;

  auto vstorage = cfd->current()->storage_info();

  const auto& stats = internal_stats_.output_level_stats;

  double read_write_amp = 0.0;

  double write_amp = 0.0;

  double bytes_read_per_sec = 0;

  double bytes_written_per_sec = 0;

  const uint64_t bytes_read_non_output_and_blob =

      stats.bytes_read_non_output_levels + stats.bytes_read_blob;

  const uint64_t bytes_read_all =

      stats.bytes_read_output_level + bytes_read_non_output_and_blob;

  const uint64_t bytes_written_all =

      stats.bytes_written + stats.bytes_written_blob;