//  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 <condition_variable>

#include <exception>

#include <memory>

#include <mutex>

#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/blob/missing_blob_repair_iterator.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/file_util.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/format.h"

#include "table/merging_iterator.h"

#include "table/meta_blocks.h"

#include "table/table_builder.h"

#include "table/unique_id_impl.h"

#include "test_util/sync_point.h"

#include "util/defer.h"

#include "util/hash_containers.h"

#include "util/stop_watch.h"

namespace ROCKSDB_NAMESPACE {

namespace {

class SubcompactionCompletion {

 public:

  SubcompactionCompletion() = default;

  SubcompactionCompletion(const SubcompactionCompletion&) = delete;

  SubcompactionCompletion& operator=(const SubcompactionCompletion&) = delete;

  SubcompactionCompletion(SubcompactionCompletion&&) = delete;

  SubcompactionCompletion& operator=(SubcompactionCompletion&&) = delete;

  class Guard {

   public:

    explicit Guard(SubcompactionCompletion& completion)

        : completion_(&completion) {}

    ~Guard() {

      if (completion_ != nullptr) {

        completion_->Complete();

      }

    }

    Guard(const Guard&) = delete;

    Guard& operator=(const Guard&) = delete;

    Guard(Guard&&) = delete;

    Guard& operator=(Guard&&) = delete;

    void Release() { completion_ = nullptr; }

   private:

    SubcompactionCompletion* completion_;

  };

  void Add() {

    std::lock_guard<std::mutex> lock(mutex_);

    ++remaining_;

  }

  ~SubcompactionCompletion() {

    std::unique_lock<std::mutex> lock(mutex_);

    TEST_SYNC_POINT("CompactionJob::RunSubcompactions:Wait");

    cv_.wait(lock, [&] { return remaining_ == 0; });

  }

 private:

  void Complete() {

    std::lock_guard<std::mutex> lock(mutex_);

    assert(remaining_ > 0);

    --remaining_;

    cv_.notify_one();

  }

  std::mutex mutex_;

  std::condition_variable cv_;

  size_t remaining_ = 0;

};

}  // 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::kReadTriggered:

      return "ReadTriggered";

    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";

  }

}

// Static constant for compaction abort flag - always false, used for

// compaction service jobs that don't support abort signaling

const std::atomic<int> CompactionJob::kCompactionAbortedFalse{0};

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::atomic<int>& compaction_aborted, 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,

    std::atomic<int>* num_running_remote_compactions)

    : 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_)),

      file_options_for_compaction_input_read_(file_options_for_read_),

      versions_(versions),

      shutting_down_(shutting_down),

      manual_compaction_canceled_(manual_compaction_canceled),

      compaction_aborted_(compaction_aborted),

      db_directory_(db_directory),

      blob_output_directory_(blob_output_directory),

      db_mutex_(db_mutex),

      db_error_handler_(db_error_handler),

      // job_context cannot be nullptr, but we will assert later in the body of

      // the constructor.

      earliest_snapshot_(job_context

                             ? job_context->GetEarliestSnapshotSequence()

                             : kMaxSequenceNumber),

      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),

      num_running_remote_compactions_(num_running_remote_compactions) {

  assert(job_stats_ != nullptr);

  assert(log_buffer_ != nullptr);

  assert(job_context);

  assert(job_context->snapshot_context_initialized);

  if (db_options_.use_direct_io_for_compaction_reads &&

      !db_options_.use_direct_reads) {

    file_options_for_compaction_input_read_.use_direct_reads = true;

  }

  ReportStartedCompaction(compaction);

  InitializeThreadStatus();

}

CompactionJob::~CompactionJob() {

  assert(compact_ == nullptr);

  ThreadStatusUtil::ResetThreadStatus();

}

void CompactionJob::InitializeThreadStatus() {

  auto* compaction = compact_->compaction;

  ThreadStatusUtil::SetEnableTracking(db_options_.enable_thread_tracking);

  ThreadStatusUtil::SetColumnFamily(compaction->column_family_data());

  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));

  ThreadStatusUtil::SetThreadOperationProperty(

      ThreadStatus::COMPACTION_TOTAL_INPUT_BYTES,

      compaction->CalculateTotalInputSize());

  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);

}

void CompactionJob::ReportStartedCompaction(Compaction* 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 = compaction->CalculateTotalInputSize();

}

void CompactionJob::Prepare(

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

        known_single_subcompact,

    const CompactionProgress& compaction_progress,

    log::Writer* compaction_progress_writer) {

  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);

  }

  MaybeAssignCompactionProgressAndWriter(compaction_progress,

                                         compaction_progress_writer);

  // 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

  // For a bottommost-file compaction, which exists only because the marker

  // (VersionStorageInfo::ComputeBottommostFilesMarkedForCompaction) decided the

  // file's largest sequence number could be zeroed, honor that same

  // version-carried preserve boundary here. The marker reads DBImpl's live,

  // DB-wide seqno->time mapping, whereas the value computed above is derived

  // only from the selected file's persisted mapping, which can be sparse or

  // empty (e.g. a file written before preserve/preclude was enabled). Taking

  // the max ensures this compaction can zero at least the sequence numbers the

  // marker considered zeroable, so a marked file makes progress instead of

  // being re-marked forever (infinite compaction loop). Other compaction

  // reasons keep the per-input-file boundary so their seqno-zeroing behavior is

  // unchanged. When preserve/preclude is inactive the version value is

  // kMaxSequenceNumber, making this a no-op. A remote compaction worker does

  // not know the reason or the live mapping; it gets this boundary from the

  // primary in CompactionServiceInput.

  if (c->compaction_reason() == CompactionReason::kBottommostFiles) {

    preserve_time_min_seqno_floor_ = storage_info->GetPreserveTimeMinSeqno();

  }

  preserve_time_min_seqno =

      std::max(preserve_time_min_seqno, preserve_time_min_seqno_floor_);

  // 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();

  // For remote compaction hardening: capture the current MANIFEST position as a

  // floor for the worker's recovery. manifest_file_size() is a physical byte

  // offset that VersionSet::ProcessManifestWrites records under the DB mutex

  // only after the whole version-edit batch is written and sync'd, so it is a

  // durable, complete-batch offset -- never mid-record or mid-atomic-group --

  // at or past the compaction's (already committed) input version.

  if (mutable_db_options_copy_.remote_compaction_manifest_floor) {

    min_manifest_file_number_ = versions_->manifest_file_number();

    min_manifest_file_size_ = versions_->manifest_file_size();

  }

}

void CompactionJob::MaybeAssignCompactionProgressAndWriter(

    const CompactionProgress& compaction_progress,

    log::Writer* compaction_progress_writer) {

  // LIMITATION: Only supports resuming single subcompaction for now

  if (compact_->sub_compact_states.size() != 1) {

    return;

  }

  if (!compaction_progress.empty()) {

    assert(compaction_progress.size() == 1);

    SubcompactionState* sub_compact = &compact_->sub_compact_states[0];

    const SubcompactionProgress& subcompaction_progress =

        compaction_progress[0];

    sub_compact->SetSubcompactionProgress(subcompaction_progress);

  }

  compaction_progress_writer_ = compaction_progress_writer;

}

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) {

  if (remote_compaction_executor_) {

    return;

  }

  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_));

}

void CompactionJob::InitializeCompactionRun() {

  AutoThreadOperationStageUpdater stage_updater(

      ThreadStatus::STAGE_COMPACTION_RUN);

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

  log_buffer_->FlushBufferToLog();

  LogCompaction();

}

void CompactionJob::RunSubcompactions() {

  TEST_SYNC_POINT("CompactionJob::RunSubcompactions:BeforeStart");

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

  assert(num_threads > 0);

  compact_->compaction->GetOrInitInputTableProperties();

  if (remote_compaction_executor_) {

    const auto run_subcompaction = [this](SubcompactionState* sub_compact) {

      try {

        ProcessKeyValueCompaction(sub_compact);

      } catch (...) {

        sub_compact->status = Status::Aborted();

      }

    };

    SubcompactionCompletion completion;

    for (size_t i = 1; i < num_threads; ++i) {

      try {

        completion.Add();

        SubcompactionCompletion::Guard submission(completion);

        remote_compaction_executor_([&, i] {

          SubcompactionCompletion::Guard completed(completion);

          run_subcompaction(&compact_->sub_compact_states[i]);

        });

        submission.Release();

      } catch (...) {

        compact_->sub_compact_states[0].status = Status::Aborted();

        for (size_t j = i; j < num_threads; ++j) {

          compact_->sub_compact_states[j].status = Status::Aborted();

        }

        break;

      }

    }

    if (compact_->sub_compact_states[0].status.ok()) {

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

    }

  } else {

    // 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();

    }

  }

  RecordRemoteCompactionInputBlobFileCount();

  RemoveEmptyOutputs();

  ReleaseSubcompactionResources();

  TEST_SYNC_POINT("CompactionJob::ReleaseSubcompactionResources");

}

void CompactionJob::RecordRemoteCompactionInputBlobFileCount() {

  const uint64_t oldest_blob_file_number =

      remote_compaction_min_blob_file_number_.load(std::memory_order_relaxed);

  if (oldest_blob_file_number == std::numeric_limits<uint64_t>::max()) {

    return;

  }

  uint64_t blob_file_count = 0;

  const VersionStorageInfo* const storage_info =

      compact_->compaction->input_version()->storage_info();

  for (const auto& blob_file : storage_info->GetBlobFiles()) {

    if (blob_file->GetBlobFileNumber() >= oldest_blob_file_number) {

      ++blob_file_count;

    }

  }

  RecordTick(stats_, REMOTE_COMPACT_INPUT_BLOB_FILE_COUNT, blob_file_count);

}

void CompactionJob::UpdateTimingStats(uint64_t start_micros) {

  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);

  }

  RecordTimeToHistogram(stats_, COMPACTION_TIME,

                        internal_stats_.output_level_stats.micros);

  RecordTimeToHistogram(stats_, COMPACTION_CPU_TIME,

                        internal_stats_.output_level_stats.cpu_micros);

}

void CompactionJob::RemoveEmptyOutputs() {

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

    state.RemoveLastEmptyOutput();

  }

}

void CompactionJob::CleanupAbortedSubcompactions() {

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

  uint64_t total_sst_files_deleted = 0;

  uint64_t total_blob_files_deleted = 0;

  // Track the first file deletion error to report at the end

  Status first_error;

  int deletion_errors = 0;

  // Mark all subcompactions as aborted and delete their output files

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

    // Mark this subcompaction as aborted

    sub_compact.status =

        Status::Incomplete(Status::SubCode::kCompactionAborted);

    // Delete all files (SST and blob) tracked during compaction.

    // GetOutputFilePaths() contains ALL file paths created, including

    // in-progress files that may have been removed from outputs_ or

    // blob_file_additions_.

    for (const bool is_proximal_level : {false, true}) {

      if (is_proximal_level &&

          !compact_->compaction->SupportsPerKeyPlacement()) {

        continue;

      }

      for (const std::string& file_path :

           sub_compact.Outputs(is_proximal_level)->GetOutputFilePaths()) {

        Status s = env_->DeleteFile(file_path);

        if (s.ok()) {

          // Count SST vs blob files by checking extension

          if (file_path.find(".sst") != std::string::npos) {

            total_sst_files_deleted++;

          } else if (file_path.find(".blob") != std::string::npos) {

            total_blob_files_deleted++;

          }

        } else if (!s.IsNotFound()) {