ceph/src/rocksdb/db/db_impl/db_impl_experimental.cc

   1 //  Copyright (c) 2011-present, Facebook, Inc.  All rights reserved.
   2 //  This source code is licensed under both the GPLv2 (found in the
   3 //  COPYING file in the root directory) and Apache 2.0 License
   4 //  (found in the LICENSE.Apache file in the root directory).
   5 //
   6 // Copyright (c) 2011 The LevelDB Authors. All rights reserved.
   7 // Use of this source code is governed by a BSD-style license that can be
   8 // found in the LICENSE file. See the AUTHORS file for names of contributors.
   9
  10 #include <cinttypes>
  11 #include <vector>
  12
  13 #include "db/column_family.h"
  14 #include "db/db_impl/db_impl.h"
  15 #include "db/job_context.h"
  16 #include "db/version_set.h"
  17 #include "logging/logging.h"
  18 #include "rocksdb/status.h"
  19 #include "util/cast_util.h"
  20
  21 namespace ROCKSDB_NAMESPACE {
  22
  23 #ifndef ROCKSDB_LITE
  24 Status DBImpl::SuggestCompactRange(ColumnFamilyHandle* column_family,
  25                                    const Slice* begin, const Slice* end) {
  26   auto cfh = static_cast_with_check<ColumnFamilyHandleImpl>(column_family);
  27   auto cfd = cfh->cfd();
  28   InternalKey start_key, end_key;
  29   if (begin != nullptr) {
  30     start_key.SetMinPossibleForUserKey(*begin);
  31   }
  32   if (end != nullptr) {
  33     end_key.SetMaxPossibleForUserKey(*end);
  34   }
  35   {
  36     InstrumentedMutexLock l(&mutex_);
  37     auto vstorage = cfd->current()->storage_info();
  38     for (int level = 0; level < vstorage->num_non_empty_levels() - 1; ++level) {
  39       std::vector<FileMetaData*> inputs;
  40       vstorage->GetOverlappingInputs(
  41           level, begin == nullptr ? nullptr : &start_key,
  42           end == nullptr ? nullptr : &end_key, &inputs);
  43       for (auto f : inputs) {
  44         f->marked_for_compaction = true;
  45       }
  46     }
  47     // Since we have some more files to compact, we should also recompute
  48     // compaction score
  49     vstorage->ComputeCompactionScore(*cfd->ioptions(),
  50                                      *cfd->GetLatestMutableCFOptions());
  51     SchedulePendingCompaction(cfd);
  52     MaybeScheduleFlushOrCompaction();
  53   }
  54   return Status::OK();
  55 }
  56
  57 Status DBImpl::PromoteL0(ColumnFamilyHandle* column_family, int target_level) {
  58   assert(column_family);
  59
  60   if (target_level < 1) {
  61     ROCKS_LOG_INFO(immutable_db_options_.info_log,
  62                    "PromoteL0 FAILED. Invalid target level %d\n", target_level);
  63     return Status::InvalidArgument("Invalid target level");
  64   }
  65
  66   Status status;
  67   VersionEdit edit;
  68   JobContext job_context(next_job_id_.fetch_add(1), true);
  69   {
  70     InstrumentedMutexLock l(&mutex_);
  71     auto* cfd = static_cast<ColumnFamilyHandleImpl*>(column_family)->cfd();
  72     const auto* vstorage = cfd->current()->storage_info();
  73
  74     if (target_level >= vstorage->num_levels()) {
  75       ROCKS_LOG_INFO(immutable_db_options_.info_log,
  76                      "PromoteL0 FAILED. Target level %d does not exist\n",
  77                      target_level);
  78       job_context.Clean();
  79       status = Status::InvalidArgument("Target level does not exist");
  80       return status;
  81     }
  82
  83     // Sort L0 files by range.
  84     const InternalKeyComparator* icmp = &cfd->internal_comparator();
  85     auto l0_files = vstorage->LevelFiles(0);
  86     std::sort(l0_files.begin(), l0_files.end(),
  87               [icmp](FileMetaData* f1, FileMetaData* f2) {
  88                 return icmp->Compare(f1->largest, f2->largest) < 0;
  89               });
  90
  91     // Check that no L0 file is being compacted and that they have
  92     // non-overlapping ranges.
  93     for (size_t i = 0; i < l0_files.size(); ++i) {
  94       auto f = l0_files[i];
  95       if (f->being_compacted) {
  96         ROCKS_LOG_INFO(immutable_db_options_.info_log,
  97                        "PromoteL0 FAILED. File %" PRIu64 " being compacted\n",
  98                        f->fd.GetNumber());
  99         job_context.Clean();
 100         status =
 101             Status::InvalidArgument("PromoteL0 called during L0 compaction");
 102         return status;
 103       }
 104
 105       if (i == 0) continue;
 106       auto prev_f = l0_files[i - 1];
 107       if (icmp->Compare(prev_f->largest, f->smallest) >= 0) {
 108         ROCKS_LOG_INFO(immutable_db_options_.info_log,
 109                        "PromoteL0 FAILED. Files %" PRIu64 " and %" PRIu64
 110                        " have overlapping ranges\n",
 111                        prev_f->fd.GetNumber(), f->fd.GetNumber());
 112         job_context.Clean();
 113         status = Status::InvalidArgument("L0 has overlapping files");
 114         return status;
 115       }
 116     }
 117
 118     // Check that all levels up to target_level are empty.
 119     for (int level = 1; level <= target_level; ++level) {
 120       if (vstorage->NumLevelFiles(level) > 0) {
 121         ROCKS_LOG_INFO(immutable_db_options_.info_log,
 122                        "PromoteL0 FAILED. Level %d not empty\n", level);
 123         job_context.Clean();
 124         status = Status::InvalidArgument(
 125             "All levels up to target_level "
 126             "must be empty");
 127         return status;
 128       }
 129     }
 130
 131     edit.SetColumnFamily(cfd->GetID());
 132     for (const auto& f : l0_files) {
 133       edit.DeleteFile(0, f->fd.GetNumber());
 134       edit.AddFile(target_level, f->fd.GetNumber(), f->fd.GetPathId(),
 135                    f->fd.GetFileSize(), f->smallest, f->largest,
 136                    f->fd.smallest_seqno, f->fd.largest_seqno,
 137                    f->marked_for_compaction, f->temperature,
 138                    f->oldest_blob_file_number, f->oldest_ancester_time,
 139                    f->file_creation_time, f->file_checksum,
 140                    f->file_checksum_func_name, f->unique_id);
 141     }
 142
 143     status = versions_->LogAndApply(cfd, *cfd->GetLatestMutableCFOptions(),
 144                                     &edit, &mutex_, directories_.GetDbDir());
 145     if (status.ok()) {
 146       InstallSuperVersionAndScheduleWork(cfd,
 147                                          &job_context.superversion_contexts[0],
 148                                          *cfd->GetLatestMutableCFOptions());
 149     }
 150   }  // lock released here
 151   LogFlush(immutable_db_options_.info_log);
 152   job_context.Clean();
 153
 154   return status;
 155 }
 156 #endif  // ROCKSDB_LITE
 157
 158 }  // namespace ROCKSDB_NAMESPACE